yantrikdb 0.23.0

//! Decoupled write path RFC, Phase 4.1 — per-DB vector index with
//! immutable cold tier + mutable delta tier.
//!
//! The classic wedge primitive (`engine/record.rs:68 vec_index.write()`
//! held across HNSW insert) is eliminated by this two-tier design:
//!
//! - **Cold** (`ArcSwap<HnswIndex>`): immutable per epoch. Reads do
//!   `cold.load()` which is an Arc clone — lock-free, no contention with
//!   writers, no parking_lot writer-priority effects.
//!
//! - **Delta** (`RwLock<Vec<DeltaEntry>>`): bounded mutable buffer of
//!   recent writes that haven't been compacted into cold yet. Reads do
//!   exact-distance scan over the delta (small N, SIMD-friendly). Writes
//!   acquire the delta write lock briefly to append. Reader latency adds
//!   `O(delta.len() * dim)` distance computations — bounded by `delta_max`.
//!
//! - **Compaction** (Phase 5): periodically clones cold + applies delta
//!   entries to build a new cold, then atomically swaps via ArcSwap. The
//!   delta is sealed during compaction and a new mutable delta is allocated
//!   so writers don't block.
//!
//! ## Lock-acquisition contract
//!
//! - Reads: `cold.load()` (no lock) + `delta.read()` (parking_lot RwLock
//!   read; only contends with the brief append lock and the
//!   sealing-for-compaction swap).
//! - Writes: `delta.write()` for one append + drop. Never touches cold.
//! - Compaction (Phase 5): clones cold off-thread, swaps cold via ArcSwap,
//!   swaps delta atomically. Brief writer lock, no impact on readers.
//!
//! ## Visibility / read-your-writes
//!
//! Writes are visible to subsequent reads as soon as the delta append
//! returns — `delta.read()` will include them. Strict RYW (caller wants
//! to wait for cold to absorb the write) is a Phase 6 concern via
//! sequence numbers; it is not the default visibility contract.
//!
//! ## Tombstone semantics
//!
//! `tombstone(rid)` marks the entry in the delta as deleted. The
//! `search()` path filters tombstoned entries out of the merged result.
//! For rids that exist only in cold, the tombstone is appended to delta
//! as a deletion marker so readers see the rid as deleted. Compaction
//! resolves these by removing the rid from cold during rebuild.

use std::sync::Arc;

use arc_swap::ArcSwap;
use parking_lot::RwLock;

use crate::error::{Result, YantrikDbError};
use crate::vector::hnsw::HnswIndex;

/// One entry in the mutable delta tier.
///
/// `seq` is the monotonic sequence number assigned at append time. This is
/// the basis for read-your-writes via Phase 6's `recall_with_seq`.
///
/// `tombstoned` distinguishes a live insert from a deletion marker. A delta
/// entry with `tombstoned=true` MUST suppress the rid from search results
/// even if cold has a non-tombstoned copy of the same rid.
#[derive(Clone, Debug)]
pub struct DeltaEntry {
    pub rid: String,
    pub embedding: Vec<f32>,
    /// Precomputed Euclidean norm of `embedding` (v0.9.3 speed work) —
    /// computed once at append instead of on every search's linear scan.
    /// 0.0 for tombstone markers (whose embedding is never read).
    pub norm: f64,
    pub seq: u64,
    pub tombstoned: bool,
    /// **v0.10 Item 3 — reservation.** `false` means this entry is
    /// RESERVED but not yet visible: a correction has appended its new
    /// vector to hold the slot, but its SQL transaction has not committed.
    /// An unpublished entry is INVISIBLE to search (so during the commit
    /// window the record is still retrieved under its old, still-durable
    /// vector — no new-vector/old-text mismatch) and is SKIPPED by
    /// compaction (so it cannot be sealed into cold before the commit
    /// decision). On commit it is published (made visible); on commit
    /// failure it is removed. `true` for every normal append.
    pub published: bool,
}

/// Default soft cap on delta size. Hitting this triggers backpressure on
/// `append()`. Phase 5 compaction will keep the delta below this in steady
/// state by sealing + swapping when it crosses the cap. Tunable per
/// deployment via `DeltaIndex::with_capacity` or the `YANTRIKDB_DELTA_MAX`
/// environment variable.
///
/// **Default 256 (was 1024 in v0.6.6).** Cross-platform empirical study by
/// yantrikdb-server (2026-05-07): at `delta_max=256` vs the previous 1024,
/// write throughput rose from ~586/s to ~996/s (+70% over v0.6.6, ≈2.2× over
/// v0.6.5) while read p50 dropped from ~300ms to ~141ms — better than even
/// v0.6.5's pre-wedge baseline. The intuition is that a smaller delta keeps
/// per-search linear scans cheap and triggers compaction often enough that the
/// cold HNSW absorbs most of the working set, so most reads never touch the
/// hot path's RwLock at all. Larger caps amortize compaction overhead but pay
/// for it in steady-state read latency.
pub const DEFAULT_DELTA_MAX: usize = 256;

/// Default max dirty age before age-based compaction fires.
///
/// **The aging gap.** The size-based trigger (`delta_len >= delta_max/2`)
/// only fires when a namespace is actively writing. A namespace whose
/// delta sits at e.g. 50 dirty entries with no new writes will *never*
/// hit the size threshold, so reads against it pay the linear delta scan
/// indefinitely. ChatGPT's 2026-05-08 review (epic 5 task 13) flagged
/// aging as the single biggest concrete gap in our scheduler-by-
/// construction design.
///
/// Fix: the compactor also fires when the *oldest* entry in a non-empty
/// delta has been sitting for more than this duration. 60s is a starting
/// default — long enough that bursty workloads still rely on the size
/// trigger (avoiding compaction churn), short enough that idle namespaces
/// merge into cold within a window where read latency hasn't deteriorated
/// noticeably. Tunable per deployment via `DeltaIndex::with_capacity_and_age`
/// or the `YANTRIKDB_MAX_DIRTY_AGE_SECS` environment variable.
pub const DEFAULT_MAX_DIRTY_AGE: std::time::Duration = std::time::Duration::from_secs(60);

/// Per-DB two-tier vector index used by the decoupled write path.
///
/// Wraps an `HnswIndex` cold tier (atomically swapped via ArcSwap at
/// compaction) and a bounded mutable delta. Replaces the foreground
/// `vec_index.write()` lock pattern that produced the wedge.
pub struct DeltaIndex {
    cold: ArcSwap<HnswIndex>,
    delta: RwLock<Vec<DeltaEntry>>,
    delta_max: usize,
    dim: usize,
    /// Wall-clock instant when the delta most recently transitioned from
    /// empty to non-empty. Cleared (set to `None`) at every successful
    /// `seal_delta_for_compaction`. The compactor reads this to decide
    /// whether the age-based trigger fires.
    ///
    /// Lives on `DeltaIndex` rather than per-`DeltaEntry` so a busy
    /// workload with frequent appends doesn't pay one Instant per entry —
    /// we only care about the *oldest* unflushed entry's age.
    oldest_dirty_at: parking_lot::Mutex<Option<crate::time::Instant>>,
    /// Compaction trigger threshold for `oldest_dirty_at` — once the
    /// delta's oldest entry has been sitting longer than this, the
    /// compactor fires regardless of delta size.
    max_dirty_age: std::time::Duration,
    /// **Saga task 18 Option 4 (v0.7.2).** Event-driven compactor
    /// wake. `append()` and `tombstone()` signal this condvar when the
    /// delta crosses ~80% of `delta_max`, so the compactor wakes within
    /// microseconds of pressure rather than waiting up to its 250ms
    /// poll tick. The condvar's paired sentinel mutex is `()` — no
    /// data lives behind it; it's there because parking_lot::Condvar
    /// requires a guard. The 250ms tick stays as a backstop for the
    /// age-trigger path and graceful shutdown responsiveness. Confirmed
    /// architectural pull by yantrikdb-server msg b9c98a4d 2026-05-08
    /// (90s bench showed read p99 spikes during compactor sleep
    /// windows; this closes that gap).
    compactor_wake_cv: parking_lot::Condvar,
    compactor_wake_mu: parking_lot::Mutex<()>,
    /// Number of delta entries currently inside a compaction build window
    /// — snapshotted by `snapshot_published_for_compaction`, physically
    /// removed by `retire_compacted`. Those entries stay in the delta for
    /// read visibility, but they are logically LEAVING, so `append_inner`
    /// subtracts this from the occupancy it holds against `delta_max`.
    /// Without the subtraction the snapshot design starves writers: the
    /// delta sits at capacity for the whole build and every write 503s —
    /// the exact wedge `spawn_all_workers_bundles_materializer_and_compactor`
    /// pins (this regressed when seal-and-drain became snapshot-and-retire,
    /// caught by that test before commit). Worst-case transient occupancy
    /// is `2 * delta_max`: one build in flight plus a fully refilled delta.
    /// Stores happen while holding the `delta` lock (read in snapshot,
    /// write in retire), and the admission load happens under the write
    /// lock, so the lock provides the ordering; the atomic is only there
    /// so `append_inner` needs no extra lock on its hot path.
    compacting: std::sync::atomic::AtomicUsize,
}

