verdant_runtime/

store.rs

//! Content-addressed payload store. The cache key is the blake3 hash of
//! the canonical *input* bytes for a tool call (not the output); the value
//! is the exact output payload bytes that the MCP tool fed back to the
//! model on its original execution. The store is append-only for M1; M2
//! will add eviction.
//!
//! Layout on disk:
//!
//! ```text
//! <root>/
//!   ab/                       # first two hex chars of the key
//!     .lock                   # 0-byte shard lock (see SHARD_LOCK)
//!     ab12cd...ef.payload     # raw bytes
//!     ab12cd...ef.meta.json   # invalidation metadata (raw_hash, size, kind)
//! ```
//!
//! Keys are 64-char lowercase hex strings; we shard by the first two chars
//! so a single project does not produce a directory with 100k+ entries.

use serde::{Deserialize, Serialize};
use std::fs;
use std::hash::RandomState;
use std::hash::{BuildHasher, Hasher};
use std::io::{self, Read as _, Write as _};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::OnceLock;
use std::time::{SystemTime, UNIX_EPOCH};

#[derive(Debug, thiserror::Error)]
pub enum StoreError {
    #[error("io: {0}")]
    Io(#[from] io::Error),
    #[error("malformed key: {0}")]
    BadKey(String),
    #[error("metadata decode failed: {0}")]
    Meta(#[from] serde_json::Error),
    #[error(
        "integrity check failed for key {key}: stored payload hash {actual} != expected {expected}"
    )]
    Integrity {
        key: String,
        expected: String,
        actual: String,
    },
}

/// Lowercase hex blake3 digest. Wrapping in a newtype so we can't confuse
/// a payload-hash with the cache key (which hashes inputs, not outputs).
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct Key(pub String);

impl Key {
    pub fn from_bytes(bytes: &[u8]) -> Self {
        Key(blake3::hash(bytes).to_hex().to_string())
    }

    pub fn as_str(&self) -> &str {
        &self.0
    }

