eidetic-engine 0.15.2

//! Filesystem-backed L2 cache for serialized context-pack JSON.

use std::fmt;
use std::fs::{self, File, FileTimes, OpenOptions};
use std::io::{self, Read, Write};
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD};
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;

pub const PACK_L2_CACHE_ENTRY_SCHEMA_V1: &str = "ee.pack.l2_cache.entry.v1";
pub const PACK_L2_CACHE_ENTRY_SCHEMA_V2: &str = "ee.pack.l2_cache.entry.v2";
pub const DEFAULT_MAX_BYTES: u64 = 256 * 1024 * 1024;
pub const DEFAULT_MAX_ENTRY_BYTES: u64 = 1024 * 1024;
const PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1: &str = "zstd_frame_v1";
const PACK_L2_COMPRESSION_LEVEL: i32 = 3;
const DEFAULT_MAX_AGE: Duration = Duration::from_secs(30 * 24 * 60 * 60);
static TEMP_FILE_COUNTER: AtomicU64 = AtomicU64::new(0);

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2CacheOptions {
    pub max_bytes: u64,
    pub max_entry_bytes: u64,
    pub max_age: Duration,
}

impl PackL2CacheOptions {
    #[must_use]
    pub const fn new(max_bytes: u64, max_age: Duration) -> Self {
        Self {
            max_bytes,
            max_entry_bytes: DEFAULT_MAX_ENTRY_BYTES,
            max_age,
        }
    }

    #[must_use]
    pub const fn with_max_entry_bytes(mut self, max_entry_bytes: u64) -> Self {
        self.max_entry_bytes = max_entry_bytes;
        self
    }
}

impl Default for PackL2CacheOptions {
    fn default() -> Self {
        Self {
            max_bytes: DEFAULT_MAX_BYTES,
            max_entry_bytes: DEFAULT_MAX_ENTRY_BYTES,
            max_age: DEFAULT_MAX_AGE,
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2Cache {
    root: PathBuf,
    options: PackL2CacheOptions,
}

impl PackL2Cache {
    #[must_use]
    pub fn new(root: impl Into<PathBuf>, options: PackL2CacheOptions) -> Self {
        Self {
            root: root.into(),
            options,
        }
    }

    #[must_use]
    pub fn root(&self) -> &Path {
        &self.root
    }

    #[must_use]
    pub fn options(&self) -> &PackL2CacheOptions {
        &self.options
    }

    #[must_use]
    pub fn entry_path(&self, key: &str) -> PathBuf {
        self.root.join(cache_file_name(key))
    }

    #[must_use]
    pub fn entry_path_for_body_hash(&self, key: &str, body_hash_prefix: &str) -> PathBuf {
        self.root
            .join(cache_file_name_with_body_hash(key, body_hash_prefix))
    }

    pub fn get(&self, key: &str) -> Result<PackL2CacheLookup, PackL2CacheError> {
        self.get_at(key, system_time_seconds(SystemTime::now())?)
    }

    pub fn get_at(
        &self,
        key: &str,
        now_epoch_seconds: u64,
    ) -> Result<PackL2CacheLookup, PackL2CacheError> {
        let fallback_path = self.entry_path(key);
        let candidates = self.entry_candidates(key)?;
        if candidates.is_empty() {
            return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
                key: key.to_owned(),
                path: fallback_path,
                reason: PackL2CacheMissReason::NotFound,
            }));
        }

        let mut last_miss = None;
        for path in candidates {
            match self.get_candidate_at(key, path, now_epoch_seconds)? {
                PackL2CacheLookup::Hit(hit) => return Ok(PackL2CacheLookup::Hit(hit)),
                PackL2CacheLookup::Miss(miss) => {
                    last_miss = Some(miss);
                }
            }
        }

        Ok(PackL2CacheLookup::Miss(last_miss.unwrap_or(
            PackL2CacheMiss {
                key: key.to_owned(),
                path: fallback_path,
                reason: PackL2CacheMissReason::NotFound,
            },
        )))
    }

