sqz_engine/

cache_manager.rs

use std::collections::HashMap;
use std::path::Path;
use std::time::Duration;

use sha2::{Digest, Sha256};

use crate::delta_encoder::DeltaEncoder;
use crate::error::Result;
use crate::pipeline::{CompressionPipeline, SessionContext};
use crate::preset::Preset;
use crate::session_store::SessionStore;
use crate::types::CompressedContent;

/// Outcome of a cache lookup in [`CacheManager`].
///
/// The cache has three possible outcomes:
/// - **Dedup**: exact match, returns a tiny `§ref:HASH§` token (~13 tokens)
/// - **Delta**: near-duplicate, returns a compact diff against the cached version
/// - **Fresh**: cache miss, returns the full compressed output
pub enum CacheResult {
    /// Previously seen content — returns a short inline reference (~13 tokens).
    Dedup {
        /// Inline token of the form `§ref:<hash_prefix>§`.
        inline_ref: String,
        /// Approximate token cost of the reference (always 13).
        token_cost: u32,
    },
    /// Near-duplicate of cached content — returns a compact delta.
    Delta {
        /// The delta text (header + changed lines).
        delta_text: String,
        /// Approximate token cost of the delta.
        token_cost: u32,
        /// Similarity to the cached version (0.0–1.0).
        similarity: f64,
    },
    /// Content not seen before — full compression result.
    Fresh { output: CompressedContent },
}

/// Result of resolving a dedup-ref prefix via
/// [`CacheManager::expand_prefix`].
///
/// Two shapes because not every stored entry can be round-tripped to the
/// raw bytes: cache entries written before the `original` column was
/// introduced (or by a code path that didn't thread originals through)
/// fall into `CompressedOnly`. The CLI uses this to pick a message for
/// the user.
#[derive(Debug, Clone)]
pub enum ExpandResult {
    /// Full pre-compression bytes available — this is what `sqz expand`
    /// was invented to return.
    Original {
        /// Full 64-hex SHA-256 of the original content. Lets callers
        /// print `expanded a1b2c3d4 (full hash a1b2c3d4…)`.
        hash: String,
        /// The raw bytes the agent asked to see.
        bytes: Vec<u8>,
    },
    /// Only the compressed form is stored. Agent still gets legible
    /// output; the CLI attaches a note so the user knows why re-running
    /// uncompressed may be worthwhile.
    CompressedOnly {
        hash: String,
        /// Compressed text as served to the LLM originally.
        compressed: String,
    },
}

/// A dedup hit with the metadata regret tracking needs, returned by
/// [`CacheManager::check_dedup_with_meta`].
#[derive(Debug, Clone)]
pub struct DedupHit {
    /// The `§ref:…§` token to serve instead of the full content.
    pub inline_ref: String,
    /// Full 64-hex SHA-256 of the content.
    pub hash: String,
    /// The entry's `accessed_at` before this lookup bumped it. `None`
    /// when the column could not be read.
    pub prev_accessed_at: Option<chrono::DateTime<chrono::Utc>>,
}

/// A ranged re-read of content the model already has, returned by
/// [`CacheManager::check_slice_with_meta`]. The agent read a file in
/// full earlier and is now reading lines `start_line..=end_line` of it.
#[derive(Debug, Clone)]
pub struct SliceHit {
    /// `§ref:<hash_prefix>:L<start>-<end>§`, or a plain `§ref:<hash_prefix>§`
    /// when the range covers the whole original.
    pub inline_ref: String,
    /// Full 64-hex SHA-256 of the containing original.
    pub hash: String,
    pub start_line: usize,
    pub end_line: usize,
    /// Line count of the containing original.
    pub total_lines: usize,
    /// Approximate token cost of the reference.
    pub token_cost: u32,
}

/// Smallest slice worth referencing. Below this the ref token costs
/// about as much as the text, and short multi-line snippets are common
/// enough to match by coincidence.
const MIN_SLICE_BYTES: usize = 160;
/// How many recent originals to scan per lookup.
const SLICE_CANDIDATES: usize = 32;

/// Tracks when a dedup ref was last sent, so we can detect staleness.
///
/// Historically used for an in-memory per-process turn counter; now kept
/// only for interface compatibility (clear on notify_compaction). Actual
/// staleness is computed from SQLite `accessed_at` timestamps so it works
/// across the shell-hook invocation model where each sqz process is short-
/// lived. See the comment on `is_ref_fresh` for details.
#[derive(Debug, Clone)]
#[allow(dead_code)]
struct RefEntry {
    /// The turn number when this ref was last sent to the LLM.
    last_sent_turn: u64,
}

/// SHA-256 content-hash deduplication cache backed by [`SessionStore`],
/// with delta encoding for near-duplicate content and compaction awareness.
///
/// # Freshness model
///
/// A dedup ref is considered fresh (safe to serve instead of the full
/// content) when the cache entry's `accessed_at` timestamp in SQLite is
/// within `max_ref_age` of now. When sqz is invoked from shell hooks each
/// invocation is a short-lived process, so the freshness check must be
/// persistent — in-memory state is gone the moment the process exits.
///
/// The previous turn-counter heuristic was in-memory only and therefore
/// never registered freshness across hook invocations, which silently
/// disabled the dedup feature in production. Issue found April 18 2026.
///
/// Default TTL: 30 minutes. Empirically matches a typical active coding
/// session before a context compaction. Use
/// [`with_ref_age`](CacheManager::with_ref_age) to tune.
pub struct CacheManager {
    store: SessionStore,
    max_size_bytes: u64,
    delta_encoder: DeltaEncoder,
    /// Retained for notify_compaction's semantic ("forget all tracked refs"),
    /// but no longer consulted for freshness checks.
    #[allow(dead_code)]
    turn_counter: std::cell::Cell<u64>,
    /// Retained for notify_compaction; cleared on compaction events.
    #[allow(dead_code)]
    ref_tracker: std::cell::RefCell<HashMap<String, RefEntry>>,
    /// Maximum wall-clock age before a dedup ref is considered stale.
    /// After this duration we assume the LLM's context window has rolled
    /// over enough to have dropped the original content, so we re-send the
    /// full version instead of a dangling ref.
    max_ref_age: Duration,
    /// Records the instant at which the in-memory compaction flag was set.
    /// Any cache entry whose `accessed_at` predates this instant is stale.
    /// Reset by [`notify_compaction`].
    compaction_marker: std::cell::Cell<Option<chrono::DateTime<chrono::Utc>>>,
    /// Session-lifetime MinHash LSH index over stored entries, used to find
    /// near-duplicate candidates for delta encoding beyond the recent-LRU
    /// window. In-memory only; rebuilt organically as entries are stored.
    lsh: std::cell::RefCell<crate::minhash_lsh::MinHashLsh>,
    /// doc_id (index) → content hash for entries in `lsh`.
    lsh_hashes: std::cell::RefCell<Vec<String>>,
}

impl CacheManager {
    /// Create a new cache manager backed by the given session store.
    ///
    /// `max_size_bytes` controls when LRU eviction kicks in. A good default
    /// is 512 MB (`512 * 1024 * 1024`). Dedup refs go stale after 30 minutes
    /// of wall-clock time by default; use [`with_ref_age`](Self::with_ref_age)
    /// to tune.
    pub fn new(store: SessionStore, max_size_bytes: u64) -> Self {
        Self::with_ref_age_duration(store, max_size_bytes, Duration::from_secs(30 * 60))
    }

    /// Create a CacheManager with a custom ref staleness threshold measured
    /// in turns. The turn count is converted to wall-clock time by assuming
    /// ~1 minute per turn (a rough approximation; the real freshness check
    /// uses SQLite timestamps). This constructor exists for backward
    /// compatibility with tests that previously advanced a turn counter.
    #[doc(hidden)]
    pub fn with_ref_age(store: SessionStore, max_size_bytes: u64, max_ref_age_turns: u64) -> Self {
        Self::with_ref_age_duration(
            store,
            max_size_bytes,
            Duration::from_secs(max_ref_age_turns.saturating_mul(60)),
        )
    }

    /// Create a CacheManager with an explicit wall-clock ref-age cap.
    pub fn with_ref_age_duration(
        store: SessionStore,
        max_size_bytes: u64,
        max_ref_age: Duration,
    ) -> Self {
        Self {
            store,
            max_size_bytes,
            delta_encoder: DeltaEncoder::new(),
            turn_counter: std::cell::Cell::new(0),
            ref_tracker: std::cell::RefCell::new(HashMap::new()),
            max_ref_age,
            compaction_marker: std::cell::Cell::new(None),
            lsh: std::cell::RefCell::new(crate::minhash_lsh::MinHashLsh::new()),
            lsh_hashes: std::cell::RefCell::new(Vec::new()),
        }
    }

    /// Register a stored entry in the similarity index. Capped so a very
    /// long session can't grow the index unboundedly; beyond the cap the
    /// recent-LRU fallback in [`Self::try_delta_encode`] still applies.
    fn index_for_similarity(&self, hash: &str, data: &str) {
        const MAX_INDEXED: usize = 512;
        let mut hashes = self.lsh_hashes.borrow_mut();
        if hashes.len() >= MAX_INDEXED || hashes.iter().any(|h| h == hash) {
            return;
        }
        let doc_id = hashes.len() as u64;
        self.lsh.borrow_mut().insert(doc_id, data);
        hashes.push(hash.to_string());
    }

    /// Compute the SHA-256 hex digest of `bytes`.
    pub fn sha256_hex(bytes: &[u8]) -> String {
        let mut hasher = Sha256::new();
        hasher.update(bytes);
        format!("{:x}", hasher.finalize())
    }

    /// Advance the turn counter. Retained for API compatibility; not used
    /// for freshness. The context_evictor still reads `current_turn` for
    /// LRU scoring during `sqz compact`.
    pub fn advance_turn(&self) {
        self.turn_counter.set(self.turn_counter.get() + 1);
    }

    /// Get the current turn number. Used by the context_evictor for scoring.
    pub fn current_turn(&self) -> u64 {
        self.turn_counter.get()
    }