/// Outcome of [`DeltaIndex::append_reserved`]. `#[must_use]` is the
/// load-bearing part: it makes `append_reserved(...)?;` — the statement
/// shape that silently discarded the old bool return (sol 4a.6d-3 r1
/// finding 1: two correction sites did exactly that) — a compiler warning,
/// so every caller must decide which arm it is on.
#[must_use = "AlreadyPresent means the caller owns NO reservation — \
              publishing or removing the existing entry corrupts a prior \
              write's vector"]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReservedAppend {
    /// A new unpublished entry was inserted; the caller owes
    /// publish-on-commit / remove-on-failure (the ReservationGuard rule).
    Inserted,
    /// An identical (rid, seq) already exists; the caller owes NOTHING.
    AlreadyPresent,
}

impl DeltaIndex {
    /// Create a new empty `DeltaIndex` with the given embedding dimension.
    /// Cold starts as a fresh empty `HnswIndex(dim)`. Delta is empty.
    pub fn new(dim: usize) -> Self {
        Self::with_capacity(dim, DEFAULT_DELTA_MAX)
    }

    /// Create a new `DeltaIndex` with a custom delta capacity.
    pub fn with_capacity(dim: usize, delta_max: usize) -> Self {
        Self::with_capacity_and_age(dim, delta_max, DEFAULT_MAX_DIRTY_AGE)
    }

    /// Create a new `DeltaIndex` with a custom delta capacity AND a custom
    /// age-based compaction trigger. Used by tests that need to exercise
    /// the age trigger in seconds rather than the production-default 60s,
    /// and by the engine constructor when reading the
    /// `YANTRIKDB_MAX_DIRTY_AGE_SECS` env override.
    pub fn with_capacity_and_age(
        dim: usize,
        delta_max: usize,
        max_dirty_age: std::time::Duration,
    ) -> Self {
        Self {
            cold: ArcSwap::new(Arc::new(HnswIndex::new(dim))),
            delta: RwLock::new(Vec::with_capacity(delta_max.min(4096))),
            delta_max,
            dim,
            oldest_dirty_at: parking_lot::Mutex::new(None),
            max_dirty_age,
            compactor_wake_cv: parking_lot::Condvar::new(),
            compactor_wake_mu: parking_lot::Mutex::new(()),
            compacting: std::sync::atomic::AtomicUsize::new(0),
        }
    }

    /// Construct a `DeltaIndex` whose cold tier is a pre-built `HnswIndex`.
    /// Used during engine open() when the index is rebuilt from the
    /// SQLite source of truth on disk.
    pub fn from_cold(cold: HnswIndex, delta_max: usize) -> Self {
        Self::from_cold_with_age(cold, delta_max, DEFAULT_MAX_DIRTY_AGE)
    }

    /// `from_cold` sibling that also takes a custom max_dirty_age — used
    /// by the engine constructor when honoring YANTRIKDB_MAX_DIRTY_AGE_SECS.
    pub fn from_cold_with_age(
        cold: HnswIndex,
        delta_max: usize,
        max_dirty_age: std::time::Duration,
    ) -> Self {
        let dim = cold.dim();
        Self {
            cold: ArcSwap::new(Arc::new(cold)),
            delta: RwLock::new(Vec::with_capacity(delta_max.min(4096))),
            delta_max,
            dim,
            oldest_dirty_at: parking_lot::Mutex::new(None),
            max_dirty_age,
            compactor_wake_cv: parking_lot::Condvar::new(),
            compactor_wake_mu: parking_lot::Mutex::new(()),
            compacting: std::sync::atomic::AtomicUsize::new(0),
        }
    }

    /// Embedding dimension for this index.
    pub fn dim(&self) -> usize {
        self.dim
    }

    /// Soft cap on delta size before backpressure fires. Exposed so
    /// callers (notably yantrikdb-server's tick loop, RFC scheduler
    /// pressure rule) can scale enrichment thresholds proportionally
    /// rather than hard-coding a count that's wrong for non-default
    /// `delta_max` deployments.
    pub fn delta_max(&self) -> usize {
        self.delta_max
    }

    /// Append an entry to the delta tier.
    ///
    /// Backpressure: if the delta is at or above `delta_max`, returns
    /// `Error::Backpressure`. The compactor (Phase 5) is responsible for
    /// draining the delta before this happens; receiving Backpressure
    /// here means the compactor is behind.
    ///
    /// Idempotent on rid+seq: if an identical (rid, seq) already exists,
    /// the second append is silently a no-op (recovery may replay the
    /// same op multiple times).
    pub fn append(&self, rid: String, embedding: Vec<f32>, seq: u64) -> Result<()> {
        self.append_inner(rid, embedding, seq, true).map(|_| ())
    }

    /// **v0.10 Item 3 — reserved append.** Append the new vector as an
    /// UNPUBLISHED entry: it holds the delta slot (counts toward capacity,
    /// so Backpressure is reported before a correction commits) but is
    /// invisible to search and skipped by compaction until
    /// [`Self::publish`]. A correction reserves here BEFORE its SQL commit,
    /// then publishes on commit / [`Self::remove_appended`]s on failure —
    /// so a correction's vector never becomes visible (or gets sealed into
    /// cold) unless its text change is durable.
    ///
    /// Returns whether an entry was INSERTED (4a.6d-3). `AlreadyPresent`
    /// means an identical `(rid, seq)` exists — the idempotent-replay no-op —
    /// and the caller owns NO obligation for it: it must neither publish it
    /// (the existing entry's published flag belongs to the write that made
    /// it) nor remove it on failure (that would delete a prior write's
    /// possibly-PUBLISHED vector). Writers minting fresh rids/seqs treat
    /// `AlreadyPresent` as an invariant violation; the deterministic-replay
    /// path (`record_with_rid` with a caller seq) treats it as "already
    /// applied". The outcome is a `#[must_use]` ENUM, not a bool, precisely
    /// so `append_reserved(...)?;` cannot silently discard it (sol 4a.6d-3
    /// r1 finding 1: two correction sites did exactly that — `?` consumes
    /// the Result, and a bare bool then drops without a whisper).
    pub fn append_reserved(
        &self,
        rid: String,
        embedding: Vec<f32>,
        seq: u64,
    ) -> Result<ReservedAppend> {
        self.append_inner(rid, embedding, seq, false)
            .map(|inserted| {
                if inserted {
                    ReservedAppend::Inserted
                } else {
                    ReservedAppend::AlreadyPresent
                }
            })
    }