    fn get_candidate_at(
        &self,
        key: &str,
        path: PathBuf,
        now_epoch_seconds: u64,
    ) -> Result<PackL2CacheLookup, PackL2CacheError> {
        ensure_no_symlink_components(&path, "inspect_entry")?;
        let bytes = match read_cache_entry_file(&path, self.options.max_entry_bytes) {
            Ok(bytes) => bytes,
            Err(error) if error.kind() == io::ErrorKind::NotFound => {
                return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
                    key: key.to_owned(),
                    path,
                    reason: PackL2CacheMissReason::NotFound,
                }));
            }
            Err(error) => {
                return Err(PackL2CacheError::Io {
                    path,
                    operation: "read",
                    source: error,
                });
            }
        };

        if let Some(expected_body_hash_prefix) = body_hash_prefix_from_path(&path) {
            let actual_body_hash_prefix = body_hash_prefix(&bytes);
            if actual_body_hash_prefix != expected_body_hash_prefix {
                remove_cache_entry_best_effort(&path);
                return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
                    key: key.to_owned(),
                    path,
                    reason: PackL2CacheMissReason::BodyHashMismatch {
                        expected: expected_body_hash_prefix,
                        actual: actual_body_hash_prefix,
                    },
                }));
            }
        }

        let byte_len = bytes.len() as u64;
        if byte_len > self.options.max_entry_bytes {
            remove_cache_entry_best_effort(&path);
            return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
                key: key.to_owned(),
                path,
                reason: PackL2CacheMissReason::TooLarge {
                    byte_len,
                    max_entry_bytes: self.options.max_entry_bytes,
                },
            }));
        }

        let entry = match decode_pack_l2_cache_entry(&bytes, self.options.max_entry_bytes) {
            Ok(entry) => entry,
            Err(reason) => {
                remove_cache_entry_best_effort(&path);
                return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
                    key: key.to_owned(),
                    path,
                    reason,
                }));
            }
        };

        if entry.key != key {
            remove_cache_entry_best_effort(&path);
            return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
                key: key.to_owned(),
                path,
                reason: PackL2CacheMissReason::KeyMismatch {
                    stored_key: entry.key,
                },
            }));
        }
        if is_expired(
            entry.stored_at_epoch_seconds,
            now_epoch_seconds,
            self.options.max_age,
        ) {
            return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
                key: key.to_owned(),
                path,
                reason: PackL2CacheMissReason::Expired {
                    stored_at_epoch_seconds: entry.stored_at_epoch_seconds,
                },
            }));
        }

        touch_cache_entry_mtime_best_effort(&path, now_epoch_seconds);
        Ok(PackL2CacheLookup::Hit(PackL2CacheHit {
            key: entry.key,
            path,
            stored_at_epoch_seconds: entry.stored_at_epoch_seconds,
            pack_json: entry.pack_json,
            byte_len,
            compression: entry.compression,
        }))
    }

    pub fn put(
        &self,
        key: &str,
        pack_json: &JsonValue,
    ) -> Result<PackL2WriteReport, PackL2CacheError> {
        self.put_at(key, pack_json, system_time_seconds(SystemTime::now())?)
    }

    pub fn put_at(
        &self,
        key: &str,
        pack_json: &JsonValue,
        stored_at_epoch_seconds: u64,
    ) -> Result<PackL2WriteReport, PackL2CacheError> {
        let path = self.entry_path(key);
        let entry = PackL2CacheEntry {
            schema: PACK_L2_CACHE_ENTRY_SCHEMA_V1.to_owned(),
            key: key.to_owned(),
            stored_at_epoch_seconds,
            pack_json: pack_json.clone(),
        };
        let bytes = serde_json::to_vec(&entry).map_err(|source| PackL2CacheError::Json {
            path: path.clone(),
            operation: "serialize",
            source,
        })?;
        let byte_len = bytes.len() as u64;
        let body_hash_prefix = body_hash_prefix(&bytes);
        let path = self.entry_path_for_body_hash(key, &body_hash_prefix);
        if byte_len > self.options.max_entry_bytes {
            return Ok(PackL2WriteReport {
                key: key.to_owned(),
                path,
                byte_len,
                uncompressed_byte_len: byte_len,
                compression: None,
                outcome: PackL2WriteOutcome::SkippedTooLarge {
                    max_entry_bytes: self.options.max_entry_bytes,
                },
                eviction: PackL2EvictionReport::default(),
            });
        }

        ensure_cache_dir(&self.root)?;
        let temp_path = self.temp_path(key, &body_hash_prefix, stored_at_epoch_seconds);
        ensure_no_symlink_components(&path, "inspect_entry")?;
        ensure_no_symlink_components(&temp_path, "inspect_temp")?;

        write_synced_file(&temp_path, &bytes)?;
        publish_cache_entry_temp_file(&temp_path, &path)?;
        touch_cache_entry_mtime_best_effort(&path, stored_at_epoch_seconds);
        sync_directory(&self.root)?;
        let duplicate_cleanup = self.prune_duplicate_key_entries_best_effort(key, &path)?;
        let eviction = merge_cache_cleanup_reports(
            duplicate_cleanup,
            self.evict_best_effort_at(stored_at_epoch_seconds)?,
        );

        Ok(PackL2WriteReport {
            key: key.to_owned(),
            path,
            byte_len,
            uncompressed_byte_len: byte_len,
            compression: None,
            outcome: PackL2WriteOutcome::Stored,
            eviction,
        })
    }

    pub fn put_compressed(
        &self,
        key: &str,
        pack_json: &JsonValue,
    ) -> Result<PackL2WriteReport, PackL2CacheError> {
        self.put_compressed_with_dictionary_at(
            key,
            pack_json,
            None,
            system_time_seconds(SystemTime::now())?,
        )
    }

    pub fn put_compressed_at(
        &self,
        key: &str,
        pack_json: &JsonValue,
        stored_at_epoch_seconds: u64,
    ) -> Result<PackL2WriteReport, PackL2CacheError> {
        self.put_compressed_with_dictionary_at(key, pack_json, None, stored_at_epoch_seconds)
    }

    pub fn put_compressed_with_dictionary_at(
        &self,
        key: &str,
        pack_json: &JsonValue,
        dictionary: Option<&PackL2CompressionDictionary>,
        stored_at_epoch_seconds: u64,
    ) -> Result<PackL2WriteReport, PackL2CacheError> {
        let path = self.entry_path(key);
        let uncompressed =
            serde_json::to_vec(pack_json).map_err(|source| PackL2CacheError::Json {
                path: path.clone(),
                operation: "serialize_uncompressed",
                source,
            })?;
        let uncompressed_byte_len = uncompressed.len() as u64;
        let compression_start = Instant::now();
        let compressed = zstd_compress(&uncompressed, dictionary)?;
        let compression_latency_ms = elapsed_millis(compression_start.elapsed());
        let compressed_byte_len = compressed.len() as u64;
        let entry = PackL2CacheEntryV2 {
            schema: PACK_L2_CACHE_ENTRY_SCHEMA_V2.to_owned(),
            key: key.to_owned(),
            stored_at_epoch_seconds,
            compression: PackL2CacheCompressionPayload {
                algorithm: PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1.to_owned(),
                compressed_payload_base64: BASE64_STANDARD.encode(&compressed),
                compressed_byte_len,
                uncompressed_byte_len,
                uncompressed_hash: blake3_hash(&uncompressed),
                dictionary: dictionary.map(PackL2CacheCompressionDictionaryRef::from_dictionary),
            },
        };
        let bytes = serde_json::to_vec(&entry).map_err(|source| PackL2CacheError::Json {
            path: path.clone(),
            operation: "serialize_compressed",
            source,
        })?;
        let byte_len = bytes.len() as u64;
        let body_hash_prefix = body_hash_prefix(&bytes);
        let path = self.entry_path_for_body_hash(key, &body_hash_prefix);
        let compression_report = PackL2CompressionWriteReport {
            algorithm: PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1.to_owned(),
            dictionary_id: dictionary.map(|dictionary| dictionary.id.clone()),
            compressed_bytes: compressed_byte_len,
            uncompressed_bytes: uncompressed_byte_len,
            compression_latency_ms,
        };
        if byte_len > self.options.max_entry_bytes {
            return Ok(PackL2WriteReport {
                key: key.to_owned(),
                path,
                byte_len,
                uncompressed_byte_len,
                compression: Some(compression_report),
                outcome: PackL2WriteOutcome::SkippedTooLarge {
                    max_entry_bytes: self.options.max_entry_bytes,
                },
                eviction: PackL2EvictionReport::default(),
            });
        }

        ensure_cache_dir(&self.root)?;
        let temp_path = self.temp_path(key, &body_hash_prefix, stored_at_epoch_seconds);
        ensure_no_symlink_components(&path, "inspect_entry")?;
        ensure_no_symlink_components(&temp_path, "inspect_temp")?;

        write_synced_file(&temp_path, &bytes)?;
        publish_cache_entry_temp_file(&temp_path, &path)?;
        touch_cache_entry_mtime_best_effort(&path, stored_at_epoch_seconds);
        sync_directory(&self.root)?;
        let duplicate_cleanup = self.prune_duplicate_key_entries_best_effort(key, &path)?;
        let eviction = merge_cache_cleanup_reports(
            duplicate_cleanup,
            self.evict_best_effort_at(stored_at_epoch_seconds)?,
        );

        Ok(PackL2WriteReport {
            key: key.to_owned(),
            path,
            byte_len,
            uncompressed_byte_len,
            compression: Some(compression_report),
            outcome: PackL2WriteOutcome::Stored,
            eviction,
        })
    }

    pub fn evict_best_effort(&self) -> Result<PackL2EvictionReport, PackL2CacheError> {
        self.evict_best_effort_at(system_time_seconds(SystemTime::now())?)
    }

    pub fn evict_best_effort_at(
        &self,
        now_epoch_seconds: u64,
    ) -> Result<PackL2EvictionReport, PackL2CacheError> {
        ensure_no_symlink_components(&self.root, "inspect_root")?;
        let mut report = PackL2EvictionReport::default();
        let mut candidates = Vec::new();
        let entries = match fs::read_dir(&self.root) {
            Ok(entries) => entries,
            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(report),
            Err(error) => {
                return Err(PackL2CacheError::Io {
                    path: self.root.clone(),
                    operation: "read_dir",
                    source: error,
                });
            }
        };

        for entry in entries {
            let Ok(entry) = entry else {
                report.skipped = report.skipped.saturating_add(1);
                continue;
            };
            let path = entry.path();
            if path.extension().and_then(|extension| extension.to_str()) != Some("json") {
                continue;
            }
            let Ok(file_type) = entry.file_type() else {
                report.skipped = report.skipped.saturating_add(1);
                continue;
            };
            if file_type.is_symlink() {
                report.skipped = report.skipped.saturating_add(1);
                continue;
            }
            if !file_type.is_file() {
                continue;
            }
            let Ok(metadata) = fs::symlink_metadata(&path) else {
                report.skipped = report.skipped.saturating_add(1);
                continue;
            };
            let byte_len = metadata.len();
            report.bytes_before = report.bytes_before.saturating_add(byte_len);
            let fallback_epoch_seconds = metadata
                .modified()
                .ok()
                .and_then(|modified| system_time_seconds(modified).ok())
                .unwrap_or(0);
            let stored_epoch_seconds = cache_entry_stored_at(&path, self.options.max_entry_bytes)
                .unwrap_or(fallback_epoch_seconds);
            let last_used_epoch_seconds = fallback_epoch_seconds;
            let expired = stored_epoch_seconds == 0
                || is_expired(
                    stored_epoch_seconds,
                    now_epoch_seconds,
                    self.options.max_age,
                );
            candidates.push(EvictionCandidate {
                path,
                byte_len,
                stored_epoch_seconds,
                last_used_epoch_seconds,
                expired,
            });
        }

        candidates.sort_by(|left, right| {
            left.expired
                .cmp(&right.expired)
                .reverse()
                .then_with(|| {
                    left.last_used_epoch_seconds
                        .cmp(&right.last_used_epoch_seconds)
                })
                .then_with(|| left.stored_epoch_seconds.cmp(&right.stored_epoch_seconds))
                .then_with(|| left.path.cmp(&right.path))
        });

        let mut bytes_current = report.bytes_before;
        for candidate in candidates {
            if !candidate.expired && bytes_current <= self.options.max_bytes {
                break;
            }
            remove_eviction_candidate_file(&candidate, &mut report, &mut bytes_current);
        }
        report.bytes_after = bytes_current;
        Ok(report)
    }

    fn entry_candidates(&self, key: &str) -> Result<Vec<PathBuf>, PackL2CacheError> {
        ensure_no_symlink_components(&self.root, "inspect_root")?;
        let key_stem = cache_file_stem(key);
        let mut candidates = Vec::new();
        let entries = match fs::read_dir(&self.root) {
            Ok(entries) => entries,
            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(error) => {
                return Err(PackL2CacheError::Io {
                    path: self.root.clone(),
                    operation: "read_dir",
                    source: error,
                });
            }
        };

        for entry in entries {
            let entry = entry.map_err(|source| PackL2CacheError::Io {
                path: self.root.clone(),
                operation: "read_dir_entry",
                source,
            })?;
            let path = entry.path();
            let Some(file_name) = path.file_name().and_then(|file_name| file_name.to_str()) else {
                continue;
            };
            if file_name == cache_file_name(key)
                || body_hashed_file_name_matches(file_name, &key_stem)
            {
                let preference_epoch =
                    cache_entry_preference_epoch_seconds(&path, self.options.max_entry_bytes);
                candidates.push((path, preference_epoch));
            }
        }

        candidates.sort_by(|(left_path, left_epoch), (right_path, right_epoch)| {
            right_epoch
                .cmp(left_epoch)
                .then_with(|| left_path.cmp(right_path))
        });
        Ok(candidates.into_iter().map(|(path, _)| path).collect())
    }

    fn prune_duplicate_key_entries_best_effort(
        &self,
        key: &str,
        retained_path: &Path,
    ) -> Result<PackL2EvictionReport, PackL2CacheError> {
        ensure_no_symlink_components(&self.root, "inspect_root")?;
        let key_stem = cache_file_stem(key);
        let mut report = PackL2EvictionReport::default();
        let mut bytes_current = 0_u64;
        let entries = match fs::read_dir(&self.root) {
            Ok(entries) => entries,
            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(report),
            Err(error) => {
                return Err(PackL2CacheError::Io {
                    path: self.root.clone(),
                    operation: "read_dir",
                    source: error,
                });
            }
        };

        for entry in entries {
            let Ok(entry) = entry else {
                report.skipped = report.skipped.saturating_add(1);
                continue;
            };
            let path = entry.path();
            if path == retained_path {
                continue;
            }
            let Some(file_name) = path.file_name().and_then(|file_name| file_name.to_str()) else {
                continue;
            };
            if file_name != cache_file_name(key)
                && !body_hashed_file_name_matches(file_name, &key_stem)
            {
                continue;
            }
            let Ok(file_type) = entry.file_type() else {
                report.skipped = report.skipped.saturating_add(1);
                continue;
            };
            if file_type.is_symlink() {
                report.skipped = report.skipped.saturating_add(1);
                continue;
            }
            if !file_type.is_file() {
                continue;
            }
            let Ok(metadata) = fs::symlink_metadata(&path) else {
                report.skipped = report.skipped.saturating_add(1);
                continue;
            };
            let byte_len = metadata.len();
            report.bytes_before = report.bytes_before.saturating_add(byte_len);
            bytes_current = bytes_current.saturating_add(byte_len);
            let candidate = EvictionCandidate {
                path,
                byte_len,
                stored_epoch_seconds: 0,
                last_used_epoch_seconds: 0,
                expired: false,
            };
            remove_eviction_candidate_file(&candidate, &mut report, &mut bytes_current);
        }

        report.bytes_after = bytes_current;
        Ok(report)
    }

    fn temp_path(
        &self,
        key: &str,
        body_hash_prefix: &str,
        stored_at_epoch_seconds: u64,
    ) -> PathBuf {
        let process_id = std::process::id();
        let temp_counter = TEMP_FILE_COUNTER.fetch_add(1, Ordering::Relaxed);
        self.root.join(format!(
            ".{}.{}.{}.{}.{}.tmp",
            cache_file_stem(key),
            body_hash_prefix,
            process_id,
            stored_at_epoch_seconds,
            temp_counter
        ))
    }
}

#[derive(Clone, Debug, PartialEq)]
pub enum PackL2CacheLookup {
    Hit(PackL2CacheHit),
    Miss(PackL2CacheMiss),
}

impl PackL2CacheLookup {
    #[must_use]
    pub const fn is_hit(&self) -> bool {
        matches!(self, Self::Hit(_))
    }
}