    /// Notify the cache that a context compaction has occurred.
    ///
    /// Persists a compaction timestamp into the session store so any cache
    /// entry whose `accessed_at` predates the marker is considered stale
    /// by **every subsequent sqz process**, not just this one. The shell-
    /// hook invocation model means this method is typically called from a
    /// short-lived `sqz hook precompact` process, and the check runs in a
    /// different `sqz compress` process milliseconds later.
    ///
    /// Call this when:
    /// - The harness signals a compaction event (PreCompact hook)
    /// - A session is resumed after being idle
    /// - The user runs `sqz compact`
    pub fn notify_compaction(&self) {
        let now = chrono::Utc::now();
        self.compaction_marker.set(Some(now));
        self.ref_tracker.borrow_mut().clear();
        // Persist the marker so other sqz processes see the invalidation.
        // Silently swallow a write error: losing the marker means some
        // refs may survive the compaction and show as dedup hits in the
        // next few calls — annoying, not wrong (the agent still receives
        // valid content; it just sees a short-ref it has to resolve).
        let _ = self
            .store
            .set_metadata("last_compaction_at", &now.to_rfc3339());
    }

    /// Check if a dedup ref for the given hash is still fresh (likely still
    /// in the LLM's context window).
    ///
    /// Uses the SQLite `accessed_at` timestamp rather than the in-memory
    /// turn counter. This works across sqz process invocations: shell hooks
    /// spawn a new sqz process per intercepted command, so any in-memory
    /// counter would reset every time. The database survives.
    ///
    /// The compaction marker is read from SQLite on every check so that
    /// a `sqz hook precompact` call from another process immediately
    /// invalidates refs in the current process. Without the persistent
    /// read, the invalidation would only affect the process that called
    /// notify_compaction — which is never the same process that serves
    /// dedup hits.
    fn is_ref_fresh(&self, hash: &str) -> bool {
        // Pinned entries bypass TTL and compaction-marker checks. They
        // represent stable knowledge-base content that's persistently
        // re-injected into every agent, so the original is never gone.
        if let Ok(true) = self.store.is_cache_entry_pinned(hash) {
            return true;
        }
        let accessed = match self.store.get_cache_entry_accessed_at(hash) {
            Ok(Some(ts)) => ts,
            _ => return false,
        };
        // In-memory compaction marker (set in this process).
        if let Some(marker) = self.compaction_marker.get() {
            if accessed < marker {
                return false;
            }
        }
        // Persistent compaction marker — set by `sqz hook precompact` in
        // a different process. Without this read the in-memory marker is
        // never consulted because each hook invocation is a fresh process.
        if let Ok(Some(raw)) = self.store.get_metadata("last_compaction_at") {
            if let Ok(marker) = raw.parse::<chrono::DateTime<chrono::Utc>>() {
                if accessed < marker {
                    return false;
                }
            }
        }
        let age = (chrono::Utc::now() - accessed)
            .to_std()
            .unwrap_or(Duration::from_secs(0));
        age < self.max_ref_age
    }

    /// Record that a dedup ref was sent for the given hash. Updates the
    /// persistent `accessed_at` timestamp so subsequent freshness checks
    /// see this send. Silently swallows SQLite errors — losing a touch
    /// means the next call may treat the ref as stale and re-send, which
    /// is strictly worse on tokens but never wrong.
    fn record_ref_sent(&self, hash: &str) {
        let _ = self.store.touch_cache_entry(hash);
    }

    /// Look up `content` in the cache with compaction awareness.
    ///
    /// - On exact dedup with fresh ref: return `CacheResult::Dedup` (~13 tokens).
    /// - On exact dedup with stale ref: re-compress and return `CacheResult::Fresh`
    ///   (the original content may have been compacted out of the LLM's context).
    /// - On near-duplicate: return `CacheResult::Delta` with a compact diff.
    /// - On cache miss: compress via `pipeline`, persist, return `CacheResult::Fresh`.
    pub fn get_or_compress(
        &self,
        path: &Path,
        content: &[u8],
        pipeline: &CompressionPipeline,
    ) -> Result<CacheResult> {
        let ctx = SessionContext {
            session_id: "cache".to_string(),
        };
        let preset = Preset::default();
        self.get_or_compress_inner(path, content, &|text| pipeline.compress(text, &ctx, &preset))
    }

    /// [`get_or_compress`](Self::get_or_compress) with the caller's own
    /// compressor for misses and stale refs, so a cached result goes through
    /// the same verification as an uncached one.
    pub fn get_or_compress_with(
        &self,
        path: &Path,
        content: &[u8],
        compress: &dyn Fn(&str) -> Result<CompressedContent>,
    ) -> Result<CacheResult> {
        self.get_or_compress_inner(path, content, compress)
    }

    /// Lossless variant used by the file-read MCP tools (`sqz_read_file` /
    /// `sqz_grep` / `sqz_list_dir`): on a cache miss the content is returned
    /// faithfully via [`CompressionPipeline::compress_lossless`] (no entropy
    /// truncation / pruning / re-encoding). The general `compress` tool keeps
    /// using [`get_or_compress`](Self::get_or_compress) for aggressive
    /// compression. See issue #32.
    pub fn get_or_compress_lossless(
        &self,
        path: &Path,
        content: &[u8],
        pipeline: &CompressionPipeline,
    ) -> Result<CacheResult> {
        let ctx = SessionContext {
            session_id: "cache".to_string(),
        };
        let preset = Preset::default();
        self.get_or_compress_inner(path, content, &|text| {
            pipeline.compress_lossless(text, &ctx, &preset)
        })
    }

    fn get_or_compress_inner(
        &self,
        _path: &Path,
        content: &[u8],
        compress: &dyn Fn(&str) -> Result<CompressedContent>,
    ) -> Result<CacheResult> {
        let hash = Self::sha256_hex(content);

        // Exact match — check if the ref is still fresh
        // Exact match — probe without touching accessed_at, then check
        // freshness. Touching on the probe would make every ref appear
        // fresh immediately (the timestamp we just wrote is `now`).
        let exists = self.store.cache_entry_exists(&hash)?;
        if exists {
            if self.is_ref_fresh(&hash) {
                // Ref is fresh — the LLM likely still has the original in context
                let hash_prefix = &hash[..16];
                let inline_ref = format!("§ref:{hash_prefix}§");
                // Update the sent timestamp
                self.record_ref_sent(&hash);
                let _ = self.store.record_cache_hit().map_err(|e| eprintln!("[sqz] cache hit record error: {e}"));
                return Ok(CacheResult::Dedup {
                    inline_ref,
                    token_cost: 13,
                });
            } else {
                // Ref is stale — re-send the full compressed content.
                // The original may have been compacted out of the LLM's context.
                let _ = self.store.record_cache_hit().map_err(|e| eprintln!("[sqz] cache stale-hit record error: {e}"));
                let text = String::from_utf8_lossy(content).into_owned();
                let compressed = compress(&text)?;
                // Record that we re-sent this content
                self.record_ref_sent(&hash);
                return Ok(CacheResult::Fresh { output: compressed });
            }
        }

        // Ranged re-read of a file the model already has in full.
        if let Some(hit) = self.check_slice_with_meta(content)? {
            return Ok(CacheResult::Dedup {
                inline_ref: hit.inline_ref,
                token_cost: hit.token_cost,
            });
        }

        // Near-duplicate check: compare against recent cache entries
        let text = String::from_utf8_lossy(content).into_owned();
        let _ = self.store.record_cache_miss().map_err(|e| eprintln!("[sqz] cache miss record error: {e}"));
        if let Some(delta_result) = self.try_delta_encode(&text)? {
            // Store the new content in cache for future exact matches
            let compressed = compress(&text)?;
            // Persist the raw bytes so `sqz expand <prefix>` can round-trip.
            self.store
                .save_cache_entry_with_original(&hash, &compressed, Some(content))?;
            self.index_for_similarity(&hash, &compressed.data);
            self.record_ref_sent(&hash);

            let token_cost = (delta_result.delta_text.len() / 4) as u32;
            return Ok(CacheResult::Delta {
                delta_text: delta_result.delta_text,
                token_cost: token_cost.max(5),
                similarity: delta_result.similarity,
            });
        }

        let compressed = compress(&text)?;
        self.store
            .save_cache_entry_with_original(&hash, &compressed, Some(content))?;
        self.index_for_similarity(&hash, &compressed.data);
        // Record that this content was sent at the current turn
        self.record_ref_sent(&hash);

        Ok(CacheResult::Fresh { output: compressed })
    }

    /// Try to delta-encode content against cached entries.
    /// Returns Some(DeltaResult) if a near-duplicate was found.
    ///
    /// Candidates come from two sources, in order:
    /// 1. The MinHash LSH index — finds near-duplicates anywhere in the
    ///    session, not just the recent window. Candidates are ranked by
    ///    estimated Jaccard similarity and the best 5 are tried.
    /// 2. Fallback: the 10 most recent LRU entries (covers content stored
    ///    before this process built its in-memory index).
    fn try_delta_encode(
        &self,
        new_content: &str,
    ) -> Result<Option<crate::delta_encoder::DeltaResult>> {
        // LSH path.
        let candidates: Vec<String> = {
            let lsh = self.lsh.borrow();
            let hashes = self.lsh_hashes.borrow();
            let query_sig = crate::minhash_lsh::MinHashLsh::compute_signature(new_content);
            let mut scored: Vec<(f64, &String)> = lsh
                .query(new_content)
                .into_iter()
                .filter_map(|doc_id| {
                    let hash = hashes.get(doc_id as usize)?;
                    let sim = lsh.get_signature(doc_id)?.jaccard_similarity(&query_sig);
                    (sim >= 0.5).then_some((sim, hash))
                })
                .collect();
            scored.sort_by(|a, b| b.0.partial_cmp(&a.0).unwrap_or(std::cmp::Ordering::Equal));
            scored.into_iter().take(5).map(|(_, h)| h.clone()).collect()
        };
        for hash in &candidates {
            if let Some(delta) = self.try_delta_against(hash, new_content)? {
                return Ok(Some(delta));
            }
        }

        // Recent-LRU fallback.
        let entries = self.store.list_cache_entries_lru()?;
        let check_count = entries.len().min(10);
        for (hash, _) in entries.iter().rev().take(check_count) {
            if candidates.iter().any(|c| c == hash) {
                continue;
            }
            if let Some(delta) = self.try_delta_against(hash, new_content)? {
                return Ok(Some(delta));
            }
        }

        Ok(None)
    }