    fn validate(&self) -> Result<(), StoreError> {
        if self.0.len() != 64 || !self.0.chars().all(|c| c.is_ascii_hexdigit()) {
            return Err(StoreError::BadKey(self.0.clone()));
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PayloadMeta {
    /// blake3 of the raw payload bytes; recorded so reads can detect a
    /// torn write or external file corruption rather than silently
    /// returning bad bytes.
    pub payload_hash: String,
    /// Length of the payload bytes.
    pub bytes: u64,
    /// Tool kind tag, free-form ("read", "bash", etc.) — used for
    /// telemetry and for tool-specific revalidation logic that the
    /// runtime layer applies on green hits.
    pub tool_kind: String,
    /// File dependencies of this entry. Each entry pairs a path with the
    /// blake3 of that file at the time the cache entry was written; on
    /// every green-hit lookup the file is re-blake3'd and the entry is
    /// invalidated on mismatch. Stored on disk so a fresh process can
    /// restore the cache state without depending on an in-memory
    /// registry that does not survive restart.
    #[serde(default)]
    pub file_roots: Vec<FileRootSerde>,
    /// Upstream cache-key dependencies. For LlmCall entries this is the
    /// set of tool-call cache keys whose results appeared in the
    /// prompt's `tool_result` blocks; when one of those tool entries is
    /// invalidated by a file edit, every LlmCall whose upstream set
    /// contains that key is invalidated too. Tool-call entries
    /// typically leave this empty (their dependencies are encoded in
    /// `file_roots`); future M3+ extensions may use it for nested
    /// composite nodes.
    #[serde(default)]
    pub upstream_keys: Vec<String>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FileRootSerde {
    pub path: String,
    pub expected_hash: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Payload {
    pub bytes: Vec<u8>,
    pub meta: PayloadMeta,
}

/// Storage backend trait. Concrete impls: `FileStore` (local
/// content-addressed disk), and (M4 step 7+) `RemoteStore` over HTTP.
/// `LiveCache` owns one `Box<dyn Store>` and routes every
/// content-addressed read/write through this trait so the same cache
/// state machine works against either backend without conditional
/// compilation in the runtime layer.
pub trait Store: Send + Sync {
    fn persist_with_upstreams(
        &self,
        key: &Key,
        bytes: &[u8],
        tool_kind: &str,
        file_roots: Vec<FileRootSerde>,
        upstream_keys: Vec<String>,
    ) -> Result<(), StoreError>;

    fn lookup(&self, key: &Key) -> Result<Option<Payload>, StoreError>;

    fn remove(&self, key: &Key) -> Result<(), StoreError>;

    fn total_bytes(&self) -> Result<u64, StoreError>;

    fn evict_to_cap(&self, cap_bytes: u64) -> Result<usize, StoreError>;

    fn iter_meta(&self) -> Result<Vec<(Key, PayloadMeta)>, StoreError>;

    fn contains(&self, key: &Key) -> bool;

    fn persist(
        &self,
        key: &Key,
        bytes: &[u8],
        tool_kind: &str,
        file_roots: Vec<FileRootSerde>,
    ) -> Result<(), StoreError> {
        self.persist_with_upstreams(key, bytes, tool_kind, file_roots, Vec::new())
    }
}

#[derive(Debug)]
pub struct FileStore {
    root: PathBuf,
}

impl FileStore {
    pub fn open(root: impl Into<PathBuf>) -> Result<Self, StoreError> {
        let root = root.into();
        fs::create_dir_all(&root)?;
        Ok(Self { root })
    }

    pub fn root(&self) -> &Path {
        &self.root
    }

    fn shard_dir(&self, key: &Key) -> PathBuf {
        self.root.join(&key.0[..2])
    }

    fn payload_path(&self, key: &Key) -> PathBuf {
        self.shard_dir(key).join(format!("{}.payload", key.0))
    }

    fn meta_path(&self, key: &Key) -> PathBuf {
        self.shard_dir(key).join(format!("{}.meta.json", key.0))
    }

    /// Write a payload. Uses tempfile + rename so a crash mid-write leaves
    /// the store in a consistent state (either the entry is fully there
    /// or it is not), which keeps `lookup` from ever observing torn data.
    pub fn persist(
        &self,
        key: &Key,
        bytes: &[u8],
        tool_kind: &str,
        file_roots: Vec<FileRootSerde>,
    ) -> Result<(), StoreError> {
        self.persist_with_upstreams(key, bytes, tool_kind, file_roots, Vec::new())
    }

    /// Persist a payload that depends on previously-cached upstream
    /// entries. The LlmCall path uses this to record which tool-call
    /// keys it consumed, so downstream invalidation can walk the
    /// dependency edge and drop dependent entries when a tool key is
    /// marked dirty.
    pub fn persist_with_upstreams(
        &self,
        key: &Key,
        bytes: &[u8],
        tool_kind: &str,
        file_roots: Vec<FileRootSerde>,
        upstream_keys: Vec<String>,
    ) -> Result<(), StoreError> {
        key.validate()?;
        let shard = self.shard_dir(key);
        fs::create_dir_all(&shard)?;

        let payload_hash = blake3::hash(bytes).to_hex().to_string();
        let meta = PayloadMeta {
            payload_hash,
            bytes: bytes.len() as u64,
            tool_kind: tool_kind.to_string(),
            file_roots,
            upstream_keys,
        };
        let meta_bytes = serde_json::to_vec(&meta)?;

        let payload_tmp = TempGuard::write_in(&shard, bytes)?;
        let meta_tmp = TempGuard::write_in(&shard, &meta_bytes)?;
        let _lock = lock_shard(&shard, LockMode::Exclusive)?
            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "shard directory vanished"))?;
        payload_tmp.persist(&self.payload_path(key))?;
        meta_tmp.persist(&self.meta_path(key))?;
        Ok(())
    }

    /// Delete the payload + meta for a key. Used by `LiveCache` when an
    /// upstream invalidation makes a cached entry definitely-stale; the
    /// caller has already removed the registry entry, and this drops
    /// the bytes from disk so a future rehydration does not resurrect
    /// the entry.
    pub fn remove(&self, key: &Key) -> Result<(), StoreError> {
        key.validate()?;
        let Some(_lock) = lock_shard(&self.shard_dir(key), LockMode::Exclusive)? else {
            return Ok(());
        };
        remove_if_present(&self.payload_path(key))?;
        remove_if_present(&self.meta_path(key))?;
        Ok(())
    }

    /// Total bytes occupied by the store, summed across every payload
    /// and meta file under the root. Used by `evict_to_cap` and by
    /// operator-facing stats; cheap on small stores, linear-walk on
    /// large ones (we accept the cost because eviction is a periodic
    /// operation, not a hot path).
    pub fn total_bytes(&self) -> Result<u64, StoreError> {
        let mut total: u64 = 0;
        let entries = match fs::read_dir(&self.root) {
            Ok(e) => e,
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(0),
            Err(e) => return Err(e.into()),
        };
        for shard in entries.flatten() {
            let shard_path = shard.path();
            if !shard_path.is_dir() {
                continue;
            }
            for entry in fs::read_dir(&shard_path)?.flatten() {
                #[allow(clippy::collapsible_if)]
                if let Ok(md) = entry.metadata() {
                    if md.is_file() {
                        total = total.saturating_add(md.len());
                    }
                }
            }
        }
        Ok(total)
    }

    /// Evict oldest entries until `total_bytes() <= cap_bytes`. Order
    /// is by payload-file `mtime` ascending so the least-recently-
    /// modified entry is removed first; on filesystems where reads
    /// update atime but not mtime this is a true write-order eviction
    /// (entries that have never been re-persisted go first), which is
    /// the right default for an append-only cache because a hot key
    /// gets re-persisted on every backend fall-through and a cold key
    /// does not. Returns the number of entries dropped.
    ///
    /// The function intentionally does not touch the in-memory
    /// `LiveCache` registry; the caller is expected to either call
    /// this at process startup (before `LiveCache::new` rehydrates)
    /// or to recreate the cache afterwards.
    pub fn evict_to_cap(&self, cap_bytes: u64) -> Result<usize, StoreError> {
        let mut current = self.total_bytes()?;
        if current <= cap_bytes {
            return Ok(0);
        }
        let entries = self.entry_stats()?;

        let mut dropped = 0usize;
        for (_, key, size) in entries {
            if current <= cap_bytes {
                break;
            }
            if self.remove(&key).is_ok() {
                current = current.saturating_sub(size);
                dropped += 1;
            }
        }
        Ok(dropped)
    }

    /// Per-entry disk stats as `(payload mtime, key, payload+meta bytes)`,
    /// sorted oldest-first with the store key as a deterministic tiebreaker
    /// (filesystem mtime granularity collapses many same-session writes onto
    /// identical timestamps, so without it the ordering at a cap boundary
    /// would vary across runs). A meta.json whose sibling payload is absent
    /// (the half-written state a crash between the two atomic renames leaves)
    /// is included with its own size so callers count and reclaim it rather
    /// than over-evicting healthy entries.
    pub fn entry_stats(&self) -> Result<Vec<(std::time::SystemTime, Key, u64)>, StoreError> {
        let mut entries: Vec<(std::time::SystemTime, Key, u64)> = Vec::new();
        let dir = match fs::read_dir(&self.root) {
            Ok(e) => e,
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(entries),
            Err(e) => return Err(e.into()),
        };
        for shard in dir.flatten() {
            let shard_path = shard.path();
            if !shard_path.is_dir() {
                continue;
            }
            for entry in fs::read_dir(&shard_path)?.flatten() {
                let p = entry.path();
                let name = match p.file_name().and_then(|n| n.to_str()) {
                    Some(s) => s.to_string(),
                    None => continue,
                };
                if let Some(stem) = name.strip_suffix(".payload") {
                    let key = Key(stem.to_string());
                    if key.validate().is_err() {
                        continue;
                    }
                    let md = entry.metadata()?;
                    let payload_len = md.len();
                    let meta_len = fs::metadata(self.meta_path(&key))
                        .map(|m| m.len())
                        .unwrap_or(0);
                    let mtime = md.modified().unwrap_or(std::time::UNIX_EPOCH);
                    entries.push((mtime, key, payload_len + meta_len));
                } else if let Some(stem) = name.strip_suffix(".meta.json") {
                    let key = Key(stem.to_string());
                    if key.validate().is_err() || self.payload_path(&key).exists() {
                        continue;
                    }
                    let md = entry.metadata()?;
                    let mtime = md.modified().unwrap_or(std::time::UNIX_EPOCH);
                    entries.push((mtime, key, md.len()));
                }
            }
        }
        entries.sort_by(|(ta, ka, _), (tb, kb, _)| ta.cmp(tb).then_with(|| ka.0.cmp(&kb.0)));
        Ok(entries)
    }

    /// Iterate every (key, meta) pair in the store. Used by `LiveCache`
    /// to rehydrate its in-memory registry on startup so cross-process
    /// cache hits work — without this, a fresh MCP server would see an
    /// empty registry and miss every lookup until it re-populated each
    /// entry from scratch.
    pub fn iter_meta(&self) -> Result<Vec<(Key, PayloadMeta)>, StoreError> {
        let mut out = Vec::new();
        let entries = match fs::read_dir(&self.root) {
            Ok(e) => e,
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(out),
            Err(e) => return Err(e.into()),
        };
        for shard in entries.flatten() {
            let shard_path = shard.path();
            if !shard_path.is_dir() {
                continue;
            }
            for entry in fs::read_dir(&shard_path)?.flatten() {
                let p = entry.path();
                let name = match p.file_name().and_then(|n| n.to_str()) {
                    Some(s) if s.ends_with(".meta.json") => s.to_string(),
                    _ => continue,
                };
                let key_hex = name.trim_end_matches(".meta.json").to_string();
                let key = Key(key_hex);
                if key.validate().is_err() {
                    continue;
                }
                let meta: PayloadMeta = match fs::read(&p)
                    .ok()
                    .and_then(|b| serde_json::from_slice(&b).ok())
                {
                    Some(m) => m,
                    None => continue,
                };
                out.push((key, meta));
            }
        }
        Ok(out)
    }

    /// Look up a payload. Returns `None` if absent. Returns
    /// `StoreError::Integrity` if the payload bytes on disk do not match
    /// the recorded hash, which indicates corruption (torn write that
    /// somehow survived, on-disk tamper, hardware fault) and must not
    /// silently return wrong bytes to the model.
    pub fn lookup(&self, key: &Key) -> Result<Option<Payload>, StoreError> {
        key.validate()?;
        let Some(_lock) = lock_shard(&self.shard_dir(key), LockMode::Shared)? else {
            return Ok(None);
        };
        let pp = self.payload_path(key);
        let mp = self.meta_path(key);
        // NotFound on either file is a miss, not an io error: the key is
        // absent, or one file is an orphan left by a crash between the two
        // renames or by a binary that predates the shard lock.
        let mut bytes = Vec::new();
        match fs::File::open(&pp).and_then(|mut f| f.read_to_end(&mut bytes)) {
            Ok(_) => {}
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None),
            Err(e) => return Err(e.into()),
        }
        let meta_bytes = match fs::read(&mp) {
            Ok(b) => b,
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None),
            Err(e) => return Err(e.into()),
        };
        let meta: PayloadMeta = serde_json::from_slice(&meta_bytes)?;

        let actual = blake3::hash(&bytes).to_hex().to_string();
        if actual != meta.payload_hash {
            return Err(StoreError::Integrity {
                key: key.0.clone(),
                expected: meta.payload_hash.clone(),
                actual,
            });
        }
        Ok(Some(Payload { bytes, meta }))
    }

    /// True if the key has a complete entry (both payload and meta).
    /// Useful for tests and stats; the integrity check still fires on
    /// `lookup`, so this is an existence-only check.
    pub fn contains(&self, key: &Key) -> bool {
        key.validate().is_ok() && self.payload_path(key).exists() && self.meta_path(key).exists()
    }
}

impl Store for FileStore {
    fn persist_with_upstreams(
        &self,
        key: &Key,
        bytes: &[u8],
        tool_kind: &str,
        file_roots: Vec<FileRootSerde>,
        upstream_keys: Vec<String>,
    ) -> Result<(), StoreError> {
        FileStore::persist_with_upstreams(self, key, bytes, tool_kind, file_roots, upstream_keys)
    }