#[derive(Clone, Debug, PartialEq)]
pub struct PackL2CacheHit {
    pub key: String,
    pub path: PathBuf,
    pub stored_at_epoch_seconds: u64,
    pub pack_json: JsonValue,
    pub byte_len: u64,
    pub compression: Option<PackL2CompressionHit>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2CompressionHit {
    pub algorithm: String,
    pub dictionary_id: Option<String>,
    pub compressed_bytes: u64,
    pub uncompressed_bytes: u64,
    pub decompression_latency_ms: u64,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2CacheMiss {
    pub key: String,
    pub path: PathBuf,
    pub reason: PackL2CacheMissReason,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum PackL2CacheMissReason {
    NotFound,
    Expired {
        stored_at_epoch_seconds: u64,
    },
    Corrupt(String),
    BodyHashMismatch {
        expected: String,
        actual: String,
    },
    KeyMismatch {
        stored_key: String,
    },
    TooLarge {
        byte_len: u64,
        max_entry_bytes: u64,
    },
    CompressionDictionaryMissing {
        dictionary_id: String,
    },
    CompressionDictionaryCorrupt {
        dictionary_id: String,
        message: String,
    },
    CompressionDecode {
        message: String,
    },
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2WriteReport {
    pub key: String,
    pub path: PathBuf,
    pub byte_len: u64,
    pub uncompressed_byte_len: u64,
    pub compression: Option<PackL2CompressionWriteReport>,
    pub outcome: PackL2WriteOutcome,
    pub eviction: PackL2EvictionReport,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2CompressionWriteReport {
    pub algorithm: String,
    pub dictionary_id: Option<String>,
    pub compressed_bytes: u64,
    pub uncompressed_bytes: u64,
    pub compression_latency_ms: u64,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2CompressionDictionary {
    pub id: String,
    pub byte_hash: String,
    pub bytes: Vec<u8>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum PackL2WriteOutcome {
    Stored,
    SkippedTooLarge { max_entry_bytes: u64 },
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct PackL2EvictionReport {
    pub removed: u64,
    pub skipped: u64,
    pub bytes_before: u64,
    pub bytes_removed: u64,
    pub bytes_after: u64,
}

#[derive(Debug)]
pub enum PackL2CacheError {
    Io {
        path: PathBuf,
        operation: &'static str,
        source: io::Error,
    },
    Json {
        path: PathBuf,
        operation: &'static str,
        source: serde_json::Error,
    },
    Compression {
        operation: &'static str,
        source: io::Error,
    },
    TimeBeforeUnixEpoch {
        source: std::time::SystemTimeError,
    },
}

impl fmt::Display for PackL2CacheError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Io {
                path,
                operation,
                source,
            } => write!(
                formatter,
                "failed to {operation} pack L2 cache path {}: {source}",
                path.display()
            ),
            Self::Json {
                path,
                operation,
                source,
            } => write!(
                formatter,
                "failed to {operation} pack L2 cache JSON at {}: {source}",
                path.display()
            ),
            Self::Compression { operation, source } => {
                write!(
                    formatter,
                    "failed to {operation} pack L2 cache entry: {source}"
                )
            }
            Self::TimeBeforeUnixEpoch { source } => {
                write!(formatter, "system time predates Unix epoch: {source}")
            }
        }
    }
}

impl std::error::Error for PackL2CacheError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io { source, .. } => Some(source),
            Self::Json { source, .. } => Some(source),
            Self::Compression { source, .. } => Some(source),
            Self::TimeBeforeUnixEpoch { source } => Some(source),
        }
    }
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PackL2CacheEntryEnvelope {
    schema: String,
    stored_at_epoch_seconds: u64,
}

#[derive(Debug)]
struct DecodedPackL2CacheEntry {
    key: String,
    stored_at_epoch_seconds: u64,
    pack_json: JsonValue,
    compression: Option<PackL2CompressionHit>,
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct PackL2CacheEntry {
    schema: String,
    key: String,
    stored_at_epoch_seconds: u64,
    pack_json: JsonValue,
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct PackL2CacheEntryV2 {
    schema: String,
    key: String,
    stored_at_epoch_seconds: u64,
    compression: PackL2CacheCompressionPayload,
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct PackL2CacheCompressionPayload {
    algorithm: String,
    compressed_payload_base64: String,
    compressed_byte_len: u64,
    uncompressed_byte_len: u64,
    uncompressed_hash: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    dictionary: Option<PackL2CacheCompressionDictionaryRef>,
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct PackL2CacheCompressionDictionaryRef {
    dictionary_id: String,
    dictionary_byte_hash: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    dictionary_bytes_base64: Option<String>,
}

impl PackL2CacheCompressionDictionaryRef {
    fn from_dictionary(dictionary: &PackL2CompressionDictionary) -> Self {
        Self {
            dictionary_id: dictionary.id.clone(),
            dictionary_byte_hash: dictionary.byte_hash.clone(),
            dictionary_bytes_base64: Some(BASE64_STANDARD.encode(&dictionary.bytes)),
        }
    }
}

#[derive(Debug)]
struct EvictionCandidate {
    path: PathBuf,
    byte_len: u64,
    stored_epoch_seconds: u64,
    last_used_epoch_seconds: u64,
    expired: bool,
}

fn remove_cache_entry_best_effort(path: &Path) {
    match fs::remove_file(path) {
        Ok(()) => {}
        Err(error) if error.kind() == io::ErrorKind::NotFound => {}
        Err(_) => {}
    }
}

fn remove_eviction_candidate_file(
    candidate: &EvictionCandidate,
    report: &mut PackL2EvictionReport,
    bytes_current: &mut u64,
) {
    match fs::remove_file(&candidate.path) {
        Ok(()) => record_eviction_candidate_removed(candidate, report, bytes_current),
        Err(error) if error.kind() == io::ErrorKind::NotFound => {
            record_eviction_candidate_removed(candidate, report, bytes_current);
        }
        Err(_) => {
            report.skipped = report.skipped.saturating_add(1);
        }
    }
}

fn record_eviction_candidate_removed(
    candidate: &EvictionCandidate,
    report: &mut PackL2EvictionReport,
    bytes_current: &mut u64,
) {
    report.removed = report.removed.saturating_add(1);
    report.bytes_removed = report.bytes_removed.saturating_add(candidate.byte_len);
    *bytes_current = bytes_current.saturating_sub(candidate.byte_len);
}

fn merge_cache_cleanup_reports(
    duplicate_cleanup: PackL2EvictionReport,
    eviction: PackL2EvictionReport,
) -> PackL2EvictionReport {
    PackL2EvictionReport {
        removed: duplicate_cleanup.removed.saturating_add(eviction.removed),
        skipped: duplicate_cleanup.skipped.saturating_add(eviction.skipped),
        bytes_before: eviction
            .bytes_before
            .saturating_add(duplicate_cleanup.bytes_removed),
        bytes_removed: duplicate_cleanup
            .bytes_removed
            .saturating_add(eviction.bytes_removed),
        bytes_after: eviction.bytes_after,
    }
}

fn decode_pack_l2_cache_entry(
    bytes: &[u8],
    max_decompressed_entry_bytes: u64,
) -> Result<DecodedPackL2CacheEntry, PackL2CacheMissReason> {
    let envelope = serde_json::from_slice::<PackL2CacheEntryEnvelope>(bytes)
        .map_err(|error| PackL2CacheMissReason::Corrupt(error.to_string()))?;
    match envelope.schema.as_str() {
        PACK_L2_CACHE_ENTRY_SCHEMA_V1 => {
            let entry = serde_json::from_slice::<PackL2CacheEntry>(bytes)
                .map_err(|error| PackL2CacheMissReason::Corrupt(error.to_string()))?;
            Ok(DecodedPackL2CacheEntry {
                key: entry.key,
                stored_at_epoch_seconds: entry.stored_at_epoch_seconds,
                pack_json: entry.pack_json,
                compression: None,
            })
        }
        PACK_L2_CACHE_ENTRY_SCHEMA_V2 => {
            decode_compressed_pack_l2_cache_entry(bytes, max_decompressed_entry_bytes)
        }
        schema => Err(PackL2CacheMissReason::Corrupt(format!(
            "unexpected schema {schema}"
        ))),
    }
}

fn decode_compressed_pack_l2_cache_entry(
    bytes: &[u8],
    max_decompressed_entry_bytes: u64,
) -> Result<DecodedPackL2CacheEntry, PackL2CacheMissReason> {
    let entry = serde_json::from_slice::<PackL2CacheEntryV2>(bytes)
        .map_err(|error| PackL2CacheMissReason::Corrupt(error.to_string()))?;
    if entry.compression.algorithm != PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1 {
        return Err(PackL2CacheMissReason::Corrupt(format!(
            "unsupported compression algorithm {}",
            entry.compression.algorithm
        )));
    }
    let compressed = BASE64_STANDARD
        .decode(&entry.compression.compressed_payload_base64)
        .map_err(|error| PackL2CacheMissReason::CompressionDecode {
            message: format!("compressed payload is not base64: {error}"),
        })?;
    if compressed.len() as u64 != entry.compression.compressed_byte_len {
        return Err(PackL2CacheMissReason::CompressionDecode {
            message: format!(
                "compressed byte length mismatch expected={} actual={}",
                entry.compression.compressed_byte_len,
                compressed.len()
            ),
        });
    }
    let dictionary_bytes = decode_pack_l2_dictionary_bytes(entry.compression.dictionary.as_ref())?;
    if entry.compression.uncompressed_byte_len > max_decompressed_entry_bytes {
        return Err(PackL2CacheMissReason::CompressionDecode {
            message: format!(
                "uncompressed byte length {} exceeds the {max_decompressed_entry_bytes}-byte decompression cap",
                entry.compression.uncompressed_byte_len
            ),
        });
    }
    let capacity = usize::try_from(entry.compression.uncompressed_byte_len).map_err(|_| {
        PackL2CacheMissReason::CompressionDecode {
            message: format!(
                "uncompressed byte length does not fit usize: {}",
                entry.compression.uncompressed_byte_len
            ),
        }
    })?;
    let decompress_start = Instant::now();
    let uncompressed = zstd_decompress(&compressed, capacity, dictionary_bytes.as_deref())
        .map_err(|source| PackL2CacheMissReason::CompressionDecode {
            message: source.to_string(),
        })?;
    let decompression_latency_ms = elapsed_millis(decompress_start.elapsed());
    if uncompressed.len() as u64 != entry.compression.uncompressed_byte_len {
        return Err(PackL2CacheMissReason::CompressionDecode {
            message: format!(
                "uncompressed byte length mismatch expected={} actual={}",
                entry.compression.uncompressed_byte_len,
                uncompressed.len()
            ),
        });
    }
    let actual_hash = blake3_hash(&uncompressed);
    if actual_hash != entry.compression.uncompressed_hash {
        return Err(PackL2CacheMissReason::CompressionDecode {
            message: format!(
                "uncompressed hash mismatch expected={} actual={actual_hash}",
                entry.compression.uncompressed_hash
            ),
        });
    }
    let pack_json = serde_json::from_slice::<JsonValue>(&uncompressed).map_err(|error| {
        PackL2CacheMissReason::CompressionDecode {
            message: format!("decompressed pack JSON is malformed: {error}"),
        }
    })?;
    Ok(DecodedPackL2CacheEntry {
        key: entry.key,
        stored_at_epoch_seconds: entry.stored_at_epoch_seconds,
        pack_json,
        compression: Some(PackL2CompressionHit {
            algorithm: entry.compression.algorithm,
            dictionary_id: entry
                .compression
                .dictionary
                .map(|dictionary| dictionary.dictionary_id),
            compressed_bytes: compressed.len() as u64,
            uncompressed_bytes: uncompressed.len() as u64,
            decompression_latency_ms,
        }),
    })
}

fn decode_pack_l2_dictionary_bytes(
    dictionary: Option<&PackL2CacheCompressionDictionaryRef>,
) -> Result<Option<Vec<u8>>, PackL2CacheMissReason> {
    let Some(dictionary) = dictionary else {
        return Ok(None);
    };
    let Some(encoded) = &dictionary.dictionary_bytes_base64 else {
        return Err(PackL2CacheMissReason::CompressionDictionaryMissing {
            dictionary_id: dictionary.dictionary_id.clone(),
        });
    };
    let bytes = BASE64_STANDARD.decode(encoded).map_err(|error| {
        PackL2CacheMissReason::CompressionDictionaryCorrupt {
            dictionary_id: dictionary.dictionary_id.clone(),
            message: format!("dictionary bytes are not base64: {error}"),
        }
    })?;
    let actual_hash = blake3_hash(&bytes);
    if actual_hash != dictionary.dictionary_byte_hash {
        return Err(PackL2CacheMissReason::CompressionDictionaryCorrupt {
            dictionary_id: dictionary.dictionary_id.clone(),
            message: format!(
                "dictionary byte hash mismatch expected={} actual={actual_hash}",
                dictionary.dictionary_byte_hash
            ),
        });
    }
    Ok(Some(bytes))
}

fn zstd_compress(
    payload: &[u8],
    dictionary: Option<&PackL2CompressionDictionary>,
) -> Result<Vec<u8>, PackL2CacheError> {
    let mut compressor = match dictionary {
        Some(dictionary) => {
            zstd::bulk::Compressor::with_dictionary(PACK_L2_COMPRESSION_LEVEL, &dictionary.bytes)
        }
        None => zstd::bulk::Compressor::new(PACK_L2_COMPRESSION_LEVEL),
    }
    .map_err(|source| PackL2CacheError::Compression {
        operation: "initialize_compressor",
        source,
    })?;
    compressor
        .compress(payload)
        .map_err(|source| PackL2CacheError::Compression {
            operation: "compress",
            source,
        })
}

fn zstd_decompress(
    payload: &[u8],
    capacity: usize,
    dictionary: Option<&[u8]>,
) -> io::Result<Vec<u8>> {
    let mut decompressor = match dictionary {
        Some(dictionary) => zstd::bulk::Decompressor::with_dictionary(dictionary)?,
        None => zstd::bulk::Decompressor::new()?,
    };
    decompressor.decompress(payload, capacity)
}

fn elapsed_millis(duration: Duration) -> u64 {
    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}

fn cache_file_name(key: &str) -> String {
    format!("{}.json", cache_file_stem(key))
}

fn cache_file_name_with_body_hash(key: &str, body_hash_prefix: &str) -> String {
    format!("{}.{}.json", cache_file_stem(key), body_hash_prefix)
}

fn cache_file_stem(key: &str) -> String {
    blake3::hash(key.as_bytes()).to_hex().to_string()
}

fn body_hash_prefix(bytes: &[u8]) -> String {
    blake3::hash(bytes).to_hex()[..16].to_owned()
}

fn blake3_hash(bytes: &[u8]) -> String {
    format!("blake3:{}", blake3::hash(bytes).to_hex())
}

fn body_hashed_file_name_matches(file_name: &str, key_stem: &str) -> bool {
    let Some(rest) = file_name.strip_prefix(key_stem) else {
        return false;
    };
    let Some(body_hash_prefix) = rest
        .strip_prefix('.')
        .and_then(|rest| rest.strip_suffix(".json"))
    else {
        return false;
    };
    body_hash_prefix.len() == 16
        && body_hash_prefix
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit())
}

fn body_hash_prefix_from_path(path: &Path) -> Option<String> {
    let file_name = path.file_name()?.to_str()?;
    let body_hash_prefix = file_name
        .strip_suffix(".json")?
        .rsplit_once('.')?
        .1
        .to_owned();
    (body_hash_prefix.len() == 16
        && body_hash_prefix
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit()))
    .then_some(body_hash_prefix)
}

fn is_expired(stored_at_epoch_seconds: u64, now_epoch_seconds: u64, max_age: Duration) -> bool {
    now_epoch_seconds.saturating_sub(stored_at_epoch_seconds) > max_age.as_secs()
}

fn system_time_seconds(time: SystemTime) -> Result<u64, PackL2CacheError> {
    time.duration_since(UNIX_EPOCH)
        .map(|duration| duration.as_secs())
        .map_err(|source| PackL2CacheError::TimeBeforeUnixEpoch { source })
}

fn touch_cache_entry_mtime_best_effort(path: &Path, epoch_seconds: u64) {
    let modified_at = UNIX_EPOCH + Duration::from_secs(epoch_seconds);
    let times = FileTimes::new().set_modified(modified_at);
    if let Ok(file) = open_cache_entry_file_for_touch(path) {
        let _ = file.set_times(times);
    }
}

fn cache_entry_stored_at(path: &Path, max_entry_bytes: u64) -> Option<u64> {
    if first_existing_symlink_component(path)
        .ok()
        .flatten()
        .is_some()
    {
        return None;
    }
    // Same cap-on-read defense as `read_cache_entry_file`. Eviction
    // scans call this for every `.json` in the cache root (line 458),
    // so a single corrupted oversized entry would otherwise pin a
    // proportional allocation per pass.
    let bytes = read_cache_entry_file(path, max_entry_bytes).ok()?;
    serde_json::from_slice::<PackL2CacheEntryEnvelope>(&bytes)
        .ok()
        .map(|entry| entry.stored_at_epoch_seconds)
}

fn cache_entry_preference_epoch_seconds(path: &Path, max_entry_bytes: u64) -> u64 {
    cache_entry_stored_at(path, max_entry_bytes)
        .or_else(|| {
            fs::symlink_metadata(path)
                .ok()
                .and_then(|metadata| metadata.modified().ok())
                .and_then(|modified| system_time_seconds(modified).ok())
        })
        .unwrap_or(0)
}

fn ensure_cache_dir(path: &Path) -> Result<(), PackL2CacheError> {
    ensure_no_symlink_components(path, "inspect_root")?;
    fs::create_dir_all(path).map_err(|source| PackL2CacheError::Io {
        path: path.to_path_buf(),
        operation: "create_dir_all",
        source,
    })?;
    ensure_no_symlink_components(path, "inspect_root")?;
    #[cfg(unix)]
    fs::set_permissions(path, fs::Permissions::from_mode(0o700)).map_err(|source| {
        PackL2CacheError::Io {
            path: path.to_path_buf(),
            operation: "set_permissions",
            source,
        }
    })?;
    Ok(())
}

fn ensure_no_symlink_components(
    path: &Path,
    operation: &'static str,
) -> Result<(), PackL2CacheError> {
    if let Some(symlink_path) =
        first_existing_symlink_component(path).map_err(|source| PackL2CacheError::Io {
            path: path.to_path_buf(),
            operation,
            source,
        })?
    {
        return Err(PackL2CacheError::Io {
            path: path.to_path_buf(),
            operation,
            source: io::Error::new(
                io::ErrorKind::PermissionDenied,
                format!(
                    "pack L2 cache path traverses symbolic link {}",
                    symlink_path.display()
                ),
            ),
        });
    }
    Ok(())
}

fn first_existing_symlink_component(path: &Path) -> io::Result<Option<PathBuf>> {
    match crate::core::path_safety::first_existing_symlink_component(path) {
        // A regular-file (non-directory) component terminates the
        // existing-prefix scan: nothing deeper can exist, and any symlink up
        // to that component was already detected without following it. The
        // subsequent filesystem operation reports the honest failure for the
        // unusable path; every other error kind still propagates.
        Err(error) if error.kind() == io::ErrorKind::NotADirectory => Ok(None),
        other => other,
    }
}

struct TempPackL2FileGuard<'a> {
    path: &'a Path,
    armed: bool,
}

impl<'a> TempPackL2FileGuard<'a> {
    fn disarmed(path: &'a Path) -> Self {
        Self { path, armed: false }
    }

    fn arm(&mut self) {
        self.armed = true;
    }

    fn disarm(&mut self) {
        self.armed = false;
    }
}

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

fn write_synced_file(path: &Path, bytes: &[u8]) -> Result<(), PackL2CacheError> {
    let mut cleanup_guard = TempPackL2FileGuard::disarmed(path);
    let mut file =
        open_cache_temp_file_for_create(path).map_err(|source| PackL2CacheError::Io {
            path: path.to_path_buf(),
            operation: "open_temp",
            source,
        })?;
    cleanup_guard.arm();
    file.write_all(bytes)
        .and_then(|()| file.sync_all())
        .map_err(|source| PackL2CacheError::Io {
            path: path.to_path_buf(),
            operation: "write_sync",
            source,
        })?;
    // Apply 0o600 via the open file descriptor (`File::set_permissions`
    // → `fchmod`) rather than `fs::set_permissions(path, ...)` →
    // `chmod`. The prior path-based shape opened a TOCTOU window
    // between the O_CREAT|O_EXCL|O_NOFOLLOW `open_cache_temp_file_for_create`
    // call above and this chmod: a peer with write access to the cache
    // directory could `unlink(path); symlink("/target", path)` between
    // the two syscalls, and `chmod` would follow the symlink and
    // tighten permissions on `/target` instead. `fchmod` operates on the
    // already-open fd, so the symlink swap on the path cannot redirect
    // the metadata change. The exploit window is narrow (between
    // `open_cache_temp_file_for_create` and this call) and the practical
    // impact is bounded by the running user's chown rights, but the
    // race is real and the fix is mechanical. Same defense the init
    // hardening pass (edd17760) routed through `rustix::fs::fchmod`.
    #[cfg(unix)]
    file.set_permissions(fs::Permissions::from_mode(0o600))
        .map_err(|source| PackL2CacheError::Io {
            path: path.to_path_buf(),
            operation: "set_file_permissions",
            source,
        })?;
    cleanup_guard.disarm();
    Ok(())
}

fn publish_cache_entry_temp_file(temp_path: &Path, path: &Path) -> Result<(), PackL2CacheError> {
    ensure_no_symlink_components(path, "inspect_entry")?;
    ensure_no_symlink_components(temp_path, "inspect_temp")?;
    ensure_cache_temp_path_is_regular(temp_path)?;
    fs::rename(temp_path, path).map_err(|source| {
        let _ = fs::remove_file(temp_path);
        PackL2CacheError::Io {
            path: path.to_path_buf(),
            operation: "rename",
            source,
        }
    })
}

fn ensure_cache_temp_path_is_regular(path: &Path) -> Result<(), PackL2CacheError> {
    match fs::symlink_metadata(path) {
        Ok(metadata) if metadata.file_type().is_file() => Ok(()),
        Ok(_) => Err(PackL2CacheError::Io {
            path: path.to_path_buf(),
            operation: "inspect_temp",
            source: io::Error::new(
                io::ErrorKind::InvalidInput,
                "pack L2 cache temp path is not a regular file",
            ),
        }),
        Err(source) => Err(PackL2CacheError::Io {
            path: path.to_path_buf(),
            operation: "inspect_temp",
            source,
        }),
    }
}

fn sync_directory(path: &Path) -> Result<(), PackL2CacheError> {
    open_cache_directory_for_sync(path)
        .and_then(|directory| directory.sync_all())
        .map_err(|source| PackL2CacheError::Io {
            path: path.to_path_buf(),
            operation: "sync_dir",
            source,
        })
}

fn read_cache_entry_file(path: &Path, max_entry_bytes: u64) -> io::Result<Vec<u8>> {
    let file = open_cache_entry_file_for_read(path)?;
    let mut bytes = Vec::new();
    // Cap the read at `max_entry_bytes + 1`. The post-read size check
    // in `get_candidate_at` (line 174) rejects entries whose byte_len
    // exceeds `max_entry_bytes`, but the prior `read_to_end` (uncapped)
    // would already have pre-sized the buffer from the file's metadata
    // length BEFORE that check ran — so a peer that swapped a
    // legitimate ≤1 MiB entry for a multi-GiB regular file between
    // `put_at` and the next `get_at` would force a multi-GiB
    // allocation, then trip the post-read cap and treat it as a miss.
    // Pinning the read to `cap + 1` bytes makes the worst case
    // proportional to the configured cap regardless of on-disk file
    // size. The `+ 1` sentinel preserves the existing semantics: an
    // entry of *exactly* `max_entry_bytes` still parses (the read
    // captures `cap` bytes), and the post-read check at line 174
    // distinguishes "exactly at cap" (`byte_len == max_entry_bytes`,
    // accepted) from "above cap" (`byte_len == max_entry_bytes + 1`,
    // rejected as `TooLarge`). Same defensive pattern as
    // `read_limited_utf8_file` in src/hooks/installer.rs (5a4eeab4 /
    // 4f36dfa8) and the metadata-bound reads added by Round-1 fixes
    // in src/core/preflight.rs (aac04adb) and src/core/handoff.rs
    // (6d8d00e5).
    file.take(max_entry_bytes.saturating_add(1))
        .read_to_end(&mut bytes)?;
    Ok(bytes)
}

fn open_cache_entry_file_for_read(path: &Path) -> io::Result<File> {
    let mut options = OpenOptions::new();
    options.read(true);
    configure_pack_l2_open_no_follow(&mut options);
    options.open(path)
}

fn open_cache_temp_file_for_create(path: &Path) -> io::Result<File> {
    let mut options = OpenOptions::new();
    options.write(true).create_new(true);
    configure_pack_l2_open_no_follow(&mut options);
    options.open(path)
}

fn open_cache_entry_file_for_touch(path: &Path) -> io::Result<File> {
    let mut options = OpenOptions::new();
    options.write(true);
    configure_pack_l2_open_no_follow(&mut options);
    options.open(path)
}

fn open_cache_directory_for_sync(path: &Path) -> io::Result<File> {
    let mut options = OpenOptions::new();
    options.read(true);
    configure_pack_l2_open_no_follow(&mut options);
    options.open(path)
}

#[cfg(all(unix, not(any(target_os = "espidf", target_os = "horizon"))))]
fn configure_pack_l2_open_no_follow(options: &mut OpenOptions) {
    use std::os::unix::fs::OpenOptionsExt;

    options.custom_flags(rustix::fs::OFlags::NOFOLLOW.bits() as i32);
}

#[cfg(not(all(unix, not(any(target_os = "espidf", target_os = "horizon")))))]
fn configure_pack_l2_open_no_follow(_options: &mut OpenOptions) {}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    type TestResult = Result<(), String>;

    fn cache(
        max_bytes: u64,
        max_age: Duration,
    ) -> Result<(tempfile::TempDir, PackL2Cache), String> {
        cache_with_options(PackL2CacheOptions::new(max_bytes, max_age))
    }

    fn cache_with_options(
        options: PackL2CacheOptions,
    ) -> Result<(tempfile::TempDir, PackL2Cache), String> {
        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let cache = PackL2Cache::new(temp.path().join("pack-l2"), options);
        Ok((temp, cache))
    }

    fn hit_json(lookup: PackL2CacheLookup) -> Result<JsonValue, String> {
        match lookup {
            PackL2CacheLookup::Hit(hit) => Ok(hit.pack_json),
            PackL2CacheLookup::Miss(miss) => {
                Err(format!("expected hit, got miss: {:?}", miss.reason))
            }
        }
    }

    fn raw_entry_bytes(
        key: &str,
        pack_json: JsonValue,
        stored_at_epoch_seconds: u64,
    ) -> Result<Vec<u8>, String> {
        serde_json::to_vec(&PackL2CacheEntry {
            schema: PACK_L2_CACHE_ENTRY_SCHEMA_V1.to_owned(),
            key: key.to_owned(),
            stored_at_epoch_seconds,
            pack_json,
        })
        .map_err(|error| error.to_string())
    }

    fn write_raw_entry(cache: &PackL2Cache, key: &str, bytes: &[u8]) -> Result<PathBuf, String> {
        ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;
        let path = cache.entry_path_for_body_hash(key, &body_hash_prefix(bytes));
        fs::write(&path, bytes).map_err(|error| error.to_string())?;
        Ok(path)
    }

    fn raw_compressed_entry_bytes(
        key: &str,
        payload: PackL2CacheCompressionPayload,
        stored_at_epoch_seconds: u64,
    ) -> Result<Vec<u8>, String> {
        serde_json::to_vec(&PackL2CacheEntryV2 {
            schema: PACK_L2_CACHE_ENTRY_SCHEMA_V2.to_owned(),
            key: key.to_owned(),
            stored_at_epoch_seconds,
            compression: payload,
        })
        .map_err(|error| error.to_string())
    }

    fn modified_epoch_seconds(path: &Path) -> Result<u64, String> {
        let modified = fs::metadata(path)
            .map_err(|error| error.to_string())?
            .modified()
            .map_err(|error| error.to_string())?;
        system_time_seconds(modified).map_err(|error| error.to_string())
    }

    #[test]
    fn temp_pack_l2_file_guard_removes_armed_orphan() -> TestResult {
        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let path = temp.path().join("entry.tmp");
        fs::write(&path, b"orphan").map_err(|error| error.to_string())?;
        {
            let mut guard = TempPackL2FileGuard::disarmed(&path);
            guard.arm();
        }

        assert!(
            !path.exists(),
            "armed pack L2 temp guard should remove an owned orphan"
        );
        Ok(())
    }

    #[test]
    fn temp_pack_l2_file_guard_disarm_preserves_success_temp() -> TestResult {
        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let path = temp.path().join("entry.tmp");
        fs::write(&path, b"ready-to-publish").map_err(|error| error.to_string())?;
        {
            let mut guard = TempPackL2FileGuard::disarmed(&path);
            guard.arm();
            guard.disarm();
        }

        assert!(
            path.exists(),
            "disarmed pack L2 temp guard must preserve the successful temp file"
        );
        assert_eq!(
            fs::read(&path).map_err(|error| error.to_string())?,
            b"ready-to-publish",
            "disarmed guard must not change temp file contents"
        );
        Ok(())
    }

    #[test]
    fn temp_pack_l2_file_guard_unarmed_ignores_missing_path() -> TestResult {
        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let path = temp.path().join("never-created.tmp");
        {
            let _guard = TempPackL2FileGuard::disarmed(&path);
        }

        assert!(
            !path.exists(),
            "unarmed pack L2 temp guard must not create or remove a missing path"
        );
        Ok(())
    }

    #[test]
    fn default_options_use_pass2_size_limits() {
        let options = PackL2CacheOptions::default();

        assert_eq!(
            options.max_bytes, DEFAULT_MAX_BYTES,
            "default cache cap should stay at the pass-2 256 MiB budget"
        );
        assert_eq!(
            options.max_entry_bytes, DEFAULT_MAX_ENTRY_BYTES,
            "default per-entry cap should keep pathological packs out of L2"
        );
    }

    #[test]
    fn happy_path_roundtrip_returns_stored_pack_json() -> TestResult {
        let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
        let pack = json!({"hash": "blake3:test", "items": [{"id": "mem_1"}]});

        let report = cache
            .put_at("blake3:key-a", &pack, 100)
            .map_err(|error| error.to_string())?;
        assert!(
            report.path.exists(),
            "write should publish final cache file"
        );
        let file_name = report
            .path
            .file_name()
            .and_then(|file_name| file_name.to_str())
            .ok_or_else(|| "cache path should have a UTF-8 file name".to_owned())?;
        let parts = file_name
            .strip_suffix(".json")
            .ok_or_else(|| format!("cache file should end in .json: {file_name}"))?
            .split('.')
            .collect::<Vec<_>>();
        assert_eq!(parts.len(), 2, "cache file should have key and body hash");
        assert_eq!(parts[0].len(), 64, "key hash should be full BLAKE3 hex");
        assert_eq!(parts[1].len(), 16, "body hash should be truncated hex");
        assert!(
            body_hashed_file_name_matches(file_name, &cache_file_stem("blake3:key-a")),
            "cache file name should bind key hash and body hash"
        );

        let stored = hit_json(
            cache
                .get_at("blake3:key-a", 120)
                .map_err(|error| error.to_string())?,
        )?;
        assert_eq!(stored, pack, "cache hit should preserve pack JSON exactly");
        Ok(())
    }

    #[test]
    fn repeated_writes_return_newest_stored_entry_not_lexical_first() -> TestResult {
        let (_temp, cache) = cache(10_000, Duration::from_secs(10_000))?;
        let key = "blake3:repeat-shadow";
        let old_pack = json!({"payload": "old"});
        let new_pack = json!({"payload": "new"});

        let old_report = cache
            .put_at(key, &old_pack, 100)
            .map_err(|error| error.to_string())?;
        let new_report = cache
            .put_at(key, &new_pack, 200)
            .map_err(|error| error.to_string())?;
        assert!(
            old_report.path < new_report.path,
            "fixture should cover the old filename-ordering bug"
        );

        let lookup = cache.get_at(key, 210).map_err(|error| error.to_string())?;

        match lookup {
            PackL2CacheLookup::Hit(hit) => {
                assert_eq!(hit.path, new_report.path);
                assert_eq!(hit.stored_at_epoch_seconds, 200);
                assert_eq!(hit.pack_json, new_pack);
            }
            PackL2CacheLookup::Miss(miss) => {
                return Err(format!("newest duplicate-key entry should hit: {miss:?}"));
            }
        }
        Ok(())
    }

    #[test]
    fn same_second_compressed_rewrite_prunes_stale_same_key_entry() -> TestResult {
        let (_temp, cache) = cache(10_000, Duration::from_secs(10_000))?;
        let key = "blake3:same-second-repeat";
        let old_pack = json!({"payload": "old"});
        let new_pack = json!({"payload": "new"});

        let old_report = cache
            .put_compressed_at(key, &old_pack, 100)
            .map_err(|error| error.to_string())?;
        let new_report = cache
            .put_compressed_at(key, &new_pack, 100)
            .map_err(|error| error.to_string())?;

        assert_ne!(
            old_report.path, new_report.path,
            "different same-second payloads should publish distinct body-hashed entries"
        );
        assert!(
            !old_report.path.exists(),
            "a same-second rewrite must remove the prior same-key entry"
        );
        assert!(
            new_report.path.exists(),
            "same-second rewrite must retain the newly published entry"
        );
        assert_eq!(
            new_report.eviction.removed, 1,
            "same-key cleanup should be counted in the write cleanup report"
        );

        let lookup = cache.get_at(key, 100).map_err(|error| error.to_string())?;
        match lookup {
            PackL2CacheLookup::Hit(hit) => {
                assert_eq!(hit.path, new_report.path);
                assert_eq!(hit.stored_at_epoch_seconds, 100);
                assert_eq!(hit.pack_json, new_pack);
            }
            PackL2CacheLookup::Miss(miss) => {
                return Err(format!(
                    "same-second rewrite should hit the retained entry: {miss:?}"
                ));
            }
        }
        Ok(())
    }

    #[test]
    fn compressed_v2_roundtrip_returns_stored_pack_json() -> TestResult {
        let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
        let pack = json!({
            "schema": "ee.pack_l2.test_payload.v1",
            "responseJson": "{\"schema\":\"ee.response.v2\",\"success\":true,\"data\":{\"items\":[\"mem_1\"]}}"
        });

        let report = cache
            .put_compressed_at("blake3:compressed-key", &pack, 100)
            .map_err(|error| error.to_string())?;

        assert_eq!(report.outcome, PackL2WriteOutcome::Stored);
        let compression = report
            .compression
            .as_ref()
            .ok_or_else(|| "compressed write should report compression metadata".to_owned())?;
        assert_eq!(compression.algorithm, PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1);
        assert!(compression.compressed_bytes > 0);
        assert!(compression.uncompressed_bytes > 0);
        assert_eq!(report.uncompressed_byte_len, compression.uncompressed_bytes);

        let lookup = cache
            .get_at("blake3:compressed-key", 120)
            .map_err(|error| error.to_string())?;
        match lookup {
            PackL2CacheLookup::Hit(hit) => {
                assert_eq!(hit.pack_json, pack);
                let hit_compression = hit.compression.ok_or_else(|| {
                    "compressed hit should report compression metadata".to_owned()
                })?;
                assert_eq!(
                    hit_compression.algorithm,
                    PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1
                );
                assert_eq!(
                    hit_compression.compressed_bytes,
                    compression.compressed_bytes
                );
                assert_eq!(
                    hit_compression.uncompressed_bytes,
                    compression.uncompressed_bytes
                );
            }
            PackL2CacheLookup::Miss(miss) => {
                return Err(format!("compressed v2 entry should hit: {miss:?}"));
            }
        }
        Ok(())
    }

    #[test]
    fn empty_or_boundary_entry_at_exactly_max_entry_bytes_is_cached() -> TestResult {
        let key = "blake3:exact-entry-cap";
        let stored_at_epoch_seconds = 100;
        let pack = json!({"hash": "entry-cap", "items": [{"id": "mem_exact"}]});
        let entry_len = raw_entry_bytes(key, pack.clone(), stored_at_epoch_seconds)?.len() as u64;
        let (_temp, cache) = cache_with_options(
            PackL2CacheOptions::new(4096, Duration::from_secs(60)).with_max_entry_bytes(entry_len),
        )?;

        let report = cache
            .put_at(key, &pack, stored_at_epoch_seconds)
            .map_err(|error| error.to_string())?;

        assert_eq!(report.byte_len, entry_len);
        assert_eq!(report.outcome, PackL2WriteOutcome::Stored);
        assert!(
            report.path.exists(),
            "entry exactly at max_entry_bytes should be cached"
        );
        assert_eq!(
            hit_json(cache.get_at(key, 120).map_err(|error| error.to_string())?)?,
            pack
        );
        Ok(())
    }

    #[test]
    fn compressed_v2_entry_at_max_entry_bytes_plus_one_is_skipped_with_event() -> TestResult {
        let key = "blake3:compressed-oversized-entry";
        let stored_at_epoch_seconds = 100;
        let pack = json!({"hash": "entry-cap", "items": [{"id": "mem_oversized"}]});
        let (_temp, baseline_cache) =
            cache_with_options(PackL2CacheOptions::new(4096, Duration::from_secs(60)))?;
        let baseline_report = baseline_cache
            .put_compressed_at(key, &pack, stored_at_epoch_seconds)
            .map_err(|error| error.to_string())?;
        let max_entry_bytes = baseline_report
            .byte_len
            .checked_sub(1)
            .ok_or_else(|| "compressed test entry should have non-zero length".to_owned())?;
        let (_temp, cache) = cache_with_options(
            PackL2CacheOptions::new(4096, Duration::from_secs(60))
                .with_max_entry_bytes(max_entry_bytes),
        )?;

        let report = cache
            .put_compressed_at(key, &pack, stored_at_epoch_seconds)
            .map_err(|error| error.to_string())?;

        assert_eq!(report.byte_len, baseline_report.byte_len);
        assert_eq!(
            report.outcome,
            PackL2WriteOutcome::SkippedTooLarge { max_entry_bytes }
        );
        assert!(
            report.compression.is_some(),
            "skipped compressed entries should still report compression accounting"
        );
        assert!(
            !report.path.exists(),
            "oversized compressed entries should not publish a cache file"
        );
        Ok(())
    }

    #[test]
    fn empty_or_boundary_entry_at_max_entry_bytes_plus_one_is_skipped_with_event() -> TestResult {
        let key = "blake3:oversized-entry";
        let stored_at_epoch_seconds = 100;
        let pack = json!({"hash": "entry-cap", "items": [{"id": "mem_oversized"}]});
        let entry_len = raw_entry_bytes(key, pack, stored_at_epoch_seconds)?.len() as u64;
        let max_entry_bytes = entry_len
            .checked_sub(1)
            .ok_or_else(|| "test entry should have non-zero serialized length".to_owned())?;
        let (_temp, cache) = cache_with_options(
            PackL2CacheOptions::new(4096, Duration::from_secs(60))
                .with_max_entry_bytes(max_entry_bytes),
        )?;

        let report = cache
            .put_at(
                key,
                &json!({"hash": "entry-cap", "items": [{"id": "mem_oversized"}]}),
                stored_at_epoch_seconds,
            )
            .map_err(|error| error.to_string())?;

        assert_eq!(report.byte_len, entry_len);
        assert_eq!(
            report.outcome,
            PackL2WriteOutcome::SkippedTooLarge { max_entry_bytes }
        );
        assert_eq!(
            report.eviction,
            PackL2EvictionReport::default(),
            "skipped entries should not run write-through eviction"
        );
        assert!(
            !report.path.exists(),
            "oversized entries should not publish a cache file"
        );
        assert!(
            matches!(
                cache.get_at(key, 120).map_err(|error| error.to_string())?,
                PackL2CacheLookup::Miss(PackL2CacheMiss {
                    reason: PackL2CacheMissReason::NotFound,
                    ..
                })
            ),
            "skipped oversized entries should behave like cold misses"
        );
        Ok(())
    }

    #[test]
    fn happy_path_touch_on_read_advances_mtime() -> TestResult {
        let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
        let report = cache
            .put_at("blake3:touch", &json!({"hash": "mtime"}), 100)
            .map_err(|error| error.to_string())?;
        let mtime_before = modified_epoch_seconds(&report.path)?;
        assert_eq!(
            mtime_before, 100,
            "write path should seed mtime from the stored-at timestamp"
        );

        let lookup = cache
            .get_at("blake3:touch", 150)
            .map_err(|error| error.to_string())?;

        assert!(
            lookup.is_hit(),
            "fresh entry should hit before touching mtime"
        );
        assert!(
            modified_epoch_seconds(&report.path)? >= 150,
            "read path should advance mtime for portable LRU accounting"
        );
        Ok(())
    }

    #[test]
    fn empty_or_boundary_missing_key_returns_not_found_miss() -> TestResult {
        let (_temp, cache) = cache(4096, Duration::from_secs(60))?;

        let lookup = cache
            .get_at("blake3:missing", 100)
            .map_err(|error| error.to_string())?;

        assert_eq!(
            lookup,
            PackL2CacheLookup::Miss(PackL2CacheMiss {
                key: "blake3:missing".to_owned(),
                path: cache.entry_path("blake3:missing"),
                reason: PackL2CacheMissReason::NotFound,
            })
        );
        Ok(())
    }

    #[test]
    fn empty_or_boundary_expired_entry_returns_expired_miss() -> TestResult {
        let (_temp, cache) = cache(4096, Duration::from_secs(10))?;
        let report = cache
            .put_at("blake3:key-expired", &json!({"hash": "old"}), 100)
            .map_err(|error| error.to_string())?;

        let lookup = cache
            .get_at("blake3:key-expired", 111)
            .map_err(|error| error.to_string())?;

        assert_eq!(
            lookup,
            PackL2CacheLookup::Miss(PackL2CacheMiss {
                key: "blake3:key-expired".to_owned(),
                path: report.path,
                reason: PackL2CacheMissReason::Expired {
                    stored_at_epoch_seconds: 100,
                },
            })
        );
        Ok(())
    }

    #[test]
    fn error_or_invalid_oversized_existing_entry_is_removed_on_read() -> TestResult {
        let key = "blake3:old-oversized-entry";
        let pack = json!({"hash": "old-entry", "items": [{"id": "mem_old_oversized"}]});
        let bytes = raw_entry_bytes(key, pack, 100)?;
        let max_entry_bytes = (bytes.len() as u64)
            .checked_sub(1)
            .ok_or_else(|| "test entry should have non-zero serialized length".to_owned())?;
        let (_temp, cache) = cache_with_options(
            PackL2CacheOptions::new(4096, Duration::from_secs(60))
                .with_max_entry_bytes(max_entry_bytes),
        )?;
        let path = write_raw_entry(&cache, key, &bytes)?;

        let lookup = cache.get_at(key, 120).map_err(|error| error.to_string())?;

        assert_eq!(
            lookup,
            PackL2CacheLookup::Miss(PackL2CacheMiss {
                key: key.to_owned(),
                path: path.clone(),
                reason: PackL2CacheMissReason::TooLarge {
                    byte_len: bytes.len() as u64,
                    max_entry_bytes,
                },
            })
        );
        assert!(
            !path.exists(),
            "oversized legacy entries should be invalidated under the current cap"
        );
        Ok(())
    }

    #[test]
    fn error_or_invalid_corrupt_entry_returns_corrupt_miss() -> TestResult {
        let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
        let path = write_raw_entry(&cache, "blake3:corrupt", b"{not-json")?;

        let lookup = cache
            .get_at("blake3:corrupt", 100)
            .map_err(|error| error.to_string())?;

        match lookup {
            PackL2CacheLookup::Miss(miss) => {
                assert!(
                    matches!(miss.reason, PackL2CacheMissReason::Corrupt(_)),
                    "corrupt JSON should be a typed miss"
                );
            }
            PackL2CacheLookup::Hit(_) => return Err("corrupt entry must not hit".to_owned()),
        }
        assert!(
            !path.exists(),
            "corrupt cache entry should be invalidated after a typed miss"
        );
        Ok(())
    }

    #[test]
    fn error_or_invalid_corrupt_candidate_does_not_mask_valid_fallback() -> TestResult {
        let key = "blake3:multi-candidate";
        let pack = json!({"hash": "valid-fallback", "items": [{"id": "mem_valid"}]});
        let valid_bytes = raw_entry_bytes(key, pack.clone(), 100)?;
        let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
        ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;

        let corrupt_bytes = b"{not-json";
        let corrupt_path = cache.entry_path_for_body_hash(key, &body_hash_prefix(corrupt_bytes));
        fs::write(&corrupt_path, corrupt_bytes).map_err(|error| error.to_string())?;
        let valid_path = cache.entry_path(key);
        fs::write(&valid_path, valid_bytes).map_err(|error| error.to_string())?;

        let lookup = cache.get_at(key, 120).map_err(|error| error.to_string())?;

        match lookup {
            PackL2CacheLookup::Hit(hit) => {
                assert_eq!(hit.path, valid_path);
                assert_eq!(hit.pack_json, pack);
            }
            PackL2CacheLookup::Miss(miss) => {
                return Err(format!(
                    "valid fallback should hit after bad candidate: {miss:?}"
                ));
            }
        }
        assert!(
            !corrupt_path.exists(),
            "bad body-hash candidate should be invalidated before trying the valid fallback"
        );
        Ok(())
    }

    #[test]
    fn compressed_v2_missing_dictionary_returns_typed_miss_and_removes_entry() -> TestResult {
        let key = "blake3:missing-dictionary";
        let payload = PackL2CacheCompressionPayload {
            algorithm: PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1.to_owned(),
            compressed_payload_base64: BASE64_STANDARD.encode(b"not-used-before-dictionary-check"),
            compressed_byte_len: b"not-used-before-dictionary-check".len() as u64,
            uncompressed_byte_len: 128,
            uncompressed_hash: blake3_hash(b"not-present"),
            dictionary: Some(PackL2CacheCompressionDictionaryRef {
                dictionary_id: "zstd_dict_missing".to_owned(),
                dictionary_byte_hash: "blake3:missing".to_owned(),
                dictionary_bytes_base64: None,
            }),
        };
        let bytes = raw_compressed_entry_bytes(key, payload, 100)?;
        let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
        let path = write_raw_entry(&cache, key, &bytes)?;

        let lookup = cache.get_at(key, 120).map_err(|error| error.to_string())?;

        assert_eq!(
            lookup,
            PackL2CacheLookup::Miss(PackL2CacheMiss {
                key: key.to_owned(),
                path: path.clone(),
                reason: PackL2CacheMissReason::CompressionDictionaryMissing {
                    dictionary_id: "zstd_dict_missing".to_owned()
                },
            })
        );
        assert!(
            !path.exists(),
            "missing-dictionary compressed entries should be invalidated"
        );
        Ok(())
    }

    #[test]
    fn compressed_v2_corrupt_dictionary_returns_typed_miss_and_removes_entry() -> TestResult {
        let key = "blake3:corrupt-dictionary";
        let dictionary_bytes = b"dictionary bytes with the wrong recorded hash";
        let payload = PackL2CacheCompressionPayload {
            algorithm: PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1.to_owned(),
            compressed_payload_base64: BASE64_STANDARD.encode(b"not-used-before-dictionary-check"),
            compressed_byte_len: b"not-used-before-dictionary-check".len() as u64,
            uncompressed_byte_len: 128,
            uncompressed_hash: blake3_hash(b"not-present"),
            dictionary: Some(PackL2CacheCompressionDictionaryRef {
                dictionary_id: "zstd_dict_corrupt".to_owned(),
                dictionary_byte_hash: blake3_hash(b"different dictionary bytes"),
                dictionary_bytes_base64: Some(BASE64_STANDARD.encode(dictionary_bytes)),
            }),
        };
        let bytes = raw_compressed_entry_bytes(key, payload, 100)?;
        let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
        let path = write_raw_entry(&cache, key, &bytes)?;

        let lookup = cache.get_at(key, 120).map_err(|error| error.to_string())?;

        match lookup {
            PackL2CacheLookup::Miss(PackL2CacheMiss {
                reason:
                    PackL2CacheMissReason::CompressionDictionaryCorrupt {
                        dictionary_id,
                        message,
                    },
                ..
            }) => {
                assert_eq!(dictionary_id, "zstd_dict_corrupt");
                assert!(
                    message.contains("dictionary byte hash mismatch"),
                    "corrupt dictionary miss should explain the hash mismatch: {message}"
                );
            }
            other => {
                return Err(format!(
                    "corrupt dictionary should return a typed miss; got {other:?}"
                ));
            }
        }
        assert!(
            !path.exists(),
            "corrupt-dictionary compressed entries should be invalidated"
        );
        Ok(())
    }

    #[test]
    fn compressed_v2_oversized_uncompressed_length_uses_configured_entry_cap() -> TestResult {
        let key = "blake3:oversized-uncompressed";
        let max_entry_bytes = 1024_u64;
        let payload = PackL2CacheCompressionPayload {
            algorithm: PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1.to_owned(),
            compressed_payload_base64: BASE64_STANDARD.encode(b"not-a-zstd-frame"),
            compressed_byte_len: b"not-a-zstd-frame".len() as u64,
            uncompressed_byte_len: max_entry_bytes.saturating_add(1),
            uncompressed_hash: blake3_hash(b"not-present"),
            dictionary: None,
        };
        let bytes = raw_compressed_entry_bytes(key, payload, 100)?;
        assert!(
            bytes.len() as u64 <= max_entry_bytes,
            "test fixture envelope must fit under max_entry_bytes so the decompression cap is exercised"
        );
        let (_temp, cache) = cache_with_options(
            PackL2CacheOptions::new(4096, Duration::from_secs(60))
                .with_max_entry_bytes(max_entry_bytes),
        )?;
        let path = write_raw_entry(&cache, key, &bytes)?;

        let lookup = cache.get_at(key, 120).map_err(|error| error.to_string())?;

        match lookup {
            PackL2CacheLookup::Miss(PackL2CacheMiss {
                reason: PackL2CacheMissReason::CompressionDecode { message },
                ..
            }) => {
                assert!(
                    message.contains("decompression cap"),
                    "oversized compressed miss should cite decompression cap: {message}"
                );
            }
            other => {
                return Err(format!(
                    "oversized compressed entry should return a typed miss; got {other:?}"
                ));
            }
        }
        assert!(
            !path.exists(),
            "oversized compressed entries should be invalidated"
        );
        Ok(())
    }

    #[test]
    fn compressed_v2_corrupt_body_does_not_mask_valid_fallback() -> TestResult {
        let key = "blake3:compressed-multi-candidate";
        let pack = json!({"hash": "valid-fallback", "items": [{"id": "mem_valid"}]});
        let valid_bytes = raw_entry_bytes(key, pack.clone(), 100)?;
        let payload = PackL2CacheCompressionPayload {
            algorithm: PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1.to_owned(),
            compressed_payload_base64: BASE64_STANDARD.encode(b"not-a-zstd-frame"),
            compressed_byte_len: b"not-a-zstd-frame".len() as u64,
            uncompressed_byte_len: 64,
            uncompressed_hash: blake3_hash(b"not-a-json-payload"),
            dictionary: None,
        };
        let corrupt_bytes = raw_compressed_entry_bytes(key, payload, 100)?;
        let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
        ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;

        let corrupt_path = cache.entry_path_for_body_hash(key, &body_hash_prefix(&corrupt_bytes));
        fs::write(&corrupt_path, corrupt_bytes).map_err(|error| error.to_string())?;
        let valid_path = cache.entry_path(key);
        fs::write(&valid_path, valid_bytes).map_err(|error| error.to_string())?;

        let lookup = cache.get_at(key, 120).map_err(|error| error.to_string())?;

        match lookup {
            PackL2CacheLookup::Hit(hit) => {
                assert_eq!(hit.path, valid_path);
                assert_eq!(hit.pack_json, pack);
                assert!(
                    hit.compression.is_none(),
                    "valid fallback in this fixture is a legacy v1 entry"
                );
            }
            PackL2CacheLookup::Miss(miss) => {
                return Err(format!(
                    "valid fallback should hit after bad compressed candidate: {miss:?}"
                ));
            }
        }
        assert!(
            !corrupt_path.exists(),
            "bad compressed candidate should be invalidated before trying the valid fallback"
        );
        Ok(())
    }

    #[test]
    fn error_or_invalid_body_hash_mismatch_removes_entry() -> TestResult {
        let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
        ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;
        let path = cache.entry_path_for_body_hash("blake3:tampered", "0000000000000000");
        fs::write(&path, b"{\"schema\":\"tampered\"}").map_err(|error| error.to_string())?;

        let lookup = cache
            .get_at("blake3:tampered", 100)
            .map_err(|error| error.to_string())?;

        match lookup {
            PackL2CacheLookup::Miss(miss) => {
                assert_eq!(
                    miss.reason,
                    PackL2CacheMissReason::BodyHashMismatch {
                        expected: "0000000000000000".to_owned(),
                        actual: body_hash_prefix(b"{\"schema\":\"tampered\"}"),
                    }
                );
                assert_eq!(miss.path, path);
            }
            PackL2CacheLookup::Hit(_) => {
                return Err("body-hash mismatch must not hit".to_owned());
            }
        }
        assert!(
            !path.exists(),
            "body-hash mismatch should remove the corrupted entry"
        );
        Ok(())
    }

    #[test]
    fn error_or_invalid_key_mismatch_returns_miss() -> TestResult {
        let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
        let original = raw_entry_bytes("blake3:original", json!({"hash": "mismatch"}), 100)?;
        let path = write_raw_entry(&cache, "blake3:other", &original)?;

        let lookup = cache
            .get_at("blake3:other", 100)
            .map_err(|error| error.to_string())?;

        match lookup {
            PackL2CacheLookup::Miss(miss) => assert_eq!(
                miss.reason,
                PackL2CacheMissReason::KeyMismatch {
                    stored_key: "blake3:original".to_owned()
                }
            ),
            PackL2CacheLookup::Hit(_) => return Err("mismatched key must not hit".to_owned()),
        }
        assert!(
            !path.exists(),
            "key-mismatched cache entry should be invalidated"
        );
        Ok(())
    }

    #[test]
    fn error_or_invalid_unwritable_root_reports_io_error() -> TestResult {
        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let file_root = temp.path().join("not-a-directory");
        fs::write(&file_root, b"already a file").map_err(|error| error.to_string())?;
        let cache = PackL2Cache::new(file_root.clone(), PackL2CacheOptions::default());

        let error = cache
            .put_at("blake3:key", &json!({"hash": "nope"}), 100)
            .expect_err("file root should not be writable as a cache directory");

        match error {
            PackL2CacheError::Io {
                path,
                operation: "create_dir_all",
                ..
            } => assert_eq!(path, file_root),
            other => return Err(format!("unexpected error: {other}")),
        }
        Ok(())
    }

    #[test]
    fn error_or_invalid_symlink_scan_stops_cleanly_at_file_component() -> TestResult {
        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let file_component = temp.path().join("not-a-directory");
        fs::write(&file_component, b"already a file").map_err(|error| error.to_string())?;
        let path_below_file = file_component.join("child.json");

        let symlink = first_existing_symlink_component(&path_below_file)
            .map_err(|error| error.to_string())?;

        assert_eq!(
            symlink, None,
            "a non-directory component should stop the existing-prefix scan, not become an IO failure"
        );
        Ok(())
    }

    #[cfg(unix)]
    #[test]
    fn error_or_invalid_put_rejects_symlinked_cache_root() -> TestResult {
        use std::os::unix::fs::symlink;

        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let real_root = temp.path().join("real-pack-l2");
        fs::create_dir_all(&real_root).map_err(|error| error.to_string())?;
        let linked_root = temp.path().join("pack-l2");
        symlink(&real_root, &linked_root).map_err(|error| error.to_string())?;
        let cache = PackL2Cache::new(linked_root.clone(), PackL2CacheOptions::default());

        let error = cache
            .put_at("blake3:symlink-root", &json!({"hash": "unsafe"}), 100)
            .expect_err("symlinked cache root should be rejected");

        match error {
            PackL2CacheError::Io {
                path,
                operation: "inspect_root",
                source,
            } => {
                assert_eq!(path, linked_root);
                assert_eq!(source.kind(), io::ErrorKind::PermissionDenied);
            }
            other => return Err(format!("unexpected error: {other}")),
        }
        assert!(
            fs::read_dir(&real_root)
                .map_err(|error| error.to_string())?
                .next()
                .is_none(),
            "cache write must not publish through symlinked root"
        );
        Ok(())
    }

    #[cfg(unix)]
    #[test]
    fn error_or_invalid_get_rejects_symlinked_cache_root() -> TestResult {
        use std::os::unix::fs::symlink;

        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let real_root = temp.path().join("real-pack-l2");
        fs::create_dir_all(&real_root).map_err(|error| error.to_string())?;
        let linked_root = temp.path().join("pack-l2");
        symlink(&real_root, &linked_root).map_err(|error| error.to_string())?;
        let cache = PackL2Cache::new(linked_root.clone(), PackL2CacheOptions::default());

        let error = cache
            .get_at("blake3:symlink-root", 100)
            .expect_err("symlinked cache root should be rejected before lookup");

        match error {
            PackL2CacheError::Io {
                path,
                operation: "inspect_root",
                source,
            } => {
                assert_eq!(path, linked_root);
                assert_eq!(source.kind(), io::ErrorKind::PermissionDenied);
            }
            other => return Err(format!("unexpected error: {other}")),
        }
        Ok(())
    }

    #[cfg(unix)]
    #[test]
    fn error_or_invalid_get_and_put_reject_symlinked_cache_entry() -> TestResult {
        use std::os::unix::fs::symlink;

        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let cache = PackL2Cache::new(
            temp.path().join("pack-l2"),
            PackL2CacheOptions::new(4096, Duration::from_secs(60)),
        );
        ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;
        let outside_entry = temp.path().join("outside-entry.json");
        fs::write(&outside_entry, br#"{"schema":"outside"}"#).map_err(|error| error.to_string())?;
        let linked_entry =
            cache.entry_path_for_body_hash("blake3:linked-entry", "0000000000000000");
        symlink(&outside_entry, &linked_entry).map_err(|error| error.to_string())?;

        let get_error = cache
            .get_at("blake3:linked-entry", 100)
            .expect_err("symlinked final cache entry should not be read");
        match get_error {
            PackL2CacheError::Io {
                path,
                operation: "inspect_entry",
                source,
            } => {
                assert_eq!(path, linked_entry);
                assert_eq!(source.kind(), io::ErrorKind::PermissionDenied);
            }
            other => return Err(format!("unexpected get error: {other}")),
        }

        let overwrite_pack = json!({"hash": "overwrite"});
        let overwrite_bytes = raw_entry_bytes("blake3:linked-entry", overwrite_pack.clone(), 100)?;
        let linked_write_entry = cache
            .entry_path_for_body_hash("blake3:linked-entry", &body_hash_prefix(&overwrite_bytes));
        symlink(&outside_entry, &linked_write_entry).map_err(|error| error.to_string())?;
        let put_error = cache
            .put_at("blake3:linked-entry", &overwrite_pack, 100)
            .expect_err("symlinked final cache entry should not be overwritten");
        match put_error {
            PackL2CacheError::Io {
                path,
                operation: "inspect_entry",
                source,
            } => {
                assert_eq!(path, linked_write_entry);
                assert_eq!(source.kind(), io::ErrorKind::PermissionDenied);
            }
            other => return Err(format!("unexpected put error: {other}")),
        }
        assert_eq!(
            fs::read_to_string(&outside_entry).map_err(|error| error.to_string())?,
            r#"{"schema":"outside"}"#,
            "cache write must not overwrite a symlink target"
        );
        Ok(())
    }

    #[cfg(unix)]
    #[test]
    fn cache_entry_final_read_open_rejects_symlinked_entry_path() -> TestResult {
        use std::os::unix::fs::symlink;

        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let outside_entry = temp.path().join("outside-entry.json");
        fs::write(&outside_entry, br#"{"schema":"outside"}"#).map_err(|error| error.to_string())?;
        let linked_entry = temp.path().join("linked-entry.json");
        symlink(&outside_entry, &linked_entry).map_err(|error| error.to_string())?;

        let error = open_cache_entry_file_for_read(&linked_entry)
            .expect_err("final cache entry read open must reject symlinks");

        assert_ne!(
            error.kind(),
            io::ErrorKind::NotFound,
            "final symlink read should fail because the path is a symlink"
        );
        assert_eq!(
            fs::read_to_string(&outside_entry).map_err(|error| error.to_string())?,
            r#"{"schema":"outside"}"#,
            "cache read helper must not follow the symlink target"
        );
        Ok(())
    }

    #[cfg(unix)]
    #[test]
    fn cache_temp_final_create_open_rejects_symlinked_temp_path() -> TestResult {
        use std::os::unix::fs::symlink;

        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let outside_entry = temp.path().join("outside-temp.json");
        fs::write(&outside_entry, br#"{"schema":"outside"}"#).map_err(|error| error.to_string())?;
        let linked_temp = temp.path().join("entry.tmp");
        symlink(&outside_entry, &linked_temp).map_err(|error| error.to_string())?;

        let error = open_cache_temp_file_for_create(&linked_temp)
            .expect_err("final cache temp create open must reject symlinks");

        assert_ne!(
            error.kind(),
            io::ErrorKind::NotFound,
            "final symlink create should fail because the path is a symlink"
        );
        assert_eq!(
            fs::read_to_string(&outside_entry).map_err(|error| error.to_string())?,
            r#"{"schema":"outside"}"#,
            "cache temp create helper must not follow or truncate the symlink target"
        );
        Ok(())
    }

    #[cfg(unix)]
    #[test]
    fn error_or_invalid_publish_rechecks_symlinked_final_entry() -> TestResult {
        use std::os::unix::fs::symlink;

        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let cache = PackL2Cache::new(
            temp.path().join("pack-l2"),
            PackL2CacheOptions::new(4096, Duration::from_secs(60)),
        );
        ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;

        let pack_json = json!({"hash": "publish-recheck"});
        let bytes = raw_entry_bytes("blake3:publish-recheck", pack_json, 100)?;
        let body_hash = body_hash_prefix(&bytes);
        let entry_path = cache.entry_path_for_body_hash("blake3:publish-recheck", &body_hash);
        let temp_path = cache.temp_path("blake3:publish-recheck", &body_hash, 100);
        write_synced_file(&temp_path, &bytes).map_err(|error| error.to_string())?;

        let outside_entry = temp.path().join("outside-entry.json");
        fs::write(&outside_entry, br#"{"schema":"outside"}"#).map_err(|error| error.to_string())?;
        symlink(&outside_entry, &entry_path).map_err(|error| error.to_string())?;

        let error = publish_cache_entry_temp_file(&temp_path, &entry_path)
            .expect_err("symlinked final entry should be rejected before publish");
        match error {
            PackL2CacheError::Io {
                path,
                operation: "inspect_entry",
                source,
            } => {
                assert_eq!(path, entry_path);
                assert_eq!(source.kind(), io::ErrorKind::PermissionDenied);
            }
            other => return Err(format!("unexpected publish error: {other}")),
        }
        assert_eq!(
            fs::read_to_string(&outside_entry).map_err(|error| error.to_string())?,
            r#"{"schema":"outside"}"#,
            "cache publish must not mutate the symlink target"
        );
        assert!(
            temp_path.exists(),
            "cache temp entry should remain available after rejected publish"
        );
        Ok(())
    }

    #[cfg(unix)]
    #[test]
    fn eviction_skips_symlinked_json_entries_without_following_targets() -> TestResult {
        use std::os::unix::fs::symlink;

        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let cache = PackL2Cache::new(
            temp.path().join("pack-l2"),
            PackL2CacheOptions::new(0, Duration::from_secs(0)),
        );
        ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;
        let outside_entry = temp.path().join("outside-entry.json");
        fs::write(&outside_entry, br#"{"storedAtEpochSeconds":0}"#)
            .map_err(|error| error.to_string())?;
        let linked_entry = cache.root().join("linked.json");
        symlink(&outside_entry, &linked_entry).map_err(|error| error.to_string())?;

        let report = cache
            .evict_best_effort_at(100)
            .map_err(|error| error.to_string())?;

        assert_eq!(report.skipped, 1, "symlink entries should be skipped");
        assert_eq!(report.removed, 0, "symlink entries should not be removed");
        assert!(
            fs::symlink_metadata(&linked_entry)
                .map_err(|error| error.to_string())?
                .file_type()
                .is_symlink(),
            "cache eviction should leave the symlink entry untouched"
        );
        assert!(
            outside_entry.exists(),
            "cache eviction must not follow and remove a symlink target"
        );
        Ok(())
    }

    #[test]
    fn eviction_removes_expired_entries_before_fresh_entries() -> TestResult {
        let (_temp, cache) = cache(10_000, Duration::from_secs(10))?;
        cache
            .put_at("blake3:old", &json!({"payload": "old"}), 100)
            .map_err(|error| error.to_string())?;
        let fresh_report = cache
            .put_at("blake3:fresh", &json!({"payload": "fresh"}), 120)
            .map_err(|error| error.to_string())?;

        assert_eq!(
            fresh_report.eviction.removed, 1,
            "one expired entry should be removed during the next write"
        );
        assert!(
            matches!(
                cache
                    .get_at("blake3:old", 120)
                    .map_err(|error| error.to_string())?,
                PackL2CacheLookup::Miss(PackL2CacheMiss {
                    reason: PackL2CacheMissReason::NotFound,
                    ..
                })
            ),
            "old entry should be gone"
        );
        assert!(
            cache
                .get_at("blake3:fresh", 120)
                .map_err(|error| error.to_string())?
                .is_hit(),
            "fresh entry should remain"
        );
        Ok(())
    }

    #[test]
    fn eviction_reduces_cache_to_byte_cap_by_oldest_first() -> TestResult {
        let (_temp, cache) = cache(170, Duration::from_secs(10_000))?;
        cache
            .put_at(
                "blake3:first",
                &json!({"payload": "aaaaaaaaaaaaaaaaaaaaaaaa"}),
                100,
            )
            .map_err(|error| error.to_string())?;
        cache
            .put_at(
                "blake3:second",
                &json!({"payload": "bbbbbbbbbbbbbbbbbbbbbbbb"}),
                200,
            )
            .map_err(|error| error.to_string())?;
        let third_report = cache
            .put_at(
                "blake3:third",
                &json!({"payload": "cccccccccccccccccccccccc"}),
                300,
            )
            .map_err(|error| error.to_string())?;

        let report = cache
            .evict_best_effort_at(300)
            .map_err(|error| error.to_string())?;
        let removed_total = third_report.eviction.removed.saturating_add(report.removed);

        assert!(
            report.bytes_after <= cache.options().max_bytes,
            "eviction should reduce byte usage below the configured cap"
        );
        assert!(
            removed_total >= 1,
            "at least one entry should be evicted by write-through or explicit eviction"
        );
        assert!(
            matches!(
                cache
                    .get_at("blake3:first", 300)
                    .map_err(|error| error.to_string())?,
                PackL2CacheLookup::Miss(PackL2CacheMiss {
                    reason: PackL2CacheMissReason::NotFound,
                    ..
                })
            ),
            "oldest entry should be evicted first"
        );
        Ok(())
    }

    #[test]
    fn eviction_uses_touched_mtime_for_lru_order() -> TestResult {
        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let root = temp.path().join("pack-l2");
        let writer = PackL2Cache::new(
            root.clone(),
            PackL2CacheOptions::new(u64::MAX, Duration::from_secs(10_000)),
        );
        let first = writer
            .put_at("blake3:first", &json!({"payload": "first"}), 100)
            .map_err(|error| error.to_string())?;
        let _second = writer
            .put_at("blake3:second", &json!({"payload": "second"}), 200)
            .map_err(|error| error.to_string())?;
        assert!(
            writer
                .get_at("blake3:first", 300)
                .map_err(|error| error.to_string())?
                .is_hit(),
            "read should touch the first entry before size eviction"
        );
        let third = writer
            .put_at("blake3:third", &json!({"payload": "third"}), 250)
            .map_err(|error| error.to_string())?;

        let evicting = PackL2Cache::new(
            root,
            PackL2CacheOptions::new(first.byte_len + third.byte_len, Duration::from_secs(10_000)),
        );
        let report = evicting
            .evict_best_effort_at(300)
            .map_err(|error| error.to_string())?;

        assert_eq!(
            report.removed, 1,
            "size eviction should remove exactly one oldest LRU entry"
        );
        assert!(
            matches!(
                evicting
                    .get_at("blake3:second", 300)
                    .map_err(|error| error.to_string())?,
                PackL2CacheLookup::Miss(PackL2CacheMiss {
                    reason: PackL2CacheMissReason::NotFound,
                    ..
                })
            ),
            "untouched second entry should be evicted before the touched first entry"
        );
        assert!(
            evicting
                .get_at("blake3:first", 300)
                .map_err(|error| error.to_string())?
                .is_hit(),
            "touched first entry should survive LRU eviction"
        );
        assert!(
            evicting
                .get_at("blake3:third", 300)
                .map_err(|error| error.to_string())?
                .is_hit(),
            "newest third entry should survive LRU eviction"
        );
        Ok(())
    }

    #[test]
    fn concurrent_eviction_enoent_treated_as_success() -> TestResult {
        let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
        let candidate = EvictionCandidate {
            path: temp.path().join("already-evicted.json"),
            byte_len: 128,
            stored_epoch_seconds: 100,
            last_used_epoch_seconds: 100,
            expired: true,
        };
        let mut report = PackL2EvictionReport {
            bytes_before: 256,
            ..PackL2EvictionReport::default()
        };
        let mut bytes_current = report.bytes_before;

        remove_eviction_candidate_file(&candidate, &mut report, &mut bytes_current);

        assert_eq!(
            report.skipped, 0,
            "peer-removed cache entries should not count as skipped"
        );
        assert_eq!(
            report.removed, 1,
            "peer-removed cache entries count as logically removed"
        );
        assert_eq!(
            report.bytes_removed, candidate.byte_len,
            "logical byte accounting should include the raced entry"
        );
        assert_eq!(
            bytes_current, 128,
            "current byte estimate should shrink after ENOENT"
        );
        Ok(())
    }

    #[test]
    fn happy_path_cache_directory_uses_private_permissions() -> TestResult {
        let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
        cache
            .put_at("blake3:key-a", &json!({"hash": "perms"}), 100)
            .map_err(|error| error.to_string())?;

        #[cfg(unix)]
        {
            let mode = fs::metadata(cache.root())
                .map_err(|error| error.to_string())?
                .permissions()
                .mode()
                & 0o777;
            assert_eq!(mode, 0o700, "cache directory should be owner-only");
        }
        Ok(())
    }
}