    fn try_delta_against(
        &self,
        hash: &str,
        new_content: &str,
    ) -> Result<Option<crate::delta_encoder::DeltaResult>> {
        if let Some(cached) = self.store.get_cache_entry(hash)? {
            let hash_prefix = &hash[..hash.len().min(16)];
            if let Ok(Some(delta)) = self.delta_encoder.encode(&cached.data, new_content, hash_prefix) {
                if delta.delta_text.len() < new_content.len() {
                    return Ok(Some(delta));
                }
            }
        }
        Ok(None)
    }

    /// Check if `content` is already in the persistent cache (dedup lookup only).
    ///
    /// Returns `Some(inline_ref)` if cached AND the ref is still fresh,
    /// `None` if the content is not cached or the ref is stale.
    ///
    /// Unlike [`get_or_compress`](Self::get_or_compress), this method does
    /// not touch `accessed_at` until after the freshness check; otherwise
    /// every read would make itself "fresh."
    pub fn check_dedup(&self, content: &[u8]) -> Result<Option<String>> {
        Ok(self.check_dedup_with_meta(content)?.map(|hit| hit.inline_ref))
    }

    /// [`check_dedup`](Self::check_dedup) plus the metadata callers need for
    /// regret tracking: the full content hash and the entry's `accessed_at`
    /// from *before* this lookup bumped it. A hit whose previous access was
    /// seconds ago means the agent re-produced content it already had.
    pub fn check_dedup_with_meta(&self, content: &[u8]) -> Result<Option<DedupHit>> {
        let hash = Self::sha256_hex(content);
        // Read the pre-hit timestamp before the freshness probe's
        // record_ref_sent overwrites it.
        let prev_accessed_at = self.store.get_cache_entry_accessed_at(&hash).unwrap_or(None);
        let fresh = self.is_ref_fresh(&hash);
        if fresh {
            let hash_prefix = &hash[..16];
            self.record_ref_sent(&hash);
            Ok(Some(DedupHit {
                inline_ref: format!("§ref:{hash_prefix}§"),
                hash,
                prev_accessed_at,
            }))
        } else {
            // If the entry exists but is stale, don't return a dangling ref.
            // If it doesn't exist at all, same result: no dedup.
            Ok(None)
        }
    }

    /// Is `content` a line range of something the model already has?
    ///
    /// Exact-hash dedup misses the common case: the agent read a file in
    /// full, then re-reads lines 40-80 of it. Near-duplicate matching
    /// misses it too (a 40-line slice of a 500-line file has a Jaccard
    /// similarity near 0.1). This checks containment instead: the slice
    /// must appear verbatim, on line boundaries, in a recent cached
    /// original that is still fresh, and the same bytes must also appear
    /// in what was actually served for that original, so a ref never
    /// points at lines a lossy stage dropped. Touches the parent entry
    /// on a hit.
    pub fn check_slice_with_meta(&self, content: &[u8]) -> Result<Option<SliceHit>> {
        let slice = strip_one_trailing_newline(content);
        if slice.len() < MIN_SLICE_BYTES || !slice.contains(&b'\n') {
            return Ok(None);
        }
        for candidate in self.store.recent_originals(slice.len(), SLICE_CANDIDATES)? {
            let Some((start_line, end_line)) = find_line_range(&candidate.original, slice) else {
                continue;
            };
            if !self.is_ref_fresh(&candidate.hash) {
                continue;
            }
            let served: CompressedContent = match serde_json::from_str(&candidate.data_json) {
                Ok(c) => c,
                Err(_) => continue,
            };
            if !served.data.as_bytes().windows(slice.len()).any(|w| w == slice) {
                continue;
            }
            let total_lines = count_lines(&candidate.original);
            let hash_prefix = &candidate.hash[..16];
            let inline_ref = if start_line == 1 && end_line == total_lines {
                format!("§ref:{hash_prefix}§")
            } else {
                format!("§ref:{hash_prefix}:L{start_line}-{end_line}§")
            };
            let token_cost = 13 + (inline_ref.len().saturating_sub(22) as u32).div_ceil(3);
            self.record_ref_sent(&candidate.hash);
            let _ = self.store.record_cache_hit();
            return Ok(Some(SliceHit {
                inline_ref,
                hash: candidate.hash,
                start_line,
                end_line,
                total_lines,
                token_cost,
            }));
        }
        Ok(None)
    }

    /// Store a compressed result in the persistent cache, keyed by the
    /// SHA-256 hash of the original content.
    ///
    /// Also records the ref as sent at the current turn for compaction tracking.
    /// Persists `original_content` alongside `compressed` so that
    /// `sqz expand <prefix>` can recover the raw bytes for agents that
    /// cannot parse `§ref:…§` dedup tokens.
    pub fn store_compressed(
        &self,
        original_content: &[u8],
        compressed: &CompressedContent,
    ) -> Result<()> {
        let hash = Self::sha256_hex(original_content);
        self.store
            .save_cache_entry_with_original(&hash, compressed, Some(original_content))?;
        self.index_for_similarity(&hash, &compressed.data);
        self.record_ref_sent(&hash);
        Ok(())
    }

    /// Resolve a hex prefix (the 16-char tail of a `§ref:<prefix>§` token,
    /// or any longer hex string pasted by a user) to the cached content.
    ///
    /// Designed for the `sqz expand <prefix>` CLI: the agent sees a ref
    /// token it can't parse, runs `sqz expand a1b2c3d4e5f6g7h8`, and gets
    /// back the raw bytes that produced the ref. This is the
    /// user-visible escape hatch SquireNed asked for.
    ///
    /// Three outcomes:
    ///
    /// * `Ok(Some(Original))` — the original bytes were captured when
    ///   the entry was stored (new cache entries from v0.10.0+ always
    ///   capture the original). Write `bytes` to stdout.
    /// * `Ok(Some(CompressedOnly))` — the entry exists but its `original`
    ///   column is `NULL` (pre-migration data). We still return the
    ///   compressed form — always legible, always useful — and the CLI
    ///   surfaces a note that tells the user to re-run their original
    ///   command with `--no-cache` to capture a truly uncompressed copy.
    /// * `Ok(None)` — no entry matches. Usually means the ref was
    ///   truncated from a different sqz database (a different machine,
    ///   a wiped `~/.sqz/sessions.db`, etc.).
    /// * `Err(_)` — prefix was ambiguous or the DB is broken. The error
    ///   carries a user-readable message explaining what went wrong.
    ///
    /// Touches `accessed_at` on hit (consistent with `get_cache_entry`).
    pub fn expand_prefix(&self, prefix: &str) -> Result<Option<ExpandResult>> {
        let Some((hash, compressed_entry)) = self.store.get_cache_entry_by_prefix(prefix)? else {
            return Ok(None);
        };

        // Every expand is a recovery round trip the agent paid for —
        // count it so `sqz stats` can show how often refs get unwound.
        // Gap is unknown here (the prefix lookup already bumped
        // accessed_at), so record 0.0.
        let _ = self.store.log_regret("expand", "", &hash, 0.0);

        // Prefer the original bytes when we have them — that's the
        // whole point of this feature.
        if let Some(bytes) = self.store.get_cache_entry_original(&hash)? {
            return Ok(Some(ExpandResult::Original { hash, bytes }));
        }
        Ok(Some(ExpandResult::CompressedOnly {
            hash,
            compressed: compressed_entry.data,
        }))
    }

    /// Resolve a ref token in any of the forms an agent might paste:
    /// `a1b2c3d4`, `§ref:a1b2c3d4§`, `ref:a1b2c3d4`, or the ranged
    /// `§ref:a1b2c3d4:L40-80§`. A range narrows the returned bytes to
    /// those lines (1-based, inclusive).
    pub fn expand_ref(&self, raw: &str) -> Result<Option<ExpandResult>> {
        let (prefix, range) = parse_ref_token(raw);
        let Some(result) = self.expand_prefix(prefix)? else {
            return Ok(None);
        };
        let Some((start, end)) = range else {
            return Ok(Some(result));
        };
        Ok(Some(match result {
            ExpandResult::Original { hash, bytes } => ExpandResult::Original {
                hash,
                bytes: slice_lines(&bytes, start, end).to_vec(),
            },
            ExpandResult::CompressedOnly { hash, compressed } => ExpandResult::CompressedOnly {
                hash,
                compressed: String::from_utf8_lossy(slice_lines(compressed.as_bytes(), start, end))
                    .into_owned(),
            },
        }))
    }

    /// Invalidate the cache entry for `path` if its current content is known.
    ///
    /// Reads the file at `path`, computes its hash, and removes the matching
    /// entry from the store.  If the file does not exist the call is a no-op.
    pub fn invalidate(&self, path: &Path) -> Result<()> {
        if !path.exists() {
            return Ok(());
        }
        let bytes = std::fs::read(path)?;
        let hash = Self::sha256_hex(&bytes);
        self.store.delete_cache_entry(&hash)?;
        Ok(())
    }

    /// Evict least-recently-used entries until total cache size is at or below
    /// `max_size_bytes`.
    ///
    /// Returns the number of bytes freed.
    pub fn evict_lru(&self) -> Result<u64> {
        let entries = self.store.list_cache_entries_lru()?;

        // Compute current total size.
        let total: u64 = entries.iter().map(|(_, sz)| sz).sum();
        if total <= self.max_size_bytes {
            return Ok(0);
        }

        let mut freed: u64 = 0;
        let mut remaining = total;

        for (hash, size) in &entries {
            if remaining <= self.max_size_bytes {
                break;
            }
            self.store.delete_cache_entry(hash)?;
            freed += size;
            remaining -= size;
        }

        Ok(freed)
    }
}