    fn lookup(&self, key: &Key) -> Result<Option<Payload>, StoreError> {
        FileStore::lookup(self, key)
    }

    fn remove(&self, key: &Key) -> Result<(), StoreError> {
        FileStore::remove(self, key)
    }

    fn total_bytes(&self) -> Result<u64, StoreError> {
        FileStore::total_bytes(self)
    }

    fn evict_to_cap(&self, cap_bytes: u64) -> Result<usize, StoreError> {
        FileStore::evict_to_cap(self, cap_bytes)
    }

    fn iter_meta(&self) -> Result<Vec<(Key, PayloadMeta)>, StoreError> {
        FileStore::iter_meta(self)
    }

    fn contains(&self, key: &Key) -> bool {
        FileStore::contains(self, key)
    }
}

/// Name of the 0-byte lock file in each shard directory. Writers hold it
/// exclusive across an entry's two renames or two unlinks and readers hold it
/// shared across their two reads, so no process observes half of a pair. It
/// is the innermost lock: callers may hold another lock (the exec cache's
/// `blobs.lock`) while calling the store, but nothing is acquired under it.
const SHARD_LOCK: &str = ".lock";

enum LockMode {
    Shared,
    Exclusive,
}

/// Locks a shard on a fresh descriptor, since `flock` excludes per open file
/// description and one store serves many threads. Returns `None` when the
/// shard directory does not exist; shards written before the lock existed get
/// their lock file on first use. A shared lock opens an existing lock file
/// read-only, so a reader without write access to the store can still look up.
fn lock_shard(shard: &Path, mode: LockMode) -> io::Result<Option<fs::File>> {
    let path = shard.join(SHARD_LOCK);
    let existing = match mode {
        LockMode::Shared => fs::File::open(&path),
        LockMode::Exclusive => fs::OpenOptions::new().write(true).open(&path),
    };
    let file = match existing {
        Ok(f) => f,
        Err(e) if e.kind() == io::ErrorKind::NotFound => match fs::OpenOptions::new()
            .write(true)
            .create(true)
            .truncate(false)
            .open(&path)
        {
            Ok(f) => f,
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None),
            Err(e) => return Err(e),
        },
        Err(e) => return Err(e),
    };
    match mode {
        LockMode::Shared => file.lock_shared()?,
        LockMode::Exclusive => file.lock()?,
    }
    Ok(Some(file))
}

