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,
    },
}

/// 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`] 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>>>,
}

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`] 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),
        }
    }

    /// 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> {
        self.get_or_compress_inner(path, content, pipeline, false)
    }

    /// 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`] for aggressive compression. See issue #32.
    pub fn get_or_compress_lossless(
        &self,
        path: &Path,
        content: &[u8],
        pipeline: &CompressionPipeline,
    ) -> Result<CacheResult> {
        self.get_or_compress_inner(path, content, pipeline, true)
    }

    fn get_or_compress_inner(
        &self,
        _path: &Path,
        content: &[u8],
        pipeline: &CompressionPipeline,
        lossless: bool,
    ) -> 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 ctx = SessionContext {
                    session_id: "cache".to_string(),
                };
                let preset = Preset::default();
                let compressed = if lossless {
                    pipeline.compress_lossless(&text, &ctx, &preset)?
                } else {
                    pipeline.compress(&text, &ctx, &preset)?
                };
                // Record that we re-sent this content
                self.record_ref_sent(&hash);
                return Ok(CacheResult::Fresh { output: compressed });
            }
        }

        // 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 ctx = SessionContext {
                session_id: "cache".to_string(),
            };
            let preset = Preset::default();
            let compressed = if lossless {
                pipeline.compress_lossless(&text, &ctx, &preset)?
            } else {
                pipeline.compress(&text, &ctx, &preset)?
            };
            // Persist the raw bytes so `sqz expand <prefix>` can round-trip.
            self.store
                .save_cache_entry_with_original(&hash, &compressed, Some(content))?;
            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 ctx = SessionContext {
            session_id: "cache".to_string(),
        };
        let preset = Preset::default();
        let compressed = if lossless {
            pipeline.compress_lossless(&text, &ctx, &preset)?
        } else {
            pipeline.compress(&text, &ctx, &preset)?
        };
        self.store
            .save_cache_entry_with_original(&hash, &compressed, Some(content))?;
        // 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 recent cache entries.
    /// Returns Some(DeltaResult) if a near-duplicate was found.
    fn try_delta_encode(
        &self,
        new_content: &str,
    ) -> Result<Option<crate::delta_encoder::DeltaResult>> {
        let entries = self.store.list_cache_entries_lru()?;

        // Check the most recent entries (up to 10) for near-duplicates
        let check_count = entries.len().min(10);
        for (hash, _) in entries.iter().rev().take(check_count) {
            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)
                {
                    // Only use delta if it's actually smaller than the full content
                    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`], 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>> {
        let hash = Self::sha256_hex(content);
        // Probe existence without touching accessed_at.
        let fresh = self.is_ref_fresh(&hash);
        if fresh {
            let hash_prefix = &hash[..16];
            self.record_ref_sent(&hash);
            Ok(Some(format!("§ref:{hash_prefix}§")))
        } 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)
        }
    }

    /// 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.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);
        };

        // 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,
        }))
    }

    /// 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)
    }
}

// ── 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)
    }

    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,
                    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();
        for i in 0..10 {
            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() > 500);

        // 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"),
        }
    }

    #[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"
                        );
                    }
                }
            }
        }
    }
}