    fn append_inner(
        &self,
        rid: String,
        embedding: Vec<f32>,
        seq: u64,
        published: bool,
    ) -> Result<bool> {
        if embedding.len() != self.dim {
            return Err(YantrikDbError::InvalidInput(format!(
                "embedding dimension mismatch: expected {}, got {}",
                self.dim,
                embedding.len()
            )));
        }

        let mut delta = self.delta.write();

        // Idempotent on rid+seq: the existing entry stands, whatever its
        // published state — the second arrival owns nothing (4a.6d-3).
        if delta.iter().any(|e| e.rid == rid && e.seq == seq) {
            return Ok(false);
        }

        // Occupancy excludes entries inside a compaction build window:
        // they are still HERE (for read visibility) but logically leaving
        // — `retire_compacted` removes them the moment the merged cold is
        // installed. Counting them would hold the delta at capacity for
        // the whole build and 503 every concurrent write.
        let compacting = self.compacting.load(std::sync::atomic::Ordering::Acquire);
        let occupancy = delta.len().saturating_sub(compacting);
        if occupancy >= self.delta_max {
            return Err(YantrikDbError::Backpressure {
                pending: occupancy as i64,
                max: self.delta_max as i64,
                retry_after_ms: 50,
            });
        }

        let was_empty = delta.is_empty();
        let norm = crate::vector::hnsw::norm_f64(&embedding);
        delta.push(DeltaEntry {
            rid,
            embedding,
            norm,
            seq,
            tombstoned: false,
            published,
        });
        let new_len = delta.len();
        drop(delta); // release write lock before signaling

        // Stamp the dirty-age clock on first non-empty append after every
        // compaction. Subsequent appends within the same dirty window
        // don't touch it — only the oldest entry's age matters for the
        // age-based trigger.
        if was_empty {
            *self.oldest_dirty_at.lock() = Some(crate::time::Instant::now());
        }
        // **Saga task 18 Option 4 (v0.7.2).** Wake the compactor early
        // when delta crosses ~80% of capacity. Without this, the
        // compactor sleeps its 250ms tick and delta can saturate
        // before the next wake (at 1000 wps + delta_max=256, refill
        // takes ~256ms — almost exactly one tick, so half the time
        // the compactor wakes to a saturated delta and reads stall).
        // Cheap notify_one — no contention since the compactor is
        // the only waiter.
        if new_len >= self.delta_max * 80 / 100 {
            self.compactor_wake_cv.notify_one();
        }
        Ok(true)
    }

    /// Append a tombstone for `rid` to the delta tier.
    ///
    /// If the rid exists in the delta as a non-tombstoned entry, that entry
    /// is marked tombstoned in place. Otherwise a deletion marker is
    /// appended (so readers know to skip the rid even if cold contains it).
    ///
    /// Returns `true` if a live delta entry was tombstoned, `false` if the
    /// tombstone was appended as a marker only (rid was in cold or unknown).
    /// Either way the visibility effect is the same: subsequent searches
    /// will not return `rid`.
    pub fn tombstone(&self, rid: &str, seq: u64) -> bool {
        let mut delta = self.delta.write();
        let was_empty = delta.is_empty();
        for entry in delta.iter_mut() {
            // Only tombstone a PUBLISHED live entry in place. An unpublished
            // (reserved-uncommitted) entry is not the live vector and must
            // not be turned into the tombstone — leave it for its own
            // correction to publish/remove; append a marker below instead.
            //
            // And only an entry whose seq does not EXCEED the tombstone's
            // own: seq order is causal order, and a delete that was minted
            // before a write must lose to it. Without the guard, a stale
            // tombstone racing a fresh same-rid append rewrote the NEWER
            // entry's seq and killed it — the newer write silently
            // destroyed (residual race F4 in the 2026-08-15 catalog).
            // A stale delete instead falls through to the marker append
            // below, where seq resolution (highest wins, in search and
            // compaction alike) correctly ignores it.
            if entry.rid == rid && !entry.tombstoned && entry.published && entry.seq <= seq {
                entry.tombstoned = true;
                entry.seq = seq;
                // In-place mutation — delta was non-empty already, so the
                // dirty-age clock is already stamped from the original
                // append. Nothing to do.
                return true;
            }
        }
        // Not in delta (as a published live entry) — append a deletion marker.
        delta.push(DeltaEntry {
            rid: rid.to_string(),
            embedding: Vec::new(), // tombstone marker; embedding never read
            norm: 0.0,
            seq,
            tombstoned: true,
            published: true,
        });
        let new_len = delta.len();
        drop(delta); // release write lock before signaling

        // Stamp the dirty-age clock if this tombstone marker is the only
        // entry in the delta (delta was empty before the push). Same rule
        // as append(): only the *oldest* dirty entry's age matters.
        if was_empty {
            *self.oldest_dirty_at.lock() = Some(crate::time::Instant::now());
        }
        // Saga task 18 Option 4: tombstone-only paths also fill the
        // delta and should wake the compactor at the same threshold.
        if new_len >= self.delta_max * 80 / 100 {
            self.compactor_wake_cv.notify_one();
        }
        false
    }

    /// **v0.10 Item 3 — atomic-append compensation.** Remove the exact
    /// `(rid, seq)` entry appended by a prior `append()`, restoring the
    /// tier's prior visibility. Used ONLY to undo a correction's
    /// superseding append when the correction's SQL transaction fails to
    /// commit: without this the delta would keep the new vector shadowing
    /// the (unchanged) old SQL row. `seq` is monotonic and unique, so it
    /// identifies exactly one entry. Returns whether an entry was removed.
    ///
    /// This is NOT a tombstone: a tombstone would suppress the rid
    /// entirely (hiding the still-valid old vector). Removal simply undoes
    /// the append, so the older delta entry or the cold copy becomes
    /// visible again — the correct state when the correction did not land.
    pub fn remove_appended(&self, rid: &str, seq: u64) -> bool {
        let mut delta = self.delta.write();
        if let Some(pos) = delta.iter().position(|e| e.rid == rid && e.seq == seq) {
            delta.remove(pos);
            true
        } else {
            false
        }
    }

    /// **v0.10 Item 3 — publish a reserved append.** Flip the `(rid, seq)`
    /// entry appended by [`Self::append_reserved`] from reserved to
    /// visible, once the correction's SQL transaction has committed. From
    /// this point the entry participates in search (shadowing the old
    /// vector) and is eligible for compaction. Returns whether an entry was
    /// published (false if it was already sealed away by compaction — which
    /// cannot happen, since compaction skips unpublished entries).
    pub fn publish(&self, rid: &str, seq: u64) -> bool {
        let mut delta = self.delta.write();
        for entry in delta.iter_mut() {
            if entry.rid == rid && entry.seq == seq && !entry.published {
                entry.published = true;
                drop(delta);
                // **v0.10 Item 3 review r3 #4.** Ensure the age-based
                // compaction trigger tracks this now-visible entry: if the
                // dirty-age clock was cleared by a seal that ran while this
                // entry was still reserved, restamp it so the published
                // entry ages into cold rather than sitting in the delta.
                let mut clock = self.oldest_dirty_at.lock();
                if clock.is_none() {
                    *clock = Some(crate::time::Instant::now());
                }
                return true;
            }
        }
        false
    }

    /// Search for the top-k nearest neighbors of `query`.
    ///
    /// Searches both cold and delta, merges by distance, drops tombstoned
    /// rids, and returns up to `k` (rid, distance) pairs sorted ascending.
    ///
    /// If the same rid appears in both tiers, the delta entry wins (it's
    /// strictly newer; cold gets the corresponding update at compaction).
    pub fn search(&self, query: &[f32], k: usize) -> Result<Vec<(String, f64)>> {
        Ok(self
            .search_with_windows(query, k)?
            .into_iter()
            .map(|(rid, dist, _)| (rid, dist))
            .collect())
    }