fn remove_if_present(path: &Path) -> io::Result<()> {
    match fs::remove_file(path) {
        Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
        other => other,
    }
}

/// Test-only: fixes the random part of temp names so a test can occupy them.
const TEMP_SEED_ENV: &str = "VERDANT_TEST_TEMP_SEED";
const TEMP_NAME_ATTEMPTS: u64 = 8;

/// Random part of a temp name. The pid alone does not separate writers: every
/// container sharing a store runs its server as pid 1.
fn temp_name_component(attempt: u64) -> u64 {
    static SEED: OnceLock<Option<u64>> = OnceLock::new();
    let seed = *SEED.get_or_init(|| {
        std::env::var(TEMP_SEED_ENV)
            .ok()
            .and_then(|s| s.parse().ok())
    });
    if let Some(seed) = seed {
        return seed.wrapping_add(attempt);
    }
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let mut hasher = RandomState::new().build_hasher();
    hasher.write_u64(COUNTER.fetch_add(1, Ordering::Relaxed));
    hasher.write_u128(nanos);
    hasher.finish()
}

// Minimal in-tree temp-file guard. Holds a file handle plus its path; the
// guard removes the file on drop so a failed write or rename does not leak.
struct TempGuard {
    path: PathBuf,
    file: fs::File,
    armed: bool,
}