fn strip_one_trailing_newline(bytes: &[u8]) -> &[u8] {
    match bytes {
        [rest @ .., b'\n'] => rest,
        _ => bytes,
    }
}

fn count_lines(bytes: &[u8]) -> usize {
    let body = strip_one_trailing_newline(bytes);
    if body.is_empty() {
        0
    } else {
        1 + body.iter().filter(|&&b| b == b'\n').count()
    }
}

/// Find `slice` in `original` starting at a line start and ending at a line
/// end. Returns 1-based inclusive line numbers of the first such occurrence.
fn find_line_range(original: &[u8], slice: &[u8]) -> Option<(usize, usize)> {
    if slice.is_empty() || slice.len() > original.len() {
        return None;
    }
    let mut from = 0;
    while from + slice.len() <= original.len() {
        let rel = original[from..].windows(slice.len()).position(|w| w == slice)?;
        let at = from + rel;
        let end = at + slice.len();
        let starts_line = at == 0 || original[at - 1] == b'\n';
        let ends_line = end == original.len() || original[end] == b'\n';
        if starts_line && ends_line {
            let start_line = 1 + original[..at].iter().filter(|&&b| b == b'\n').count();
            let end_line = start_line + slice.iter().filter(|&&b| b == b'\n').count();
            return Some((start_line, end_line));
        }
        from = at + 1;
    }
    None
}

/// Split a ref token into its hex prefix and optional `L<start>-<end>` range.
/// Accepts `a1b2`, `ref:a1b2`, `§ref:a1b2§`, `§ref:a1b2:L40-80§`.
pub fn parse_ref_token(raw: &str) -> (&str, Option<(usize, usize)>) {
    let body = raw
        .trim()
        .trim_start_matches('§')
        .trim_start_matches("ref:")
        .trim_end_matches('§')
        .trim();
    let Some((prefix, range)) = body.split_once(":L") else {
        return (body, None);
    };
    let parsed = range
        .split_once('-')
        .and_then(|(a, b)| Some((a.parse::<usize>().ok()?, b.parse::<usize>().ok()?)))
        .filter(|(a, b)| *a >= 1 && b >= a);
    match parsed {
        Some(r) => (prefix, Some(r)),
        None => (body, None),
    }
}

/// Bytes of lines `start..=end` (1-based) of `bytes`, newlines included.
fn slice_lines(bytes: &[u8], start: usize, end: usize) -> &[u8] {
    let mut line = 1;
    let mut begin = None;
    for (i, &b) in bytes.iter().enumerate() {
        if line == start && begin.is_none() {
            begin = Some(i);
        }
        if b == b'\n' {
            if line == end {
                return &bytes[begin.unwrap_or(i)..=i];
            }
            line += 1;
        }
    }
    match begin {
        Some(b) => &bytes[b..],
        None => &[],
    }
}

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

#[cfg(test)]
mod tests {
    use super::*;
    use crate::preset::{
        BudgetConfig, CollapseArraysConfig, CompressionConfig, CondenseConfig,
        CustomTransformsConfig, ModelConfig, PresetMeta, StripNullsConfig, TerseModeConfig,
        ToolSelectionConfig, TruncateStringsConfig,
    };
    use crate::session_store::SessionStore;

    fn in_memory_store() -> (SessionStore, tempfile::TempDir) {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("test.db");
        let store = SessionStore::open_or_create(&path).unwrap();
        (store, dir)
    }

    #[test]
    fn check_dedup_with_meta_reports_previous_access() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();
        let content = b"meta test content that is unique to this test";

        // Not cached yet: no hit, no metadata.
        assert!(cm.check_dedup_with_meta(content).unwrap().is_none());