    /// [`Self::search`] that also reports which chunk window won per
    /// record: `(rid, distance, winning_chunk_idx)` with 0 = the head
    /// window. Recall uses the winner to surface the matched window of
    /// a long record (snippet projection) instead of the whole text.
    pub fn search_with_windows(&self, query: &[f32], k: usize) -> Result<Vec<(String, f64, u32)>> {
        if query.len() != self.dim {
            return Err(YantrikDbError::InvalidInput(format!(
                "query dimension mismatch: expected {}, got {}",
                self.dim,
                query.len()
            )));
        }

        // Query norm computed once for the whole delta scan (v0.9.3).
        let qnorm = crate::vector::hnsw::norm_f64(query);

        // Snapshot cold (Arc clone, no lock) + brief delta read.
        let cold = self.cold.load();
        let delta = self.delta.read();

        // Per-rid winner: highest seq wins. If the winning entry is
        // tombstoned, the rid is dead. Otherwise it is the canonical live
        // entry for the rid (and shadows any cold copy).
        //
        // This handles the archive/hydrate scenario: tombstone(rid, seq=5)
        // followed by append(rid, embedding, seq=10) — the live entry at
        // seq=10 wins, the tombstone at seq=5 loses.
        let mut winner_per_rid: std::collections::HashMap<&str, &DeltaEntry> =
            std::collections::HashMap::new();
        for entry in delta.iter() {
            // **v0.10 Item 3.** Skip RESERVED (unpublished) entries entirely:
            // they are corrections whose SQL is not yet committed, so they
            // are neither a live winner nor a tombstone. Ignoring them means
            // the record is still retrieved under its old, durable vector
            // during the commit window (and is not shadowed away from its
            // cold copy by an entry that may yet be rolled back).
            if !entry.published {
                continue;
            }
            match winner_per_rid.get(entry.rid.as_str()) {
                Some(existing) if existing.seq >= entry.seq => {}
                _ => {
                    winner_per_rid.insert(entry.rid.as_str(), entry);
                }
            }
        }
        let mut tombstoned: std::collections::HashSet<&str> = std::collections::HashSet::new();
        let mut delta_live: Vec<(&DeltaEntry, f64)> = Vec::with_capacity(winner_per_rid.len());
        for (rid, entry) in &winner_per_rid {
            if entry.tombstoned {
                tombstoned.insert(*rid);
            } else {
                // v0.9.3: one-pass dot + precomputed entry norm + qnorm
                // computed once per search (below) — was three passes.
                let d = crate::vector::hnsw::dist_from(
                    crate::vector::hnsw::dot_f64(query, &entry.embedding),
                    qnorm,
                    entry.norm,
                );
                delta_live.push((*entry, d));
            }
        }

        let delta_rid_set: std::collections::HashSet<&str> =
            delta_live.iter().map(|(e, _)| e.rid.as_str()).collect();

        // Search cold for up to k * 2 candidates so we have headroom for
        // tombstone filtering + delta-shadowing without losing top-k.
        //
        // Chunked embeddings can make that headroom insufficient: when a
        // neighborhood is chunk-DENSE (the nearest keys are many windows
        // of the same few records), collapsing to parents can leave
        // fewer than k records even though the cold tier holds more. So
        // the fetch widens geometrically and retries — the common path
        // pays nothing, and the retry is bounded by the worst possible
        // multiplicity (every record at its chunk cap) plus the "cold
        // returned fewer than asked" exit.
        let mut cold_fetch = k.saturating_mul(2).max(k);
        let cold_fetch_cap = k
            .saturating_mul(2)
            .saturating_mul(crate::vector::chunk::MAX_CHUNKS + 1);
        loop {
            let cold_results = cold.search(query, cold_fetch)?;
            let cold_exhausted = cold_results.len() < cold_fetch;

            // Merge: cold + delta, drop cold rids that are in delta_live
            // (delta wins) or tombstoned in delta.
            let mut merged: Vec<(String, f64)> =
                Vec::with_capacity(cold_results.len() + delta_live.len());
            for (rid, dist) in &cold_results {
                if tombstoned.contains(rid.as_str()) || delta_rid_set.contains(rid.as_str()) {
                    continue;
                }
                merged.push((rid.clone(), *dist));
            }
            for (entry, dist) in &delta_live {
                merged.push((entry.rid.clone(), *dist));
            }

            // Sort by distance ascending.
            merged.sort_by(|a, b| a.1.total_cmp(&b.1));
            // Chunked embeddings: collapse `rid#c<N>` window keys to
            // their parent rid, best (lowest-distance) window wins.
            // This is THE choke point — every engine consumer of this
            // index funnels through here, so downstream code (scoring-
            // cache lookups that silently drop unknown keys,
            // RecallResult.rid, MMR, dedupe sets) only ever sees real
            // memories rids, and `k` means k DISTINCT RECORDS — a long
            // record can never crowd the result list with its own
            // windows.
            let mut collapsed = crate::vector::chunk::collapse_to_parents_indexed(merged);
            if collapsed.len() >= k || cold_exhausted || cold_fetch >= cold_fetch_cap {
                collapsed.truncate(k);
                return Ok(collapsed);
            }
            cold_fetch = cold_fetch.saturating_mul(4).min(cold_fetch_cap);
        }
    }

    /// Number of entries in the delta tier (including tombstones).
    pub fn delta_len(&self) -> usize {
        self.delta.read().len()
    }

    /// Number of entries in the cold tier.
    pub fn cold_len(&self) -> usize {
        self.cold.load().len()
    }

    /// Total entry count across both tiers (cold + delta, including
    /// tombstone markers in delta). Approximation: a tombstone in delta
    /// shadowing a live cold entry is counted twice — matches the shape
    /// HnswIndex.len() exposes. Stats callers use this for ballpark
    /// health metrics, not for exact accounting.
    pub fn len(&self) -> usize {
        self.cold_len() + self.delta_len()
    }

    /// True iff both tiers are empty.
    pub fn is_empty(&self) -> bool {
        self.cold_len() == 0 && self.delta_len() == 0
    }

    /// Snapshot the delta entries — used by the compactor (Phase 5) to
    /// build a new cold tier. Returns a clone of the current delta vector;
    /// the actual seal-and-swap is `seal_delta_for_compaction`.
    pub fn snapshot_delta(&self) -> Vec<DeltaEntry> {
        self.delta.read().clone()
    }

    /// Atomically swap the current delta for a fresh empty one.
    /// Returns the sealed delta for the compactor to merge into cold.
    /// Phase 5 wires this into the compaction scheduler.
    ///
    /// Also clears the dirty-age clock — once entries are sealed, the
    /// next append against the now-empty delta restarts the age window.
    pub fn seal_delta_for_compaction(&self) -> Vec<DeltaEntry> {
        let mut delta = self.delta.write();
        // **v0.10 Item 3.** RESERVED (unpublished) entries are corrections
        // whose SQL has not yet committed — they must NOT be sealed into
        // cold (a later commit-failure must be able to remove them, and a
        // commit must be able to publish them into the live delta). Retain
        // them in the new delta; seal only the published/committed entries.
        let mut sealed = Vec::with_capacity(delta.len());
        let mut retained = Vec::with_capacity(self.delta_max.min(4096));
        for entry in delta.drain(..) {
            if entry.published {
                sealed.push(entry);
            } else {
                retained.push(entry);
            }
        }
        let retained_any = !retained.is_empty();
        *delta = retained;
        // Reset the dirty-age clock under the same write lock so the
        // age-trigger calculation can never observe a stale stamp paired
        // with an empty delta. **v0.10 Item 3 review r3 #4:** if RESERVED
        // entries were retained, re-stamp the clock now so those entries
        // (and any that publish into this non-empty delta) still age into
        // compaction — otherwise a low-volume published correction could
        // sit in the linear-scan delta forever, defeating the age trigger.
        *self.oldest_dirty_at.lock() = if retained_any {
            Some(crate::time::Instant::now())
        } else {
            None
        };
        sealed
    }

    /// Atomically install a new cold tier (post-compaction).
    /// Old readers continue against the prior Arc; new readers see the
    /// new tier on their next `cold.load()`.
    pub fn install_cold(&self, new_cold: HnswIndex) {
        self.cold.store(Arc::new(new_cold));
    }

    /// SNAPSHOT (clone, do NOT drain) the published entries for compaction.
    ///
    /// The drain-at-seal design lost read visibility: `seal` removed
    /// published entries from the live delta at the START of compaction,
    /// but the merged cold is not installed until the END (after clone +
    /// N inserts + the reachability BFS). During that window a recall saw
    /// neither the old cold (no entry yet) nor the delta (already drained)
    /// — recently-written records silently vanished on every compaction
    /// cycle (the stored-active-unfindable class). Snapshotting instead
    /// keeps them in the delta, visible, for the whole build; `retire_
    /// compacted` removes them only AFTER the new cold is installed.
    /// Opens the build window for write admission too: while the returned
    /// entries await [`Self::retire_compacted`], they stop counting toward
    /// `delta_max` (they are leaving; holding writers hostage to the build
    /// duration was the compactor-wedge regression). The count is stored
    /// while the delta lock is held, so admission (which holds the write
    /// lock) can never observe the snapshot without the count.
    pub fn snapshot_published_for_compaction(&self) -> Vec<DeltaEntry> {
        let delta = self.delta.read();
        let sealed: Vec<DeltaEntry> = delta.iter().filter(|e| e.published).cloned().collect();
        self.compacting
            .store(sealed.len(), std::sync::atomic::Ordering::Release);
        sealed
    }

    /// Remove exactly the entries merged into cold, by (rid, seq). Reserved
    /// (unpublished) entries and any published entry appended DURING the
    /// build (a newer same-rid write has a higher seq, so a different key)
    /// are retained. Resets the dirty-age clock from what remains.
    pub fn retire_compacted(&self, merged: &std::collections::HashSet<(String, u64)>) {
        let mut delta = self.delta.write();
        delta.retain(|e| !merged.contains(&(e.rid.clone(), e.seq)));
        // Close the admission window opened by the snapshot — the merged
        // entries are physically gone now, so occupancy is honest again.
        self.compacting
            .store(0, std::sync::atomic::Ordering::Release);
        let published_remains = delta.iter().any(|e| e.published);
        *self.oldest_dirty_at.lock() = if published_remains {
            Some(crate::time::Instant::now())
        } else {
            None
        };
    }