impl TempGuard {
    fn create_in(dir: &Path) -> io::Result<Self> {
        let pid = std::process::id();
        for attempt in 0..TEMP_NAME_ATTEMPTS {
            let path = dir.join(format!(
                ".verdant-tmp-{pid}-{:016x}",
                temp_name_component(attempt)
            ));
            match fs::OpenOptions::new()
                .write(true)
                .create_new(true)
                .open(&path)
            {
                Ok(file) => {
                    return Ok(Self {
                        path,
                        file,
                        armed: true,
                    })
                }
                Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
                Err(e) => return Err(e),
            }
        }
        Err(io::Error::new(
            io::ErrorKind::AlreadyExists,
            format!(
                "no free temp name in {} after {TEMP_NAME_ATTEMPTS} attempts",
                dir.display()
            ),
        ))
    }

    /// A sibling temp in the target's directory, so the later rename is
    /// atomic on POSIX.
    fn write_in(dir: &Path, bytes: &[u8]) -> io::Result<Self> {
        let mut guard = Self::create_in(dir)?;
        guard.file.write_all(bytes)?;
        guard.file.flush()?;
        Ok(guard)
    }

    fn persist(mut self, target: &Path) -> io::Result<()> {
        fs::rename(&self.path, target)?;
        self.armed = false;
        Ok(())
    }
}