        cm.get_or_compress(Path::new("meta.txt"), content, &pipeline)
            .unwrap();
        let before = chrono::Utc::now();
        let hit = cm
            .check_dedup_with_meta(content)
            .unwrap()
            .expect("expected dedup hit");
        assert!(hit.inline_ref.starts_with("§ref:"));
        assert_eq!(hit.hash, CacheManager::sha256_hex(content));
        // prev_accessed_at is the store-time touch, which happened before
        // this lookup — so it must exist and not be in the future.
        let prev = hit.prev_accessed_at.expect("expected prior access time");
        assert!(prev <= before + chrono::Duration::seconds(1));
    }

    fn source_file(lines: usize) -> Vec<u8> {
        let mut s = String::new();
        for i in 1..=lines {
            s.push_str(&format!("def handler_{i:03}(request):\n    return verify(request, scope='h{i:03}')\n"));
        }
        s.into_bytes()
    }

    fn lines_of(bytes: &[u8], start: usize, end: usize) -> Vec<u8> {
        slice_lines(bytes, start, end).to_vec()
    }

    #[test]
    fn slice_of_fresh_original_returns_line_range_ref() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();
        let file = source_file(60);
        cm.get_or_compress_lossless(Path::new("auth.py"), &file, &pipeline).unwrap();
        let hash = CacheManager::sha256_hex(&file);

        let slice = lines_of(&file, 41, 60);
        let hit = cm.check_slice_with_meta(&slice).unwrap().expect("slice hit");
        assert_eq!(hit.inline_ref, format!("§ref:{}:L41-60§", &hash[..16]));
        assert_eq!((hit.start_line, hit.end_line, hit.total_lines), (41, 60, 120));
        assert!(hit.token_cost > 13 && hit.token_cost < 25, "{}", hit.token_cost);

        // Same answer through the cache-aware compress path.
        match cm.get_or_compress_lossless(Path::new("auth.py"), &slice, &pipeline).unwrap() {
            CacheResult::Dedup { inline_ref, .. } => assert_eq!(inline_ref, hit.inline_ref),
            _ => panic!("expected slice dedup"),
        }
        // Without the trailing newline the ref is the same.
        let mut no_nl = slice.clone();
        no_nl.pop();
        assert_eq!(cm.check_slice_with_meta(&no_nl).unwrap().unwrap().inline_ref, hit.inline_ref);
    }

    #[test]
    fn slice_ref_requires_line_boundaries_and_size() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();
        let file = source_file(60);
        cm.get_or_compress_lossless(Path::new("auth.py"), &file, &pipeline).unwrap();

        // Starts mid-line: not a line range.
        let mid = &lines_of(&file, 10, 20)[4..];
        assert!(cm.check_slice_with_meta(mid).unwrap().is_none());
        // Single line, even a long one, is not worth a ref.
        let one = lines_of(&file, 10, 10);
        assert!(cm.check_slice_with_meta(&one).unwrap().is_none());
        // Two lines under 160 bytes: below the size floor.
        let two = lines_of(&file, 10, 11);
        assert!(two.len() < MIN_SLICE_BYTES);
        assert!(cm.check_slice_with_meta(&two).unwrap().is_none());
        // Content that is not in any original.
        let other = source_file(6).into_iter().map(|b| if b == b'h' { b'x' } else { b }).collect::<Vec<_>>();
        assert!(cm.check_slice_with_meta(&other).unwrap().is_none());
    }

    #[test]
    fn slice_ref_not_served_after_compaction() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();
        let file = source_file(60);
        cm.get_or_compress_lossless(Path::new("auth.py"), &file, &pipeline).unwrap();
        let slice = lines_of(&file, 1, 30);
        assert!(cm.check_slice_with_meta(&slice).unwrap().is_some());

        std::thread::sleep(std::time::Duration::from_millis(20));
        cm.notify_compaction();
        assert!(cm.check_slice_with_meta(&slice).unwrap().is_none(), "model no longer has the original");
    }

    #[test]
    fn slice_ref_only_for_lines_the_model_actually_saw() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let file = source_file(60);
        // Simulate a lossy first serving that dropped the second half.
        let served = String::from_utf8(lines_of(&file, 1, 60)).unwrap();
        let compressed = CompressedContent {
            data: served,
            tokens_compressed: 100,
            tokens_original: 200,
            stages_applied: vec!["entropy_truncate".into()],
            compression_ratio: 0.5,
            provenance: Default::default(),
            verify: None,
        };
        cm.store_compressed(&file, &compressed).unwrap();

        assert!(cm.check_slice_with_meta(&lines_of(&file, 10, 40)).unwrap().is_some(), "served lines are referenceable");
        assert!(cm.check_slice_with_meta(&lines_of(&file, 70, 100)).unwrap().is_none(), "dropped lines must be re-sent");
    }

    #[test]
    fn whole_file_minus_trailing_newline_gets_plain_ref() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();
        let file = source_file(30);
        cm.get_or_compress_lossless(Path::new("a.py"), &file, &pipeline).unwrap();
        let hash = CacheManager::sha256_hex(&file);
        let mut no_nl = file.clone();
        no_nl.pop();
        let hit = cm.check_slice_with_meta(&no_nl).unwrap().unwrap();
        assert_eq!(hit.inline_ref, format!("§ref:{}§", &hash[..16]));
    }

    #[test]
    fn expand_ref_parses_every_token_form_and_slices_lines() {
        assert_eq!(parse_ref_token("a1b2c3d4"), ("a1b2c3d4", None));
        assert_eq!(parse_ref_token("§ref:a1b2c3d4§"), ("a1b2c3d4", None));
        assert_eq!(parse_ref_token(" ref:a1b2c3d4 "), ("a1b2c3d4", None));
        assert_eq!(parse_ref_token("§ref:a1b2c3d4:L40-80§"), ("a1b2c3d4", Some((40, 80))));
        assert_eq!(parse_ref_token("a1b2c3d4:L7-7"), ("a1b2c3d4", Some((7, 7))));
        assert_eq!(parse_ref_token("a1b2c3d4:L9-3"), ("a1b2c3d4:L9-3", None), "inverted range is not a range");

        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();
        let file = source_file(20);
        cm.get_or_compress_lossless(Path::new("a.py"), &file, &pipeline).unwrap();
        let hash = CacheManager::sha256_hex(&file);

        match cm.expand_ref(&format!("§ref:{}:L5-8§", &hash[..16])).unwrap().unwrap() {
            ExpandResult::Original { bytes, .. } => assert_eq!(bytes, lines_of(&file, 5, 8)),
            _ => panic!("expected original bytes"),
        }
        match cm.expand_ref(&hash[..16]).unwrap().unwrap() {
            ExpandResult::Original { bytes, .. } => assert_eq!(bytes, file),
            _ => panic!("expected original bytes"),
        }
        // Range past the end returns what exists.
        match cm.expand_ref(&format!("{}:L39-99", &hash[..16])).unwrap().unwrap() {
            ExpandResult::Original { bytes, .. } => assert_eq!(bytes, lines_of(&file, 39, 40)),
            _ => panic!("expected original bytes"),
        }
    }

    #[test]
    fn line_helpers() {
        let text = b"a\nbb\nccc\ndddd\n";
        assert_eq!(count_lines(text), 4);
        assert_eq!(count_lines(b"a\nbb"), 2);
        assert_eq!(count_lines(b""), 0);
        assert_eq!(find_line_range(text, b"bb\nccc"), Some((2, 3)));
        assert_eq!(find_line_range(text, b"b\nccc"), None);
        assert_eq!(find_line_range(text, b"bb\ncc"), None);
        assert_eq!(find_line_range(text, b"a\nbb\nccc\ndddd"), Some((1, 4)));
        assert_eq!(slice_lines(text, 2, 3), b"bb\nccc\n");
        assert_eq!(slice_lines(text, 4, 9), b"dddd\n");
        assert_eq!(slice_lines(text, 5, 9), b"");
        assert_eq!(slice_lines(b"x\ny", 2, 2), b"y");
    }

    #[test]
    fn expand_prefix_logs_an_expand_regret() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();
        let content = b"expand regret test content";
        cm.get_or_compress(Path::new("regret.txt"), content, &pipeline)
            .unwrap();

        let hash = CacheManager::sha256_hex(content);
        let result = cm.expand_prefix(&hash[..16]).unwrap();
        assert!(result.is_some());

        let regret = cm.store.regret_stats().unwrap();
        assert_eq!(regret.expands, 1);
        assert_eq!(regret.reruns, 0);
    }

    fn test_preset() -> Preset {
        Preset {
            preset: PresetMeta {
                name: "test".into(),
                version: "1.0".into(),
                description: String::new(),
            },
            compression: CompressionConfig {
                stages: vec![],
                keep_fields: None,
                strip_fields: None,
                condense: Some(CondenseConfig {
                    enabled: true,
                    max_repeated_lines: 3,
                }),
                git_diff_fold: None,
                strip_nulls: Some(StripNullsConfig { enabled: true }),
                flatten: None,
                truncate_strings: Some(TruncateStringsConfig {
                    enabled: true,
                    max_length: 500,
                }),
                collapse_arrays: Some(CollapseArraysConfig {
                    enabled: true,
                    max_items: 5,
                    max_chars: 32_000,
                    summary_template: "... and {remaining} more items".into(),
                }),
                custom_transforms: Some(CustomTransformsConfig { enabled: true }),
            },
            tool_selection: ToolSelectionConfig {
                max_tools: 5,
                similarity_threshold: 0.7,
                default_tools: vec![],
            },
            budget: BudgetConfig {
                warning_threshold: 0.70,
                ceiling_threshold: 0.85,
                default_window_size: 200_000,
                agents: Default::default(),
            },
            terse_mode: TerseModeConfig {
                enabled: false,
                level: crate::preset::TerseLevel::Moderate,
            },
            model: ModelConfig {
                family: "anthropic".into(),
                primary: "claude-sonnet-4-20250514".into(),
                local: String::new(),
                complexity_threshold: 0.4,
                pricing: None,
            },
        }
    }

    fn make_pipeline() -> CompressionPipeline {
        CompressionPipeline::new(&test_preset())
    }

    #[test]
    fn first_read_is_miss() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();
        let content = b"hello world";
        let result = cm
            .get_or_compress(Path::new("file.txt"), content, &pipeline)
            .unwrap();
        assert!(matches!(result, CacheResult::Fresh { .. }));
    }

    /// Regression for https://github.com/ojuschugh1/sqz/issues/32 — the MCP
    /// file-read path (`get_or_compress_lossless`, used by sqz_read_file /
    /// sqz_grep / sqz_list_dir) returns content faithfully, while the general
    /// `compress` tool keeps using the lossy `get_or_compress`.
    #[test]
    fn get_or_compress_lossless_does_not_truncate_source() {
        let pipeline = make_pipeline();

        // >500 bytes of multi-segment non-JSON "source" that the lossy path
        // entropy-truncates by roughly half.
        let mut segs = Vec::new();
        // 24 segments: clears the benign-aware 1000-byte threshold.
        for i in 0..24 {
            if i % 2 == 0 {
                segs.push(format!("SEG{i} llllllllllllllllllllllllllllllllllllllll"));
            } else {
                segs.push(format!(
                    "SEG{i} the quick brown fox jumps over {i} lazy dogs by rivers"
                ));
            }
        }
        let content = segs.join("\n\n");
        assert!(content.len() > 1000);

        // Lossy path (the general `compress` tool) still truncates.
        let (s1, _d1) = in_memory_store();
        let lossy = CacheManager::new(s1, u64::MAX);
        match lossy
            .get_or_compress(Path::new("x"), content.as_bytes(), &pipeline)
            .unwrap()
        {
            CacheResult::Fresh { output } => {
                assert!(output.data.contains("omitted"), "lossy path should truncate");
            }
            _ => panic!("expected Fresh on a cold read"),
        }

        // Lossless path (file tools) keeps every segment.
        let (s2, _d2) = in_memory_store();
        let cm = CacheManager::new(s2, u64::MAX);
        match cm
            .get_or_compress_lossless(Path::new("src/lib.rs"), content.as_bytes(), &pipeline)
            .unwrap()
        {
            CacheResult::Fresh { output } => {
                assert!(
                    !output.data.contains("omitted"),
                    "file-read path truncated content: {}",
                    output.data
                );
                for i in 0..10 {
                    assert!(
                        output.data.contains(&format!("SEG{i}")),
                        "file-read path dropped SEG{i}"
                    );
                }
            }
            _ => panic!("expected CacheResult::Fresh on a cold read"),
        }
    }

    /// MinHash LSH finds a near-duplicate even after 12 unrelated entries
    /// have pushed the original out of the recent-10 LRU fallback window.
    #[test]
    fn lsh_finds_near_duplicate_beyond_lru_window() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();

        let base: String = (0..40)
            .map(|i| format!("stable log line number {i} with steady content here\n"))
            .collect();
        cm.get_or_compress(Path::new("base"), base.as_bytes(), &pipeline)
            .unwrap();

        // Bury it under 12 distinct entries.
        for n in 0..12 {
            let filler: String = (0..30)
                .map(|i| format!("filler doc {n} row {i} totally different subject matter\n"))
                .collect();
            cm.get_or_compress(Path::new("filler"), filler.as_bytes(), &pipeline)
                .unwrap();
        }

        // One line changed: a near-duplicate of the buried base.
        let variant = base.replace(
            "stable log line number 7 with steady content here",
            "stable log line number 7 with slightly altered content",
        );
        let result = cm
            .get_or_compress(Path::new("variant"), variant.as_bytes(), &pipeline)
            .unwrap();
        let kind = match &result {
            CacheResult::Dedup { .. } => "Dedup",
            CacheResult::Delta { .. } => "Delta",
            CacheResult::Fresh { .. } => "Fresh",
        };
        assert!(
            matches!(result, CacheResult::Delta { .. }),
            "near-duplicate beyond the LRU window must resolve via LSH, got {kind}"
        );
    }

    #[test]
    fn second_read_is_hit() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();
        let content = b"hello world";
        let path = Path::new("file.txt");

        // First read — miss
        cm.get_or_compress(path, content, &pipeline).unwrap();

        // Second read — hit
        let result = cm.get_or_compress(path, content, &pipeline).unwrap();
        match result {
            CacheResult::Dedup {
                inline_ref,
                token_cost,
            } => {
                assert!(inline_ref.starts_with("§ref:"));
                assert!(inline_ref.ends_with('§'));
                assert_eq!(token_cost, 13);
            }
            CacheResult::Fresh { .. } | CacheResult::Delta { .. } => panic!("expected cache hit"),
        }
    }

    #[test]
    fn different_content_is_miss() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();
        let path = Path::new("file.txt");

        cm.get_or_compress(path, b"content v1", &pipeline).unwrap();
        let result = cm
            .get_or_compress(path, b"content v2", &pipeline)
            .unwrap();
        assert!(matches!(result, CacheResult::Fresh { .. } | CacheResult::Delta { .. }));
    }

    #[test]
    fn evict_lru_frees_bytes_when_over_limit() {
        let (store, _dir) = in_memory_store();
        // Very small limit so eviction triggers immediately.
        let cm = CacheManager::new(store, 1);
        let pipeline = make_pipeline();
        let path = Path::new("f.txt");

        // Populate cache with a few entries.
        cm.get_or_compress(path, b"entry one", &pipeline).unwrap();
        cm.get_or_compress(path, b"entry two", &pipeline).unwrap();
        cm.get_or_compress(path, b"entry three", &pipeline).unwrap();

        let freed = cm.evict_lru().unwrap();
        assert!(freed > 0, "expected bytes to be freed");
    }

    #[test]
    fn evict_lru_no_op_when_under_limit() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();

        cm.get_or_compress(Path::new("f.txt"), b"data", &pipeline)
            .unwrap();

        let freed = cm.evict_lru().unwrap();
        assert_eq!(freed, 0);
    }

    #[test]
    fn invalidate_removes_entry() {
        let dir = tempfile::tempdir().unwrap();
        let file_path = dir.path().join("test.txt");
        std::fs::write(&file_path, b"some content").unwrap();

        let store_path = dir.path().join("store.db");
        let store = SessionStore::open_or_create(&store_path).unwrap();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();

        // Populate cache.
        let content = std::fs::read(&file_path).unwrap();
        cm.get_or_compress(&file_path, &content, &pipeline).unwrap();

        // Verify it's a hit.
        let hit = cm
            .get_or_compress(&file_path, &content, &pipeline)
            .unwrap();
        assert!(matches!(hit, CacheResult::Dedup { .. }));

        cm.invalidate(&file_path).unwrap();

        let miss = cm
            .get_or_compress(&file_path, &content, &pipeline)
            .unwrap();
        assert!(matches!(miss, CacheResult::Fresh { .. }));
    }

    #[test]
    fn invalidate_nonexistent_path_is_noop() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        // Should not error.
        cm.invalidate(Path::new("/nonexistent/path/file.txt"))
            .unwrap();
    }

    // ── Compaction / freshness tests ──────────────────────────────────────
    //
    // These tests used to exercise an in-memory turn counter. Freshness is
    // now computed from SQLite `accessed_at` timestamps so dedup works
    // across the shell-hook model (each hook invocation is a fresh
    // process). The tests below use wall-clock durations instead.

    #[test]
    fn stale_ref_returns_fresh_instead_of_dedup() {
        let (store, _dir) = in_memory_store();
        // Set max_ref_age to 0 — every ref goes stale immediately.
        let cm = CacheManager::with_ref_age_duration(store, u64::MAX, Duration::ZERO);
        let pipeline = make_pipeline();
        let content = b"hello world";
        let path = Path::new("file.txt");

        // First read — miss. accessed_at recorded.
        cm.get_or_compress(path, content, &pipeline).unwrap();

        // Second read — with TTL=0 the ref is already stale, should re-send.
        let result = cm.get_or_compress(path, content, &pipeline).unwrap();
        assert!(
            matches!(result, CacheResult::Fresh { .. }),
            "stale ref (TTL=0) should return Fresh, not Dedup"
        );
    }

    #[test]
    fn fresh_ref_returns_dedup() {
        let (store, _dir) = in_memory_store();
        // Generous TTL: one day. Refs stay fresh for the life of the test.
        let cm = CacheManager::with_ref_age_duration(
            store,
            u64::MAX,
            Duration::from_secs(86_400),
        );
        let pipeline = make_pipeline();
        let content = b"hello world";
        let path = Path::new("file.txt");

        cm.get_or_compress(path, content, &pipeline).unwrap();
        let result = cm.get_or_compress(path, content, &pipeline).unwrap();
        assert!(
            matches!(result, CacheResult::Dedup { .. }),
            "fresh ref should dedup"
        );
    }

    #[test]
    fn notify_compaction_invalidates_all_refs() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::with_ref_age_duration(
            store,
            u64::MAX,
            Duration::from_secs(86_400),
        );
        let pipeline = make_pipeline();
        let path = Path::new("file.txt");

        // Populate cache — every subsequent read is a dedup hit.
        cm.get_or_compress(path, b"content A", &pipeline).unwrap();
        cm.get_or_compress(path, b"content B", &pipeline).unwrap();
        assert!(matches!(
            cm.get_or_compress(path, b"content A", &pipeline).unwrap(),
            CacheResult::Dedup { .. }
        ));
        assert!(matches!(
            cm.get_or_compress(path, b"content B", &pipeline).unwrap(),
            CacheResult::Dedup { .. }
        ));

        // Simulate a context compaction. The compaction marker is set to
        // `now`; any cache entry whose accessed_at predates this moment is
        // treated as stale even though the TTL hasn't expired.
        // Sleep 10ms to ensure `now` is strictly after the last touch.
        std::thread::sleep(std::time::Duration::from_millis(10));
        cm.notify_compaction();

        // After compaction, refs predate the marker — re-send full content.
        assert!(matches!(
            cm.get_or_compress(path, b"content A", &pipeline).unwrap(),
            CacheResult::Fresh { .. }
        ));
        assert!(matches!(
            cm.get_or_compress(path, b"content B", &pipeline).unwrap(),
            CacheResult::Fresh { .. }
        ));
    }

    #[test]
    fn ref_refreshed_after_resend() {
        let (store, _dir) = in_memory_store();
        // Generous TTL so the "fresh after re-send" check is never a timing
        // race on a slow CI runner; staleness is forced via compaction below.
        let cm = CacheManager::with_ref_age_duration(store, u64::MAX, Duration::from_secs(86_400));
        let pipeline = make_pipeline();
        let content = b"hello world";
        let path = Path::new("file.txt");

        cm.get_or_compress(path, content, &pipeline).unwrap();
        // Force staleness deterministically: set the compaction marker strictly
        // after the seed's accessed_at, so the ref predates it (no TTL race).
        std::thread::sleep(std::time::Duration::from_millis(10));
        cm.notify_compaction();

        // Stale (predates the marker) — must re-send Fresh. The re-send bumps
        // accessed_at past the marker.
        let result = cm.get_or_compress(path, content, &pipeline).unwrap();
        assert!(matches!(result, CacheResult::Fresh { .. }));

        // Now fresh (accessed_at after the marker, within the generous TTL).
        let result = cm.get_or_compress(path, content, &pipeline).unwrap();
        assert!(
            matches!(result, CacheResult::Dedup { .. }),
            "ref should be fresh after re-send"
        );
    }

    #[test]
    fn check_dedup_returns_none_for_stale_ref() {
        let pipeline = make_pipeline();
        let content = b"test content";
        let path = Path::new("file.txt");

        // Fresh ref (generous TTL): check_dedup returns Some.
        let (store, _dir) = in_memory_store();
        let fresh =
            CacheManager::with_ref_age_duration(store, u64::MAX, Duration::from_secs(86_400));
        fresh.get_or_compress(path, content, &pipeline).unwrap();
        assert!(fresh.check_dedup(content).unwrap().is_some());

        // Stale ref (TTL=0): check_dedup returns None — deterministic, no sleep.
        let (store2, _dir2) = in_memory_store();
        let stale = CacheManager::with_ref_age_duration(store2, u64::MAX, Duration::ZERO);
        stale.get_or_compress(path, content, &pipeline).unwrap();
        assert!(
            stale.check_dedup(content).unwrap().is_none(),
            "stale ref should not be returned by check_dedup"
        );
    }

    #[test]
    fn pinned_ref_stays_fresh_past_ttl() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::with_ref_age_duration(
            store,
            u64::MAX,
            Duration::from_millis(10),
        );
        let pipeline = make_pipeline();
        let content = b"wiki: stable knowledge base content";
        let path = Path::new("wiki.md");

        // Seed the cache, then pin.
        cm.get_or_compress(path, content, &pipeline).unwrap();
        let hash = CacheManager::sha256_hex(content);
        assert!(cm.store.pin_cache_entry(&hash).unwrap());

        // Wait well past the 10ms TTL.
        std::thread::sleep(std::time::Duration::from_millis(40));

        // Pinned: still dedup, even though accessed_at is way past TTL.
        let result = cm.get_or_compress(path, content, &pipeline).unwrap();
        assert!(
            matches!(result, CacheResult::Dedup { .. }),
            "pinned entry should stay fresh past TTL"
        );
        assert!(cm.check_dedup(content).unwrap().is_some());
    }

    #[test]
    fn unpin_returns_to_ttl_behaviour() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::with_ref_age_duration(
            store,
            u64::MAX,
            Duration::from_millis(10),
        );
        let pipeline = make_pipeline();
        let content = b"transient session content";
        let path = Path::new("file.txt");

        cm.get_or_compress(path, content, &pipeline).unwrap();
        let hash = CacheManager::sha256_hex(content);
        cm.store.pin_cache_entry(&hash).unwrap();

        // Pin keeps it fresh past TTL.
        std::thread::sleep(std::time::Duration::from_millis(25));
        assert!(matches!(
            cm.get_or_compress(path, content, &pipeline).unwrap(),
            CacheResult::Dedup { .. }
        ));

        // Unpin and let TTL expire again.
        cm.store.unpin_cache_entry(&hash).unwrap();
        std::thread::sleep(std::time::Duration::from_millis(25));
        let result = cm.get_or_compress(path, content, &pipeline).unwrap();
        assert!(
            matches!(result, CacheResult::Fresh { .. }),
            "unpinned entry should obey TTL"
        );
    }

    #[test]
    fn pinned_ref_survives_compaction_marker() {
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::with_ref_age_duration(
            store,
            u64::MAX,
            Duration::from_secs(3600),
        );
        let pipeline = make_pipeline();
        let content = b"system prompt: shared across agents";
        let path = Path::new("system.md");

        cm.get_or_compress(path, content, &pipeline).unwrap();
        let hash = CacheManager::sha256_hex(content);
        cm.store.pin_cache_entry(&hash).unwrap();

        // Compaction would normally invalidate every pre-marker entry.
        std::thread::sleep(std::time::Duration::from_millis(10));
        cm.notify_compaction();

        // Pinned entry survives — wiki content is persistent across
        // compactions because it's re-injected into every agent.
        let result = cm.get_or_compress(path, content, &pipeline).unwrap();
        assert!(
            matches!(result, CacheResult::Dedup { .. }),
            "pinned entry should survive compaction"
        );
    }

    #[test]
    fn pin_visible_across_cachemanager_instances() {
        // Real-world scenario: agent A pins wiki content; agent B in a
        // separate process queries the same SQLite store and gets the
        // dedup hit because the pin lives in the shared store, not in
        // any single CacheManager's memory.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("shared.db");

        let store_a = SessionStore::open_or_create(&path).unwrap();
        let cm_a = CacheManager::with_ref_age_duration(
            store_a,
            u64::MAX,
            Duration::from_millis(5),
        );
        let pipeline = make_pipeline();
        let content = b"shared wiki content for multi-agent dedup";

        cm_a.get_or_compress(Path::new("wiki.md"), content, &pipeline)
            .unwrap();
        let hash = CacheManager::sha256_hex(content);
        cm_a.store.pin_cache_entry(&hash).unwrap();

        // Drop agent A. Wait past TTL. Spin up agent B against the same DB.
        drop(cm_a);
        std::thread::sleep(std::time::Duration::from_millis(20));

        let store_b = SessionStore::open_or_create(&path).unwrap();
        let cm_b = CacheManager::with_ref_age_duration(
            store_b,
            u64::MAX,
            Duration::from_millis(5),
        );
        let result = cm_b
            .get_or_compress(Path::new("wiki.md"), content, &pipeline)
            .unwrap();
        assert!(
            matches!(result, CacheResult::Dedup { .. }),
            "second agent should see the pin from the shared store"
        );
    }

    #[test]
    fn advance_turn_increments_counter() {
        // The counter is retained for context_evictor compatibility.
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        assert_eq!(cm.current_turn(), 0);
        cm.advance_turn();
        assert_eq!(cm.current_turn(), 1);
        cm.advance_turn();
        assert_eq!(cm.current_turn(), 2);
    }

    // ── Expand feature tests ────────────────────────────────────────────

    #[test]
    fn expand_returns_original_bytes_for_new_entry() {
        // New cache entries (written after the `original` column migration)
        // must always round-trip back to the exact bytes the agent sent.
        // This is the core guarantee `sqz expand` exists to provide.
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();
        let path = Path::new("x.txt");
        let content = b"hello\nworld\nthis is the original\n";

        // Seed the cache by running the content through get_or_compress.
        let first = cm.get_or_compress(path, content, &pipeline).unwrap();
        let CacheResult::Fresh { .. } = first else {
            panic!("first read should be a Fresh miss, got something else");
        };

        // Find the ref prefix by doing a second read (which dedups).
        let second = cm.get_or_compress(path, content, &pipeline).unwrap();
        let inline_ref = match second {
            CacheResult::Dedup { inline_ref, .. } => inline_ref,
            _ => panic!("second read should dedup"),
        };

        // Strip the `§ref:` / `§` wrappers to get the raw prefix.
        let prefix = inline_ref
            .strip_prefix("§ref:")
            .and_then(|s| s.strip_suffix('§'))
            .expect("unexpected ref format");

        let result = cm.expand_prefix(prefix).unwrap().expect("expand hit expected");
        match result {
            ExpandResult::Original { bytes, hash } => {
                assert_eq!(bytes, content, "expand must return exact original bytes");
                assert!(
                    hash.starts_with(prefix),
                    "returned full hash {hash} must start with prefix {prefix}"
                );
                assert_eq!(hash.len(), 64, "full SHA-256 must be 64 hex chars");
            }
            ExpandResult::CompressedOnly { .. } => {
                panic!("new cache entries must carry the original bytes");
            }
        }
    }

    #[test]
    fn expand_prefix_rejects_nonhex_input() {
        // Agents paste all sorts of things — ref tokens with the §'s still
        // attached, mixed-case hashes, stray whitespace. The store layer
        // normalises most of this but we also want to reject obvious
        // garbage without touching SQLite.
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        assert!(cm.expand_prefix("").unwrap().is_none(), "empty prefix is no-match");
        assert!(cm.expand_prefix("not a hex").unwrap().is_none());
        assert!(cm.expand_prefix("ABCDEF").unwrap().is_none(), "uppercase hex should not match (sqz emits lowercase)");
        assert!(cm.expand_prefix("g0g0g0").unwrap().is_none());
    }

    #[test]
    fn expand_prefix_returns_none_for_unknown_prefix() {
        // Very common in practice: agent pastes a ref from a different
        // machine, a different sqz install, or after a cache wipe.
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();
        // Seed one entry.
        let _ = cm
            .get_or_compress(
                Path::new("y.txt"),
                b"some cached content that won't match the prefix below",
                &pipeline,
            )
            .unwrap();
        assert!(
            cm.expand_prefix("0000000000000000").unwrap().is_none(),
            "prefix that matches nothing must return None, not error"
        );
    }

    #[test]
    fn expand_prefix_errors_on_ambiguous_match() {
        // Two entries whose hashes both start with the same hex prefix
        // must be detected and reported. The caller is expected to
        // surface "use a longer prefix" — we don't want to silently
        // pick one.
        let (store, _dir) = in_memory_store();
        // Hand-craft two entries with prefixes that collide on "ab".
        let compressed = crate::types::CompressedContent {
            data: "compressed-a".to_string(),
            tokens_compressed: 1,
            tokens_original: 10,
            stages_applied: vec![],
            compression_ratio: 0.1,
            provenance: Default::default(),
            verify: None,
        };
        store
            .save_cache_entry_with_original(
                &format!("ab{}", "0".repeat(62)),
                &compressed,
                Some(b"content a"),
            )
            .unwrap();
        store
            .save_cache_entry_with_original(
                &format!("ab{}", "1".repeat(62)),
                &compressed,
                Some(b"content b"),
            )
            .unwrap();

        let cm = CacheManager::new(store, u64::MAX);
        // Unambiguous 3-char prefixes resolve:
        let ok_a = cm.expand_prefix(&format!("ab{}", "0")).unwrap().unwrap();
        assert!(matches!(ok_a, ExpandResult::Original { .. }));
        // Ambiguous 2-char prefix errors out.
        let err = cm.expand_prefix("ab").unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("multiple entries") || msg.contains("longer prefix"),
            "ambiguity error should mention multiple matches / longer prefix, got: {msg}"
        );
    }

    #[test]
    fn expand_returns_compressed_only_for_pre_migration_entry() {
        // Entries written via `save_cache_entry` (without the `_with_original`
        // suffix) leave the `original` column NULL. Expand must still return
        // something useful — the compressed blob — with a note (handled by
        // the CLI). Here we assert the engine returns the right variant.
        let (store, _dir) = in_memory_store();
        let compressed = crate::types::CompressedContent {
            data: "the compressed version".to_string(),
            tokens_compressed: 4,
            tokens_original: 100,
            stages_applied: vec!["condense".to_string()],
            compression_ratio: 0.04,
            provenance: Default::default(),
            verify: None,
        };
        // Deliberately use the legacy save_cache_entry (no original).
        let hash = format!("deadbeef{}", "0".repeat(56));
        store.save_cache_entry(&hash, &compressed).unwrap();

        let cm = CacheManager::new(store, u64::MAX);
        let result = cm
            .expand_prefix("deadbeef")
            .unwrap()
            .expect("deadbeef prefix should match");
        match result {
            ExpandResult::CompressedOnly { compressed, hash: h } => {
                assert_eq!(compressed, "the compressed version");
                assert_eq!(h, hash);
            }
            ExpandResult::Original { .. } => {
                panic!("legacy entry without original bytes should return CompressedOnly");
            }
        }
    }

    #[test]
    fn expand_preserves_non_utf8_original_bytes() {
        // Shell output routinely contains non-UTF-8 sequences (terminal
        // control codes, binary blobs from `cat` on a wrong file, etc.).
        // Expand must round-trip byte-for-byte — not through a UTF-8
        // lossy conversion — or agents auditing the cache will see
        // spurious `\uFFFD` REPLACEMENT CHARACTER bytes.
        let (store, _dir) = in_memory_store();
        let cm = CacheManager::new(store, u64::MAX);
        let pipeline = make_pipeline();

        // Mix of valid UTF-8 and isolated 0xFF bytes.
        let content: Vec<u8> = vec![
            b'h', b'e', b'l', b'l', b'o', b'\n',
            0xFF, 0xFE, 0xFD,
            b'\n',
        ];
        let path = Path::new("bin.dat");
        let _ = cm.get_or_compress(path, &content, &pipeline).unwrap();
        // Round-trip via expand_prefix.
        let hash = CacheManager::sha256_hex(&content);
        let result = cm.expand_prefix(&hash[..16]).unwrap().unwrap();
        match result {
            ExpandResult::Original { bytes, .. } => {
                assert_eq!(bytes, content, "non-UTF-8 bytes must round-trip exactly");
            }
            ExpandResult::CompressedOnly { .. } => {
                panic!("fresh entry should have original bytes");
            }
        }
    }

    #[test]
    fn dedup_survives_cache_manager_restart() {
        // Regression for the April 18 bug: the turn counter was in-memory
        // only, so every new sqz process saw an empty ref tracker and the
        // dedup feature silently produced Fresh results forever. With
        // accessed_at-based freshness, a fresh CacheManager reading the
        // same SQLite store picks up the dedup correctly.
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("cache.db");
        let pipeline = make_pipeline();
        let content = b"a substantial chunk of content to dedup";
        let path = Path::new("x.txt");

        // First "process": populate cache.
        {
            let store = SessionStore::open_or_create(&db_path).unwrap();
            let cm = CacheManager::with_ref_age_duration(
                store,
                u64::MAX,
                Duration::from_secs(3600),
            );
            let first = cm.get_or_compress(path, content, &pipeline).unwrap();
            assert!(matches!(first, CacheResult::Fresh { .. }));
        }

        // Second "process": new CacheManager, same DB. Dedup must fire.
        {
            let store = SessionStore::open_or_create(&db_path).unwrap();
            let cm = CacheManager::with_ref_age_duration(
                store,
                u64::MAX,
                Duration::from_secs(3600),
            );
            let second = cm.get_or_compress(path, content, &pipeline).unwrap();
            assert!(
                matches!(second, CacheResult::Dedup { .. }),
                "second-process read must dedup — this was broken before the April 18 fix"
            );
        }
    }

    #[test]
    fn compaction_from_one_process_invalidates_refs_in_another() {
        // Regression for the PreCompact hook wiring: the host harness
        // (e.g. Claude Code) runs `sqz hook precompact` in a short-lived
        // process to signal auto-compaction. The actual dedup serving runs
        // in a DIFFERENT sqz process (the shell hook). notify_compaction
        // must persist through SQLite so the second process sees it.
        //
        // Before the fix, compaction_marker was Cell<Option<DateTime>>
        // in memory only — the precompact process set it, exited, the
        // state was lost. Next shell-hook process started with a clean
        // marker, served stale refs to the agent, and the agent saw a
        // §ref:HASH§ pointing at content no longer in its context.
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("cache.db");
        let pipeline = make_pipeline();
        let content = b"content that needs stale-marking after compaction";
        let path = Path::new("file.txt");
        let ttl = Duration::from_secs(3600);

        // Process A: populate the cache so the content is dedup-eligible.
        {
            let store = SessionStore::open_or_create(&db_path).unwrap();
            let cm = CacheManager::with_ref_age_duration(store, u64::MAX, ttl);
            cm.get_or_compress(path, content, &pipeline).unwrap();
        }
        // Sleep so the compaction marker is strictly after the touch.
        std::thread::sleep(Duration::from_millis(10));

        // Process B: simulates `sqz hook precompact`. Just calls
        // notify_compaction and exits. No reads.
        {
            let store = SessionStore::open_or_create(&db_path).unwrap();
            let cm = CacheManager::with_ref_age_duration(store, u64::MAX, ttl);
            cm.notify_compaction();
        }

        // Process C: simulates the next `sqz compress` shell-hook call.
        // Reads the same content. MUST re-send Fresh, not return a ref
        // the agent can no longer resolve.
        {
            let store = SessionStore::open_or_create(&db_path).unwrap();
            let cm = CacheManager::with_ref_age_duration(store, u64::MAX, ttl);
            let result = cm.get_or_compress(path, content, &pipeline).unwrap();
            assert!(
                matches!(result, CacheResult::Fresh { .. }),
                "post-compaction read from a fresh process must re-send Fresh; \
                 returning Dedup would be a dangling-ref bug"
            );
        }
    }

    use proptest::prelude::*;

    // ── Property 8: Cache deduplication ──────────────────────────────────────
    // **Validates: Requirements 8.1, 8.2, 18.1, 18.2**
    //
    // For any file content, reading the file twice through the CacheManager
    // (with no content change between reads) SHALL return a cache hit on the
    // second read with a reference token of approximately 13 tokens.

    proptest! {
        /// **Validates: Requirements 8.1, 8.2, 18.1, 18.2**
        ///
        /// For any file content, the second read through CacheManager SHALL be
        /// a cache hit with tokens == 13.
        #[test]
        fn prop_cache_deduplication(
            content in proptest::collection::vec(any::<u8>(), 1..=1000usize),
        ) {
            let (store, _dir) = in_memory_store();
            let cm = CacheManager::new(store, u64::MAX);
            let pipeline = make_pipeline();
            let path = Path::new("file.txt");

            // First read — must be a miss.
            let first = cm.get_or_compress(path, &content, &pipeline).unwrap();
            prop_assert!(
                matches!(first, CacheResult::Fresh { .. }),
                "first read should be a cache miss"
            );

            let second = cm.get_or_compress(path, &content, &pipeline).unwrap();
            match second {
                CacheResult::Dedup { inline_ref, token_cost } => {
                    prop_assert_eq!(
                        token_cost, 13,
                        "cache hit should report ~13 reference tokens"
                    );
                    prop_assert!(
                        inline_ref.starts_with("§ref:"),
                        "reference token should start with §ref:"
                    );
                    prop_assert!(
                        inline_ref.ends_with('§'),
                        "reference token should end with §"
                    );
                }
                CacheResult::Fresh { .. } | CacheResult::Delta { .. } => {
                    prop_assert!(false, "second read should be a cache hit, not a miss");
                }
            }
        }
    }

    // ── Property 9: Cache invalidation on content change ─────────────────────
    // **Validates: Requirements 8.3, 18.3**
    //
    // For any cached file, if the file content changes (producing a different
    // SHA-256 hash), the CacheManager SHALL treat the next read as a cache miss
    // and re-compress the updated content.

    proptest! {
        /// **Validates: Requirements 8.3, 18.3**
        ///
        /// For any two distinct byte sequences, the first read of each is a
        /// cache miss — content change always triggers re-compression.
        #[test]
        fn prop_cache_invalidation_on_content_change(
            content_a in proptest::collection::vec(any::<u8>(), 1..=500usize),
            content_b in proptest::collection::vec(any::<u8>(), 1..=500usize),
        ) {
            // Only meaningful when the two contents differ (different hashes).
            prop_assume!(content_a != content_b);

            let (store, _dir) = in_memory_store();
            let cm = CacheManager::new(store, u64::MAX);
            let pipeline = make_pipeline();
            let path = Path::new("file.txt");

            // Cache content_a.
            let r1 = cm.get_or_compress(path, &content_a, &pipeline).unwrap();
            prop_assert!(
                matches!(r1, CacheResult::Fresh { .. }),
                "first read of content_a should be a miss"
            );

            let r2 = cm.get_or_compress(path, &content_a, &pipeline).unwrap();
            prop_assert!(
                matches!(r2, CacheResult::Dedup { .. }),
                "second read of content_a should be a hit"
            );

            let r3 = cm.get_or_compress(path, &content_b, &pipeline).unwrap();
            prop_assert!(
                matches!(r3, CacheResult::Fresh { .. } | CacheResult::Delta { .. }),
                "read with changed content should be a cache miss or delta"
            );
        }
    }

    // ── Property 10: Cache LRU eviction ──────────────────────────────────────
    // **Validates: Requirements 8.5**
    //
    // For any cache state where total size exceeds the configured maximum, the
    // CacheManager SHALL evict entries in LRU order until total size is at or
    // below the limit.

    proptest! {
        /// **Validates: Requirements 8.5**
        ///
        /// After evict_lru, the total remaining cache size SHALL be at or below
        /// max_size_bytes.
        #[test]
        fn prop_cache_lru_eviction(
            // Generate 2-8 distinct content entries.
            entries in proptest::collection::vec(
                proptest::collection::vec(any::<u8>(), 10..=200usize),
                2..=8usize,
            ),
        ) {
            // Deduplicate entries so each has a unique hash.
            let mut unique_entries: Vec<Vec<u8>> = Vec::new();
            for e in &entries {
                if !unique_entries.contains(e) {
                    unique_entries.push(e.clone());
                }
            }
            prop_assume!(unique_entries.len() >= 2);

            let (store, _dir) = in_memory_store();
            // Use a very small limit (1 byte) to guarantee eviction is needed.
            let cm = CacheManager::new(store, 1);
            let pipeline = make_pipeline();
            let path = Path::new("f.txt");

            // Populate the cache.
            for entry in &unique_entries {
                cm.get_or_compress(path, entry, &pipeline).unwrap();
            }

            // Evict LRU entries.
            let freed = cm.evict_lru().unwrap();

            // Bytes freed must be > 0 since total > 1 byte.
            prop_assert!(freed > 0, "evict_lru should free bytes when over limit");

            // After eviction, total remaining size must be <= max_size_bytes (1).
            // We verify by checking that evict_lru now returns 0 (nothing left to evict).
            let freed_again = cm.evict_lru().unwrap();
            prop_assert_eq!(
                freed_again, 0,
                "second evict_lru call should free 0 bytes (already at or below limit)"
            );
        }
    }

    // ── Property 34: Cache persistence across sessions ────────────────────────
    // **Validates: Requirements 18.4**
    //
    // For any set of cache entries saved to the SessionStore, reloading the
    // store (opening the same database file) SHALL produce the same cache
    // entries, and a subsequent read with the same content hash SHALL return a
    // cache hit.

    proptest! {
        /// **Validates: Requirements 18.4**
        ///
        /// Cache entries written in one CacheManager instance SHALL survive
        /// a store close/reopen. With the wall-clock freshness model
        /// (introduced April 18 2026), a subsequent CacheManager reading
        /// the same database SHALL see the entry as fresh (within TTL) and
        /// return a Dedup hit on the very first read — this is the whole
        /// point of the cross-process fix. Previous behavior (Fresh on
        /// first read after restart) was a bug that silently disabled the
        /// dedup feature in production.
        #[test]
        fn prop_cache_persistence_across_sessions(
            content in proptest::collection::vec(any::<u8>(), 1..=500usize),
        ) {
            use crate::session_store::SessionStore;

            let dir = tempfile::tempdir().unwrap();
            let db_path = dir.path().join("cache.db");
            let path = Path::new("file.txt");

            // Session 1: populate the cache.
            {
                let store = SessionStore::open_or_create(&db_path).unwrap();
                // Explicit long TTL so tests don't race with wall-clock drift.
                let cm = CacheManager::with_ref_age_duration(
                    store,
                    u64::MAX,
                    Duration::from_secs(3600),
                );
                let pipeline = make_pipeline();

                let r = cm.get_or_compress(path, &content, &pipeline).unwrap();
                prop_assert!(
                    matches!(r, CacheResult::Fresh { .. }),
                    "first-ever read should be a miss"
                );
            }

            // Session 2: reopen the same database file.
            {
                let store = SessionStore::open_or_create(&db_path).unwrap();
                let cm = CacheManager::with_ref_age_duration(
                    store,
                    u64::MAX,
                    Duration::from_secs(3600),
                );
                let pipeline = make_pipeline();

                // First read in the new session MUST dedup. The entry was
                // just written (within TTL), so the wall-clock freshness
                // check finds it fresh. This is what makes sqz's dedup
                // actually work across shell-hook invocations.
                let r = cm.get_or_compress(path, &content, &pipeline).unwrap();
                match r {
                    CacheResult::Dedup { token_cost, .. } => {
                        prop_assert_eq!(
                            token_cost, 13,
                            "first read after restart must be a 13-token dedup ref"
                        );
                    }
                    CacheResult::Fresh { .. } | CacheResult::Delta { .. } => {
                        prop_assert!(
                            false,
                            "first read after restart must dedup — this was the \
                             April 18 bug and its fix is the whole reason this \
                             test exists"
                        );
                    }
                }
            }
        }
    }
}