    /// **Decoupled write path RFC, Phase 5 — compaction.**
    ///
    /// Drain the current delta into cold by clone-rebuilding the cold tier.
    /// Atomic from the readers' perspective: the ArcSwap.store() at the end
    /// is the visible epoch boundary. Old readers finish on the prior cold
    /// snapshot; new readers see the merged cold.
    ///
    /// Algorithm:
    ///   1. snapshot_published_for_compaction() — CLONE the published
    ///      entries; they stay in the live delta (visible to reads) but
    ///      stop counting toward `delta_max` (writers stay admitted).
    ///   2. Clone the current cold HnswIndex.
    ///   3. For each sealed entry in seq order:
    ///        - tombstoned: HnswIndex::remove(rid) on the clone
    ///        - live with rid not in cold: HnswIndex::insert
    ///        - live with rid in cold (update): remove + re-insert
    ///   4. ensure_all_reachable() on the merged clone.
    ///   5. ArcSwap.store(Arc::new(new_cold)), THEN retire_compacted()
    ///      removes exactly the merged (rid, seq) pairs from the delta.
    ///      At no instant is a sealed entry in neither tier.
    ///
    /// Returns the number of delta entries applied.
    ///
    /// Idempotent on empty delta — returns 0 without touching cold.
    pub fn compact(&self) -> Result<usize> {
        // SNAPSHOT, not drain: the sealed entries stay in the live delta —
        // visible to every recall — until the merged cold is installed
        // below. See snapshot_published_for_compaction for the visibility
        // bug this replaces.
        let sealed = self.snapshot_published_for_compaction();
        if sealed.is_empty() {
            return Ok(0);
        }

        // Clone cold off the live ArcSwap so readers continue against
        // the prior epoch while we build the new one.
        let mut new_cold: HnswIndex = (*self.cold.load_full()).clone();

        // Apply sealed entries by seq order. Same-rid duplicates within the
        // sealed batch resolve via "highest seq wins" — same rule as search.
        let mut by_rid: std::collections::HashMap<String, &DeltaEntry> =
            std::collections::HashMap::with_capacity(sealed.len());
        for entry in &sealed {
            match by_rid.get(&entry.rid) {
                Some(existing) if existing.seq >= entry.seq => {}
                _ => {
                    by_rid.insert(entry.rid.clone(), entry);
                }
            }
        }

        // Apply in SEQ ORDER, as the contract above states. `by_rid.values()`
        // iterated a HashMap — arbitrary, per-process order — which made the
        // compacted graph's edge structure non-deterministic across runs
        // (the exact property the seeded-build work exists to hold).
        let mut ordered: Vec<&DeltaEntry> = by_rid.values().copied().collect();
        ordered.sort_by_key(|e| e.seq);

        let mut applied = 0usize;
        for entry in ordered {
            if entry.tombstoned {
                new_cold.remove(&entry.rid);
            } else {
                // remove first to handle "update" semantics — if the rid
                // was already in cold, the new embedding supersedes it.
                new_cold.remove(&entry.rid);
                if let Err(e) = new_cold.insert(&entry.rid, &entry.embedding) {
                    // Abandon the build: nothing was installed, the delta
                    // still holds every sealed entry, so state is exactly
                    // pre-compaction — but the admission window opened by
                    // the snapshot must close, or occupancy stays
                    // understated by the sealed count until the next
                    // (possibly also failing) cycle.
                    self.compacting
                        .store(0, std::sync::atomic::Ordering::Release);
                    return Err(e);
                }
            }
            applied += 1;
        }

        // The same connectivity guarantee every bulk build ends with. The
        // in-degree guard during pruning is a LOCAL invariant and is known
        // to leave two-node islands (~25% of adversarial builds); until now
        // compaction was the one graph-mutating path that never ran the
        // repair, so an island formed here persisted — stored, active,
        // unfindable — until the next reopen rebuilt the index.
        let rescued = new_cold.ensure_all_reachable();
        if rescued > 0 {
            tracing::warn!(rescued, "delta compaction reconnected unreachable nodes");
        }

        // Install the merged cold FIRST — now every sealed entry is in
        // cold AND still in the delta (search dedupes by highest seq, so
        // no double-count) — THEN retire exactly the merged (rid, seq)
        // pairs from the delta. At no instant is a sealed entry in neither
        // tier, which is the whole point of the snapshot design.
        self.cold.store(Arc::new(new_cold));
        let merged: std::collections::HashSet<(String, u64)> =
            sealed.iter().map(|e| (e.rid.clone(), e.seq)).collect();
        self.retire_compacted(&merged);
        Ok(applied)
    }

    /// Whether the delta should be compacted on the next compactor tick.
    ///
    /// Two triggers, ORed:
    ///
    /// 1. **Size trigger** — `delta_len() >= delta_max / 2`. The
    ///    classic Phase 5 trigger; protects read latency under bursty
    ///    write workloads.
    ///
    /// 2. **Age trigger** — `delta_len() > 0` AND the oldest entry has
    ///    been sitting longer than `max_dirty_age` (default 60s). Closes
    ///    the gap where a low-write namespace's delta sits at, say, 50
    ///    dirty entries forever and reads pay the linear scan
    ///    indefinitely. Bolt-on per epic 5 task 13 (ChatGPT review
    ///    2026-05-08) — explicitly NOT a multi-factor scoring formula.
    ///
    /// The compactor polls this each tick (every COMPACTOR_INTERVAL).
    pub fn should_compact(&self) -> bool {
        // Size trigger.
        if self.delta_len() >= self.delta_max / 2 {
            return true;
        }
        // Age trigger.
        let stamp = *self.oldest_dirty_at.lock();
        match stamp {
            Some(t) if self.delta_len() > 0 && t.elapsed() >= self.max_dirty_age => true,
            _ => false,
        }
    }

    /// **Saga task 18 Option 4 (v0.7.2).** Compactor's wait primitive.
    /// Blocks for up to `timeout` waiting for either:
    /// - An `append()`/`tombstone()` that pushed delta past 80% capacity
    ///   (event-driven wake — wakes within microseconds of pressure).
    /// - The `timeout` expiring (backstop for the age-trigger path
    ///   and for graceful shutdown responsiveness).
    ///
    /// Pre-v0.7.2 the compactor's loop used `thread::sleep(timeout)`,
    /// which at 1000 wps + delta_max=256 meant the delta saturated
    /// inside one tick window, stalling readers. This API replaces
    /// the sleep with an event-driven wait.
    ///
    /// Returns true if woken by signal, false if the timeout fired.
    /// Caller treats both as "go check should_compact again."
    pub fn wait_for_compaction_signal(&self, timeout: std::time::Duration) -> bool {
        let mut guard = self.compactor_wake_mu.lock();
        let result = self.compactor_wake_cv.wait_for(&mut guard, timeout);
        !result.timed_out()
    }
}