impl Drop for TempGuard {
    fn drop(&mut self) {
        if self.armed {
            let _ = fs::remove_file(&self.path);
        }
    }
}

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

    fn store() -> (TempDir, FileStore) {
        let dir = TempDir::new().unwrap();
        let s = FileStore::open(dir.path().to_path_buf()).unwrap();
        (dir, s)
    }

    #[test]
    fn persist_then_lookup_roundtrip() {
        let (_d, s) = store();
        let k = Key::from_bytes(b"input-1");
        s.persist(&k, b"hello world", "read", vec![]).unwrap();
        let p = s.lookup(&k).unwrap().expect("must exist");
        assert_eq!(p.bytes, b"hello world");
        assert_eq!(p.meta.tool_kind, "read");
        assert_eq!(p.meta.bytes, 11);
    }

    #[test]
    fn lookup_missing_returns_none() {
        let (_d, s) = store();
        let k = Key::from_bytes(b"never-written");
        assert!(s.lookup(&k).unwrap().is_none());
    }

    #[test]
    fn integrity_violation_detected() {
        let (_d, s) = store();
        let k = Key::from_bytes(b"input-2");
        s.persist(&k, b"trusted", "read", vec![]).unwrap();
        // Corrupt the payload on disk under the store's feet — a real
        // failure would be hardware-induced or external tamper, but a
        // direct overwrite is the cheapest reproducible test.
        let pp = s.root.join(&k.0[..2]).join(format!("{}.payload", k.0));
        fs::write(&pp, b"tampered").unwrap();
        let err = s.lookup(&k).expect_err("integrity must fire");
        assert!(matches!(err, StoreError::Integrity { .. }));
    }

    #[test]
    fn partial_write_only_meta_returns_none() {
        // Simulates a case where meta landed but payload did not (or
        // vice-versa); since we use atomic rename per file but the *pair*
        // is not jointly atomic, a crash between the two renames can
        // leave one orphan. lookup must treat that as cache miss, not
        // partial data.
        let (_d, s) = store();
        let k = Key::from_bytes(b"input-3");
        // Manually drop only a meta file.
        fs::create_dir_all(s.root.join(&k.0[..2])).unwrap();
        let mp = s.root.join(&k.0[..2]).join(format!("{}.meta.json", k.0));
        fs::write(
            &mp,
            serde_json::to_vec(&PayloadMeta {
                payload_hash: blake3::hash(b"orphan").to_hex().to_string(),
                bytes: 6,
                tool_kind: "read".into(),
                file_roots: vec![],
                upstream_keys: vec![],
            })
            .unwrap(),
        )
        .unwrap();
        assert!(s.lookup(&k).unwrap().is_none());
    }

    #[test]
    fn lookup_orphan_missing_payload_returns_none_not_err() {
        // A crash between the two atomic renames (or eviction removing
        // one sibling) leaves a meta with no payload. lookup must treat
        // this benign race as a miss, not surface an io error.
        let (_d, s) = store();
        let k = Key::from_bytes(b"orphan-meta");
        s.persist(&k, b"payload bytes", "read", vec![]).unwrap();
        fs::remove_file(s.payload_path(&k)).unwrap();
        assert!(
            matches!(s.lookup(&k), Ok(None)),
            "payload-missing/meta-present must be Ok(None)"
        );
    }

    #[test]
    fn lookup_orphan_missing_meta_returns_none_not_err() {
        let (_d, s) = store();
        let k = Key::from_bytes(b"orphan-payload");
        s.persist(&k, b"payload bytes", "read", vec![]).unwrap();
        fs::remove_file(s.meta_path(&k)).unwrap();
        assert!(
            matches!(s.lookup(&k), Ok(None)),
            "meta-missing/payload-present must be Ok(None)"
        );
    }

    #[test]
    fn evict_reclaims_meta_only_orphans() {
        // A meta.json whose sibling payload is absent still occupies
        // disk. evict_to_cap must count and remove it, not over-evict
        // healthy entries to compensate.
        let (_d, s) = store();
        let healthy = Key::from_bytes(b"healthy");
        s.persist(&healthy, &[b'h'; 4096], "read", vec![]).unwrap();

        let orphan = Key::from_bytes(b"orphan-entry");
        fs::create_dir_all(s.shard_dir(&orphan)).unwrap();
        let orphan_meta = serde_json::to_vec(&PayloadMeta {
            payload_hash: blake3::hash(b"gone").to_hex().to_string(),
            bytes: 4,
            tool_kind: "read".into(),
            file_roots: vec![],
            upstream_keys: vec![],
        })
        .unwrap();
        fs::write(s.meta_path(&orphan), &orphan_meta).unwrap();

        let dropped = s.evict_to_cap(0).unwrap();
        assert!(dropped >= 2, "both healthy entry and orphan must drop");
        assert!(
            !s.meta_path(&orphan).exists(),
            "meta-only orphan must be removed"
        );
        assert_eq!(s.total_bytes().unwrap(), 0);
    }

    #[test]
    fn evict_order_is_deterministic_for_equal_mtimes() {
        // Filesystem mtime granularity collapses same-session writes to
        // identical timestamps. With every entry sharing one mtime the
        // primary sort key is a constant, so the eviction victim is
        // fully determined by the secondary key-order tiebreaker: the
        // lexicographically-smallest store keys must be the ones
        // dropped, independent of read_dir enumeration order.
        let fixed = std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
        let dir = TempDir::new().unwrap();
        let s = FileStore::open(dir.path().to_path_buf()).unwrap();
        let mut keys: Vec<Key> = (0..8)
            .map(|i| Key::from_bytes(format!("dk{i}").as_bytes()))
            .collect();
        for k in &keys {
            s.persist(k, &[b'x'; 4096], "read", vec![]).unwrap();
            fs::File::options()
                .write(true)
                .open(s.payload_path(k))
                .unwrap()
                .set_modified(fixed)
                .unwrap();
        }
        let before = s.total_bytes().unwrap();
        let dropped = s.evict_to_cap(before / 2).unwrap();
        assert!(dropped > 0, "eviction must drop at least one entry");

        let evicted: std::collections::HashSet<String> = keys
            .iter()
            .filter(|k| s.lookup(k).unwrap().is_none())
            .map(|k| k.0.clone())
            .collect();
        keys.sort_by(|a, b| a.0.cmp(&b.0));
        let expected: std::collections::HashSet<String> = keys
            .iter()
            .take(evicted.len())
            .map(|k| k.0.clone())
            .collect();
        assert_eq!(
            evicted, expected,
            "with equal mtimes the lowest store keys must be the deterministic victims"
        );
    }

    #[test]
    fn total_bytes_sums_payloads_and_meta() {
        let (_d, s) = store();
        assert_eq!(s.total_bytes().unwrap(), 0, "fresh store is zero bytes");
        let k = Key::from_bytes(b"size-test");
        s.persist(&k, &[b'x'; 1024], "read", vec![]).unwrap();
        let bytes = s.total_bytes().unwrap();
        assert!(bytes >= 1024, "payload alone is ≥1024, got {bytes}");
    }

    #[test]
    fn evict_to_cap_drops_oldest_first() {
        let (_d, s) = store();
        // Persist four entries with distinct mtimes (sleep briefly so
        // the filesystem mtime resolution doesn't collapse them).
        let keys: Vec<Key> = (0..4)
            .map(|i| Key::from_bytes(format!("k{i}").as_bytes()))
            .collect();
        for (i, k) in keys.iter().enumerate() {
            s.persist(k, &[b'A' + i as u8; 4096], "read", vec![])
                .unwrap();
            std::thread::sleep(std::time::Duration::from_millis(20));
        }
        let before = s.total_bytes().unwrap();
        assert!(before >= 4 * 4096);

        // Cap to roughly two entries worth.
        let cap = before / 2;
        let dropped = s.evict_to_cap(cap).unwrap();
        assert!(dropped >= 1, "should drop at least one entry");
        let after = s.total_bytes().unwrap();
        assert!(
            after <= cap,
            "after eviction must fit cap; got {after}/{cap}"
        );

        // The oldest key (k0) must be gone; the newest (k3) must
        // still be present.
        assert!(s.lookup(&keys[0]).unwrap().is_none(), "oldest must evict");
        assert!(s.lookup(&keys[3]).unwrap().is_some(), "newest must survive");
    }

    #[test]
    fn evict_below_cap_is_noop() {
        let (_d, s) = store();
        let k = Key::from_bytes(b"small");
        s.persist(&k, b"tiny", "read", vec![]).unwrap();
        let dropped = s.evict_to_cap(u64::MAX).unwrap();
        assert_eq!(dropped, 0);
        assert!(s.lookup(&k).unwrap().is_some());
    }

    #[test]
    fn malformed_key_rejected() {
        let (_d, s) = store();
        let bad = Key("not-hex".to_string());
        assert!(s.persist(&bad, b"x", "read", vec![]).is_err());
        assert!(s.lookup(&bad).is_err());
    }

    #[test]
    fn shard_dirs_distribute_keys() {
        let (_d, s) = store();
        for i in 0..16u8 {
            let k = Key::from_bytes(&[i, i, i]);
            s.persist(&k, &[i], "read", vec![]).unwrap();
        }
        // Count distinct two-char shard directories. With 16 random
        // blake3 hashes the chance of every key collapsing into one
        // shard is astronomically small; we assert that we have at
        // least four distinct shards.
        let mut shards = std::collections::HashSet::new();
        for entry in fs::read_dir(s.root()).unwrap() {
            let e = entry.unwrap();
            if e.path().is_dir() {
                shards.insert(e.file_name().to_string_lossy().to_string());
            }
        }
        assert!(shards.len() >= 4, "shards = {shards:?}");
    }

    fn hold_shard_lock(s: &FileStore, k: &Key) -> fs::File {
        let lock = fs::File::open(s.shard_dir(k).join(SHARD_LOCK)).unwrap();
        lock.lock().unwrap();
        lock
    }

    #[test]
    fn persist_waits_for_the_shard_lock_holder() {
        let (_d, s) = store();
        let s = std::sync::Arc::new(s);
        let k = Key::from_bytes(b"locked");
        s.persist(&k, b"first", "read", vec![]).unwrap();
        let held = hold_shard_lock(&s, &k);

        let (tx, rx) = std::sync::mpsc::channel();
        let writer = {
            let (s, k) = (s.clone(), k.clone());
            std::thread::spawn(move || {
                s.persist(&k, b"second", "read", vec![]).unwrap();
                tx.send(()).unwrap();
            })
        };
        assert!(
            rx.recv_timeout(std::time::Duration::from_millis(300))
                .is_err(),
            "persist must wait while another descriptor holds the shard lock"
        );
        assert_eq!(fs::read(s.payload_path(&k)).unwrap(), b"first");
        drop(held);
        rx.recv_timeout(std::time::Duration::from_secs(10))
            .expect("persist must finish once the lock is released");
        writer.join().unwrap();
        assert_eq!(s.lookup(&k).unwrap().unwrap().bytes, b"second");
    }

    #[test]
    fn shard_lock_is_free_when_each_call_returns() {
        let (_d, s) = store();
        let k = Key::from_bytes(b"innermost");
        let try_exclusive = |s: &FileStore| {
            fs::File::open(s.shard_dir(&k).join(SHARD_LOCK))
                .unwrap()
                .try_lock()
                .expect("no store call may return holding its shard lock");
        };
        s.persist(&k, b"x", "read", vec![]).unwrap();
        try_exclusive(&s);
        assert!(s.lookup(&k).unwrap().is_some());
        try_exclusive(&s);
        s.remove(&k).unwrap();
        try_exclusive(&s);
        assert!(s.lookup(&k).unwrap().is_none());
    }

    #[test]
    fn lookup_and_remove_do_not_create_missing_shards() {
        let (_d, s) = store();
        let k = Key::from_bytes(b"absent");
        assert!(s.lookup(&k).unwrap().is_none());
        s.remove(&k).unwrap();
        assert!(!s.shard_dir(&k).exists());
    }

    #[test]
    fn lookup_needs_no_write_access_to_the_lock_file() {
        use std::os::unix::fs::PermissionsExt as _;
        let (_d, s) = store();
        let k = Key::from_bytes(b"read-only-lock");
        s.persist(&k, b"payload", "read", vec![]).unwrap();
        let lock = s.shard_dir(&k).join(SHARD_LOCK);
        fs::set_permissions(&lock, fs::Permissions::from_mode(0o444)).unwrap();
        assert_eq!(s.lookup(&k).unwrap().unwrap().bytes, b"payload");
    }

    #[test]
    fn lock_file_is_not_an_entry() {
        let (_d, s) = store();
        let k = Key::from_bytes(b"with-lock");
        s.persist(&k, b"payload", "read", vec![]).unwrap();
        assert!(s.shard_dir(&k).join(SHARD_LOCK).exists());
        assert_eq!(s.iter_meta().unwrap().len(), 1);
        assert_eq!(s.entry_stats().unwrap().len(), 1);
        s.remove(&k).unwrap();
        assert_eq!(s.total_bytes().unwrap(), 0);
    }

    #[test]
    fn contains_only_when_complete() {
        let (_d, s) = store();
        let k = Key::from_bytes(b"x");
        assert!(!s.contains(&k));
        s.persist(&k, b"y", "read", vec![]).unwrap();
        assert!(s.contains(&k));
    }
}