// ── Helpers ──
//
// v0.9.3: the local `cosine_distance_f64` (three accumulators recomputing
// both norms per pair, NaN-guarded) was replaced by the shared
// `hnsw::{dot_f64, norm_f64, dist_from}` decomposition — entry norms are
// precomputed at append, the query norm once per search, and `dist_from`
// carries the issue-#60 NaN/zero guard.

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::thread;
    use std::time::Duration;

    fn vec_seed(seed: f32, dim: usize) -> Vec<f32> {
        let raw: Vec<f32> = (0..dim).map(|i| (seed + i as f32) * 0.1).collect();
        let norm: f32 = raw.iter().map(|x| x * x).sum::<f32>().sqrt().max(1e-9);
        raw.iter().map(|x| x / norm).collect()
    }

    #[test]
    fn empty_index_search_returns_empty() {
        let idx = DeltaIndex::new(64);
        let query = vec_seed(1.0, 64);
        let r = idx.search(&query, 10).unwrap();
        assert!(r.is_empty());
    }

    #[test]
    fn append_then_search_finds_in_delta() {
        let idx = DeltaIndex::new(64);
        let emb = vec_seed(1.0, 64);
        idx.append("rid_1".to_string(), emb.clone(), 1).unwrap();

        let query = vec_seed(1.0, 64);
        let r = idx.search(&query, 5).unwrap();
        assert_eq!(r.len(), 1);
        assert_eq!(r[0].0, "rid_1");
        assert!(r[0].1 < 0.001, "exact-match distance ~0, got {}", r[0].1);
    }

    #[test]
    fn reserved_append_is_invisible_until_published() {
        // v0.10 Item 3 finding 2: a reserved (unpublished) append holds the
        // slot but is invisible to search; the OLD vector still wins. On
        // publish it supersedes; on remove it vanishes with the old vector
        // restored.
        let idx = DeltaIndex::new(8);
        // Orthogonal basis vectors so old/new are unambiguously far apart.
        let mut old = vec![0.0f32; 8];
        old[0] = 1.0;
        let mut new = vec![0.0f32; 8];
        new[4] = 1.0;
        idx.append("rid".to_string(), old.clone(), 1).unwrap();

        // Reserve the new vector at a higher seq — still invisible.
        let _ = idx
            .append_reserved("rid".to_string(), new.clone(), 2)
            .unwrap();
        let hit_old = idx.search(&old, 1).unwrap();
        assert_eq!(hit_old[0].0, "rid");
        assert!(
            hit_old[0].1 < 1e-6,
            "reserved entry invisible: old vector wins"
        );
        let hit_new = idx.search(&new, 1).unwrap();
        // Query at the NEW vector still resolves to the rid via its OLD
        // (orthogonal) vector — distance ~1, never the reserved new one.
        assert!(hit_new[0].1 > 0.5, "reserved new vector not searchable yet");

        // Publish → the new vector now wins (exact match to `new`).
        assert!(idx.publish("rid", 2));
        let hit_new2 = idx.search(&new, 1).unwrap();
        assert_eq!(hit_new2[0].0, "rid");
        assert!(hit_new2[0].1 < 1e-6, "published: new vector wins");
    }

    #[test]
    fn compaction_skips_reserved_and_remove_still_works() {
        // v0.10 Item 3 finding 2: compaction must NOT seal a reserved entry
        // (so a commit-failure can still remove it, and a commit can still
        // publish it). After compaction the reserved entry remains in the
        // delta; removing it restores the compacted old vector.
        let idx = DeltaIndex::new(8);
        let mut old = vec![0.0f32; 8];
        old[0] = 1.0;
        let mut new = vec![0.0f32; 8];
        new[4] = 1.0;
        idx.append("rid".to_string(), old.clone(), 1).unwrap();
        let _ = idx
            .append_reserved("rid".to_string(), new.clone(), 2)
            .unwrap();

        // Compact: the published old entry seals into cold; the reserved
        // entry is retained in the delta.
        idx.compact().unwrap();
        assert_eq!(
            idx.delta_len(),
            1,
            "reserved entry retained past compaction"
        );
        // Old vector is now in cold and is still the search winner (reserved
        // is invisible).
        let hit = idx.search(&new, 1).unwrap();
        assert!(hit[0].1 > 0.5, "reserved still invisible after compaction");

        // Commit-failure path: remove the reserved entry → old (cold) wins.
        assert!(idx.remove_appended("rid", 2));
        assert_eq!(idx.delta_len(), 0);
        let hit_old = idx.search(&old, 1).unwrap();
        assert_eq!(hit_old[0].0, "rid");
        assert!(
            hit_old[0].1 < 1e-6,
            "old cold vector intact after reserved removal"
        );
    }

    #[test]
    fn append_dimension_mismatch_rejected() {
        let idx = DeltaIndex::new(64);
        let bad = vec![0.0f32; 32];
        let err = idx
            .append("rid_x".to_string(), bad, 1)
            .expect_err("must reject");
        assert!(matches!(err, YantrikDbError::InvalidInput(_)));
    }

    #[test]
    fn delta_full_returns_backpressure() {
        let idx = DeltaIndex::with_capacity(64, 5);
        for i in 0..5 {
            idx.append(format!("rid_{i}"), vec_seed(i as f32, 64), i as u64)
                .unwrap();
        }
        let err = idx
            .append("rid_overflow".to_string(), vec_seed(99.0, 64), 999)
            .expect_err("must backpressure");
        match err {
            YantrikDbError::Backpressure { pending, max, .. } => {
                assert_eq!(pending, 5);
                assert_eq!(max, 5);
            }
            other => panic!("expected Backpressure, got {other:?}"),
        }
    }

    #[test]
    fn append_idempotent_on_same_rid_seq() {
        let idx = DeltaIndex::new(64);
        let emb = vec_seed(1.0, 64);
        idx.append("rid_1".to_string(), emb.clone(), 1).unwrap();
        idx.append("rid_1".to_string(), emb.clone(), 1).unwrap();
        assert_eq!(idx.delta_len(), 1, "second append at same seq is no-op");
    }

    #[test]
    fn tombstone_hides_rid_from_search() {
        let idx = DeltaIndex::new(64);
        idx.append("rid_keep".to_string(), vec_seed(1.0, 64), 1)
            .unwrap();
        idx.append("rid_drop".to_string(), vec_seed(2.0, 64), 2)
            .unwrap();

        let query = vec_seed(2.0, 64);
        let r_before = idx.search(&query, 5).unwrap();
        assert_eq!(r_before.len(), 2);

        idx.tombstone("rid_drop", 3);
        let r_after = idx.search(&query, 5).unwrap();
        assert_eq!(r_after.len(), 1);
        assert_eq!(r_after[0].0, "rid_keep");
    }

    #[test]
    fn tombstone_on_cold_only_rid_appends_marker() {
        let idx = DeltaIndex::new(64);
        // Pre-populate cold by directly building one.
        let mut cold = HnswIndex::new(64);
        cold.insert("rid_in_cold", &vec_seed(5.0, 64)).unwrap();
        idx.install_cold(cold);

        // Sanity: search finds it.
        let r = idx.search(&vec_seed(5.0, 64), 5).unwrap();
        assert_eq!(r.len(), 1);
        assert_eq!(r[0].0, "rid_in_cold");

        // Tombstone — returns false (not in delta as live), but marker appended.
        let was_in_delta = idx.tombstone("rid_in_cold", 1);
        assert!(!was_in_delta);
        assert_eq!(idx.delta_len(), 1, "tombstone marker appended");

        let r2 = idx.search(&vec_seed(5.0, 64), 5).unwrap();
        assert!(r2.is_empty(), "tombstoned cold rid hidden");
    }

    #[test]
    fn cold_and_delta_merged_in_search() {
        let idx = DeltaIndex::new(64);
        let mut cold = HnswIndex::new(64);
        cold.insert("rid_cold", &vec_seed(1.0, 64)).unwrap();
        idx.install_cold(cold);

        idx.append("rid_delta".to_string(), vec_seed(2.0, 64), 1)
            .unwrap();

        let r = idx.search(&vec_seed(1.5, 64), 5).unwrap();
        let rids: Vec<&str> = r.iter().map(|(rid, _)| rid.as_str()).collect();
        assert!(rids.contains(&"rid_cold"));
        assert!(rids.contains(&"rid_delta"));
        assert_eq!(rids.len(), 2);
    }

    #[test]
    fn delta_shadows_cold_when_rid_appears_in_both() {
        // Same rid in both tiers — delta wins on visibility (it's newer).
        // The merge must not double-count.
        let idx = DeltaIndex::new(64);
        let mut cold = HnswIndex::new(64);
        cold.insert("rid_dup", &vec_seed(1.0, 64)).unwrap();
        idx.install_cold(cold);

        // Same rid with different embedding in delta (simulates an update
        // that hasn't compacted yet).
        idx.append("rid_dup".to_string(), vec_seed(5.0, 64), 1)
            .unwrap();

        let r = idx.search(&vec_seed(5.0, 64), 5).unwrap();
        assert_eq!(r.len(), 1, "no double-count");
        assert_eq!(r[0].0, "rid_dup");
        // Distance should match the DELTA embedding, not cold's.
        assert!(r[0].1 < 0.001);
    }

    #[test]
    fn seal_delta_returns_entries_and_resets() {
        let idx = DeltaIndex::new(64);
        for i in 0..5 {
            idx.append(format!("rid_{i}"), vec_seed(i as f32, 64), i as u64)
                .unwrap();
        }
        let sealed = idx.seal_delta_for_compaction();
        assert_eq!(sealed.len(), 5);
        assert_eq!(idx.delta_len(), 0, "delta reset to empty");

        // Subsequent appends go to the new delta.
        idx.append("rid_after".to_string(), vec_seed(10.0, 64), 100)
            .unwrap();
        assert_eq!(idx.delta_len(), 1);
    }

    #[test]
    fn stale_tombstone_does_not_destroy_newer_write() {
        // Residual race F4: a forget whose seq was minted BEFORE a racing
        // same-rid write used to tombstone that newer entry in place and
        // rewrite its seq — the newer write silently destroyed. Causal
        // (seq) order must decide: the stale delete loses.
        let idx = DeltaIndex::new(64);
        idx.append("rid_x".to_string(), vec_seed(1.0, 64), 10)
            .unwrap();
        assert!(
            !idx.tombstone("rid_x", 5),
            "stale tombstone must fall through to a marker, not hit in place"
        );
        let hits = idx.search(&vec_seed(1.0, 64), 4).unwrap();
        assert!(
            hits.iter().any(|(rid, _)| rid == "rid_x"),
            "a lower-seq tombstone must not destroy a newer write"
        );

        // The legitimate direction is untouched: a delete minted AFTER the
        // write still kills it.
        assert!(idx.tombstone("rid_x", 11));
        let hits = idx.search(&vec_seed(1.0, 64), 4).unwrap();
        assert!(
            !hits.iter().any(|(rid, _)| rid == "rid_x"),
            "a newer tombstone must still suppress the rid"
        );
    }

    #[test]
    fn writes_admitted_while_compaction_build_in_flight() {
        // Regression for the snapshot-and-retire redesign: keeping sealed
        // entries in the delta for read visibility must not also keep them
        // counted against delta_max, or every write during a build 503s
        // (spawn_all_workers_bundles_materializer_and_compactor caught
        // exactly that wedge). This pins the mechanism deterministically:
        // snapshot opens the admission window, retire closes it, and the
        // sealed entries stay searchable for the whole build.
        let idx = DeltaIndex::with_capacity(64, 8);
        for i in 0..8 {
            idx.append(format!("rid_{i}"), vec_seed(i as f32, 64), i as u64)
                .unwrap();
        }
        // Full: admission refused, exactly as before compaction starts.
        assert!(matches!(
            idx.append("rid_full".to_string(), vec_seed(99.0, 64), 99),
            Err(YantrikDbError::Backpressure { .. })
        ));

        // Build window opens: the sealed entries are logically leaving.
        let sealed = idx.snapshot_published_for_compaction();
        assert_eq!(sealed.len(), 8);

        // A write DURING the build is admitted (this is the line that
        // fails on the drain-counting code)...
        idx.append("rid_during".to_string(), vec_seed(50.0, 64), 100)
            .unwrap();
        // ...and the sealed entries are still visible to search — the
        // guarantee the snapshot design exists for.
        let hits = idx.search(&vec_seed(3.0, 64), 12).unwrap();
        assert!(
            hits.iter().any(|(rid, _)| rid == "rid_3"),
            "sealed entry must stay findable during the build"
        );

        // Build ends: retire the merged pairs; occupancy is honest again.
        let merged: std::collections::HashSet<(String, u64)> =
            sealed.iter().map(|e| (e.rid.clone(), e.seq)).collect();
        idx.retire_compacted(&merged);
        assert_eq!(idx.delta_len(), 1, "only the during-build write remains");
        idx.append("rid_after".to_string(), vec_seed(60.0, 64), 101)
            .unwrap();
    }

    #[test]
    fn install_cold_atomically_swaps() {
        let idx = DeltaIndex::new(64);
        assert_eq!(idx.cold_len(), 0);

        let mut new_cold = HnswIndex::new(64);
        new_cold.insert("rid_a", &vec_seed(1.0, 64)).unwrap();
        new_cold.insert("rid_b", &vec_seed(2.0, 64)).unwrap();
        idx.install_cold(new_cold);
        assert_eq!(idx.cold_len(), 2);
    }

    // ── Epic 5 task 13: age-based compaction trigger ──
    //
    // The size-based trigger fires only when delta_len >= delta_max/2.
    // Without an age trigger, a low-write namespace whose delta sits at
    // e.g. 10 dirty entries with no new writes will *never* compact, so
    // reads against it pay the linear delta scan indefinitely. These
    // tests exercise the age-trigger path that closes that gap.
    //
    // We use `with_capacity_and_age` to set `max_dirty_age` to ~50ms
    // so the test wall-clock waits stay bounded; the production default
    // is 60s.

    #[test]
    fn age_trigger_does_not_fire_on_empty_delta() {
        // Empty delta + no oldest_dirty_at stamp => should_compact == false
        // even after waiting past max_dirty_age.
        let idx = DeltaIndex::with_capacity_and_age(64, 256, Duration::from_millis(20));
        std::thread::sleep(Duration::from_millis(40));
        assert!(
            !idx.should_compact(),
            "empty delta never triggers age compaction"
        );
    }

    #[test]
    fn age_trigger_fires_after_max_dirty_age_elapses() {
        // 10 entries (well under half-cap of 128) sitting for >50ms with
        // a 20ms max_dirty_age must trigger compaction by the age path.
        let idx = DeltaIndex::with_capacity_and_age(64, 256, Duration::from_millis(20));
        for i in 0..10 {
            idx.append(format!("rid_{i}"), vec_seed(i as f32, 64), i as u64)
                .unwrap();
        }
        assert_eq!(idx.delta_len(), 10);
        assert!(
            !idx.should_compact(),
            "below half-cap and within max_dirty_age window must NOT trigger"
        );
        std::thread::sleep(Duration::from_millis(50));
        assert!(
            idx.should_compact(),
            "10 entries sitting for >max_dirty_age must trigger age path"
        );
    }

    #[test]
    fn age_trigger_resets_on_seal() {
        // After seal_delta_for_compaction, the oldest_dirty_at clock
        // resets — a subsequent append starts a fresh age window.
        let idx = DeltaIndex::with_capacity_and_age(64, 256, Duration::from_millis(20));
        idx.append("rid_a".to_string(), vec_seed(1.0, 64), 1)
            .unwrap();
        std::thread::sleep(Duration::from_millis(30));
        assert!(idx.should_compact(), "first window: age trigger fires");

        let _ = idx.seal_delta_for_compaction();
        assert!(!idx.should_compact(), "seal cleared dirty-age clock");

        // New append after seal: fresh age window, must NOT trigger immediately.
        idx.append("rid_b".to_string(), vec_seed(2.0, 64), 2)
            .unwrap();
        assert!(!idx.should_compact(), "fresh window after seal");
        std::thread::sleep(Duration::from_millis(30));
        assert!(
            idx.should_compact(),
            "second window: age trigger fires again"
        );
    }

    #[test]
    fn age_trigger_compacts_low_write_namespace_end_to_end() {
        // The "end-to-end" test: 10 entries, sit for >max_dirty_age,
        // run compact() (simulating the compactor tick), entries land in cold.
        let idx = DeltaIndex::with_capacity_and_age(64, 256, Duration::from_millis(20));
        for i in 0..10 {
            idx.append(format!("rid_{i}"), vec_seed(i as f32, 64), i as u64)
                .unwrap();
        }
        std::thread::sleep(Duration::from_millis(40));
        assert!(idx.should_compact(), "age trigger ready");
        let n = idx.compact().unwrap();
        assert_eq!(n, 10, "all 10 entries applied to cold");
        assert_eq!(idx.delta_len(), 0, "delta drained");
        assert_eq!(idx.cold_len(), 10, "cold absorbed all entries");
        // After compact, the dirty-age clock is reset (seal cleared it).
        assert!(!idx.should_compact(), "post-compact: nothing to do");
    }

    #[test]
    fn age_trigger_tombstone_only_delta_also_fires() {
        // Edge case: delta contains only a tombstone marker (no live
        // append). The age trigger must still fire — readers care about
        // the tombstone reaching cold so the rid no longer surfaces in
        // recall. Same `delta_len > 0` rule applies.
        let idx = DeltaIndex::with_capacity_and_age(64, 256, Duration::from_millis(20));
        // tombstone() against a rid not in delta appends a marker.
        let was_live = idx.tombstone("rid_remote", 1);
        assert!(!was_live, "tombstone of unknown rid is appended as marker");
        assert_eq!(idx.delta_len(), 1);
        std::thread::sleep(Duration::from_millis(30));
        assert!(
            idx.should_compact(),
            "tombstone-only delta also fires by age"
        );
    }

    #[test]
    fn concurrent_appends_and_reads_no_corruption() {
        let idx = Arc::new(DeltaIndex::with_capacity(64, 1024));

        // Spawn 4 writer threads, each appending 50 entries with disjoint rid spaces.
        let mut writer_handles = Vec::new();
        for w in 0..4 {
            let idx_c = Arc::clone(&idx);
            writer_handles.push(thread::spawn(move || {
                for i in 0..50 {
                    let rid = format!("w{w}_rid_{i}");
                    let emb = vec_seed((w * 100 + i) as f32, 64);
                    idx_c.append(rid, emb, (w * 100 + i) as u64).unwrap();
                }
            }));
        }

        // Spawn 4 reader threads, each doing 100 searches concurrently.
        let mut reader_handles = Vec::new();
        for r in 0..4 {
            let idx_c = Arc::clone(&idx);
            reader_handles.push(thread::spawn(move || {
                for i in 0..100 {
                    let q = vec_seed((r * 1000 + i) as f32, 64);
                    let _ = idx_c.search(&q, 10).unwrap();
                }
            }));
        }

        for h in writer_handles {
            h.join().unwrap();
        }
        for h in reader_handles {
            h.join().unwrap();
        }

        assert_eq!(idx.delta_len(), 200, "all 4 writers contributed 50 each");
    }

    #[test]
    fn search_returns_top_k_sorted_ascending() {
        let idx = DeltaIndex::new(64);
        // Insert 5 distinct entries; query against one of them.
        for i in 0..5 {
            idx.append(format!("rid_{i}"), vec_seed(i as f32, 64), i as u64)
                .unwrap();
        }
        let r = idx.search(&vec_seed(2.0, 64), 3).unwrap();
        assert_eq!(r.len(), 3);
        // Distances must be non-decreasing.
        for w in r.windows(2) {
            assert!(w[0].1 <= w[1].1, "distances must sort ascending");
        }
        // Top result should be rid_2 (exact match).
        assert_eq!(r[0].0, "rid_2");
    }

    #[test]
    fn from_cold_preserves_existing_entries() {
        let mut cold = HnswIndex::new(64);
        cold.insert("rid_a", &vec_seed(1.0, 64)).unwrap();
        cold.insert("rid_b", &vec_seed(2.0, 64)).unwrap();
        let idx = DeltaIndex::from_cold(cold, 64);
        assert_eq!(idx.cold_len(), 2);
        assert_eq!(idx.delta_len(), 0);

        let r = idx.search(&vec_seed(1.0, 64), 5).unwrap();
        assert_eq!(r.len(), 2);
    }

    #[test]
    fn should_compact_at_half_capacity() {
        let idx = DeltaIndex::with_capacity(64, 10);
        assert!(!idx.should_compact());
        for i in 0..4 {
            idx.append(format!("rid_{i}"), vec_seed(i as f32, 64), i as u64)
                .unwrap();
        }
        assert!(!idx.should_compact(), "below half cap");
        idx.append("rid_5".to_string(), vec_seed(5.0, 64), 5)
            .unwrap();
        assert!(idx.should_compact(), "at half cap = should compact");
    }

    #[test]
    fn compact_drains_delta_into_cold() {
        let idx = DeltaIndex::new(64);
        for i in 0..10 {
            idx.append(format!("rid_{i}"), vec_seed(i as f32, 64), i as u64)
                .unwrap();
        }
        assert_eq!(idx.delta_len(), 10);
        assert_eq!(idx.cold_len(), 0);

        let n = idx.compact().unwrap();
        assert_eq!(n, 10);
        assert_eq!(idx.delta_len(), 0, "delta drained");
        assert_eq!(idx.cold_len(), 10, "cold has all 10 entries now");

        // Search still finds them (now in cold tier).
        let r = idx.search(&vec_seed(5.0, 64), 3).unwrap();
        assert_eq!(r.len(), 3);
        assert_eq!(r[0].0, "rid_5", "exact match still found post-compaction");
    }

    #[test]
    fn compact_applies_tombstones_to_cold() {
        let idx = DeltaIndex::new(64);
        // Pre-seed cold with two rids.
        let mut cold = HnswIndex::new(64);
        cold.insert("rid_keep", &vec_seed(1.0, 64)).unwrap();
        cold.insert("rid_drop", &vec_seed(2.0, 64)).unwrap();
        idx.install_cold(cold);
        assert_eq!(idx.cold_len(), 2);

        // Tombstone rid_drop in delta.
        idx.tombstone("rid_drop", 1);
        // Compact applies the tombstone to cold.
        let n = idx.compact().unwrap();
        assert_eq!(n, 1);
        // HnswIndex.len() includes tombstoned nodes (same as pre-Phase 4),
        // but search filters them. Verify via search:
        let r = idx.search(&vec_seed(2.0, 64), 5).unwrap();
        let rids: Vec<&str> = r.iter().map(|(rid, _)| rid.as_str()).collect();
        assert!(!rids.contains(&"rid_drop"), "tombstone applied to cold");
        assert!(rids.contains(&"rid_keep"));
    }

    #[test]
    fn compact_applies_archive_then_hydrate_correctly() {
        // The scenario engine_tests::test_hydrate_memory exercises:
        //   1. record(rid)         -> append at seq=1
        //   2. archive(rid)        -> tombstone at seq=2
        //   3. hydrate(rid)        -> append at seq=3
        // Compact must produce a cold where rid_X is LIVE (highest seq wins).
        let idx = DeltaIndex::new(64);
        idx.append("rid_X".to_string(), vec_seed(5.0, 64), 1)
            .unwrap();
        idx.tombstone("rid_X", 2);
        idx.append("rid_X".to_string(), vec_seed(5.0, 64), 3)
            .unwrap();

        let n = idx.compact().unwrap();
        assert_eq!(n, 1, "highest-seq winner applied once");

        let r = idx.search(&vec_seed(5.0, 64), 5).unwrap();
        assert_eq!(r.len(), 1);
        assert_eq!(r[0].0, "rid_X", "rid alive in cold post-compaction");
    }

    #[test]
    fn compact_idempotent_on_empty_delta() {
        let idx = DeltaIndex::new(64);
        assert_eq!(idx.compact().unwrap(), 0);
        assert_eq!(idx.compact().unwrap(), 0);
    }

    #[test]
    fn compact_preserves_in_flight_writes() {
        // Writers appending during a compaction window must not be lost.
        // Sealing swaps in a fresh delta atomically, so writes that arrive
        // after the swap go to the new delta. The compactor sees only the
        // sealed entries.
        let idx = DeltaIndex::new(64);
        for i in 0..5 {
            idx.append(format!("before_{i}"), vec_seed(i as f32, 64), i as u64)
                .unwrap();
        }
        // Seal manually + start "compaction" but before applying, append more.
        let sealed = idx.seal_delta_for_compaction();
        assert_eq!(sealed.len(), 5);
        assert_eq!(idx.delta_len(), 0, "fresh delta after seal");

        for i in 0..3 {
            idx.append(
                format!("after_{i}"),
                vec_seed((100 + i) as f32, 64),
                (100 + i) as u64,
            )
            .unwrap();
        }
        assert_eq!(
            idx.delta_len(),
            3,
            "after-seal writes accumulate in new delta"
        );

        // Note: the actual compactor (compact()) does seal+apply in one step.
        // This test exercises the seal-only primitive to prove the swap
        // does not lose subsequent writes.
    }

    #[test]
    fn compact_threshold_drives_compaction() {
        // Realistic loop: every time should_compact() returns true, run
        // compact(), and the delta gets bounded.
        let idx = DeltaIndex::with_capacity(64, 10);
        for i in 0..50 {
            // If we are above half-cap, compact first to make room.
            if idx.should_compact() {
                idx.compact().unwrap();
            }
            idx.append(format!("rid_{i}"), vec_seed(i as f32, 64), i as u64)
                .unwrap();
        }
        // After 50 inserts with periodic compaction, cold has all 50.
        idx.compact().unwrap(); // drain final delta
        assert_eq!(idx.cold_len(), 50);
        assert_eq!(idx.delta_len(), 0);
    }
}