zshrs 0.13.41

//! rkyv-backed bytecode cache for zsh scripts.
//!
//! Single-file shard at `~/.zshrs/scripts.rkyv`. On 2+ runs of a given
//! script, lex/parse/compile is skipped — the cache hit is `mmap` + zero-copy
//! `ArchivedHashMap` lookup + bincode-decode of the inner `fusevm::Chunk` blob.
//!
//! Storage layout (rkyv archived):
//!   ScriptShard {
//!     header: { magic, format_version, zshrs_version, pointer_width, built_at_secs },
//!     entries: HashMap<canonical_path, ScriptEntry>,
//!   }
//!   ScriptEntry { mtime_secs, mtime_nsecs, binary_mtime_at_cache,
//!                 binary_len_at_cache, cached_at_secs, chunk_blob: `Vec<u8>` }
//!
//! Inner `chunk_blob` is bincode for now — `fusevm::Chunk` is owned by the
//! upstream `fusevm` crate and only derives `serde::Serialize`/`Deserialize`,
//! not `rkyv::Archive`, so the inner codec stays bincode inside the rkyv outer
//! container. Direct rkyv on Chunk would require either forking fusevm or a
//! mirror archived type — both are large refactors and not needed for the
//! current "kill SQLite for bytecode" goal.
//!
//! Read path:
//!   - Lazy `mmap` of the shard, kept alive for the process lifetime so repeat
//!     lookups pay validation once.
//!   - `rkyv::check_archived_root::<ScriptShard>` validates the byte image.
//!   - Header validated for magic / format_version / zshrs_version / pointer_width.
//!   - Per-entry: source mtime must match, and the entry's recorded binary
//!     identity (`binary_mtime_at_cache`, `binary_len_at_cache`) must EQUAL the
//!     running binary's (mtime, len). Bytecode is only ever replayed by the
//!     exact build that emitted it; any rebuild invalidates entries silently.
//!
//! Write path:
//!   - `bin_zsystem_flock(LOCK_EX)` on `scripts.rkyv.lock` so concurrent writers serialize.
//!   - Read existing shard into owned form, mutate, `rkyv::to_bytes`,
//!     write to `scripts.rkyv.tmp.<pid>.<nanos>`, fsync, atomic-rename.
//!   - Drop the in-process `mmap` so the next read picks up the new shard.
//!
//! Run log: a cache hit never rewrites the shard. Its time is appended to
//! `scripts.runs` (`unix_secs\tpath`, flushed with the pending puts) and
//! overlaid on `dbview scripts`; the next shard write folds it into
//! `cached_at_secs` and removes the log.
//!
//! Ported from `strykelang/strykelang/script_cache.rs` (the user's stryke
//! language has the same caching pattern; this is the same shape with `ZRSC`
//! magic, zshrs version pin, and a single `chunk_blob` per entry — zshrs has
//! no separate AST cache).

use std::collections::HashMap;
use std::fs::File;
use std::io::Write as IoWrite;
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use std::time::{SystemTime, UNIX_EPOCH};

use memmap2::Mmap;
use parking_lot::Mutex;
use rkyv::{Archive, Deserialize as RkyvDeserialize, Serialize as RkyvSerialize};
use std::os::unix::fs::MetadataExt;

/// Magic header bytes — fail-fast if a wrong-format file is mmap'd.
/// "ZRSC" little-endian.
pub const SHARD_MAGIC: u32 = 0x5A525343;
/// Bumped on incompatible rkyv schema changes.
///
/// v2 added `ScriptEntry::binary_len_at_cache`; a v1 shard has no length to
/// compare against, so it is rejected wholesale rather than half-validated.
/// v3: token chars moved to the Private Use Area (crate::token_char).
/// v5: `ScriptEntry::env_fingerprint`, and a one-byte kind tag in front of
/// every `chunk_blob` (a script's single chunk vs a sourced file's events).
pub const SHARD_FORMAT_VERSION: u32 = 5;
/// `ShardHeader` — see fields for layout.
#[derive(Archive, RkyvDeserialize, RkyvSerialize, Debug, Clone)]
#[archive(check_bytes)]
pub struct ShardHeader {
    /// `magic` field.
    pub magic: u32,
    /// `format_version` field.
    pub format_version: u32,
    /// `zshrs_version` field.
    pub zshrs_version: String,
    /// `pointer_width` field.
    pub pointer_width: u32,
    /// `built_at_secs` field.
    pub built_at_secs: u64,
}
/// `ScriptEntry` — see fields for layout.
#[derive(Archive, RkyvDeserialize, RkyvSerialize, Debug, Clone)]
#[archive(check_bytes)]
pub struct ScriptEntry {
    /// `mtime_secs` field.
    pub mtime_secs: i64,
    /// `mtime_nsecs` field.
    pub mtime_nsecs: i64,
    /// mtime of the zshrs binary that compiled `chunk_blob`.
    pub binary_mtime_at_cache: i64,
    /// Size of the zshrs binary that compiled `chunk_blob`. Paired with
    /// `binary_mtime_at_cache` to identify the emitting build: mtime alone
    /// has one-second granularity and moves in both directions (an older
    /// binary restored over a newer one keeps its old timestamp), so it
    /// cannot by itself prove the running build emitted these bytes.
    pub binary_len_at_cache: u64,
    /// `cached_at_secs` field.
    pub cached_at_secs: i64,
    /// Hash of the lexer-visible shell state (aliases, options, emulation) the
    /// program was lexed and compiled under. The compiled chunks have alias
    /// expansions and option-dependent code baked in, so an entry is served
    /// only to a run that starts from the same state.
    pub env_fingerprint: u64,
    /// `chunk_blob` field.
    pub chunk_blob: Vec<u8>,
}
/// `ScriptShard` — see fields for layout.
#[derive(Archive, RkyvDeserialize, RkyvSerialize, Debug, Clone)]
#[archive(check_bytes)]
pub struct ScriptShard {
    /// `header` field.
    pub header: ShardHeader,
    /// `entries` field.
    pub entries: HashMap<String, ScriptEntry>,
}

/// mmap + validated `*const ArchivedScriptShard`. Self-referential — the pointer
/// is valid for the lifetime of the wrapping struct.
pub struct MmappedShard {
    /// `_mmap` field.
    _mmap: Mmap,
    /// `archived` field.
    archived: *const ArchivedScriptShard,
    /// `(inode, len, mtime_ns)` of the file this mapping was taken from. The
    /// shard is replaced by atomic rename, so another process's write leaves
    /// this mapping on the OLD inode; comparing against the path's current
    /// identity is how a long-lived shell notices.
    identity: (u64, u64, i64),
}

// SAFETY: the pointer aliases an immutable mmap that lives as long as Self.
// rkyv-validated reads are immutable.
unsafe impl Send for MmappedShard {}
unsafe impl Sync for MmappedShard {}

impl MmappedShard {
    /// `open` — see implementation.
    pub fn open(path: &Path) -> Option<Self> {
        let file = File::open(path).ok()?;
        let identity = file_identity(&file.metadata().ok()?);
        let mmap = unsafe { Mmap::map(&file).ok()? };
        let archived = rkyv::check_archived_root::<ScriptShard>(&mmap[..]).ok()?;
        let archived_ptr = archived as *const ArchivedScriptShard;
        Some(Self {
            _mmap: mmap,
            archived: archived_ptr,
            identity,
        })
    }

    fn shard(&self) -> &ArchivedScriptShard {
        // SAFETY: see Self impl comment.
        unsafe { &*self.archived }
    }

    fn header_ok(&self) -> bool {
        let h = &self.shard().header;
        let magic: u32 = h.magic.into();
        let fv: u32 = h.format_version.into();
        let pw: u32 = h.pointer_width.into();
        magic == SHARD_MAGIC
            && fv == SHARD_FORMAT_VERSION
            && pw as usize == std::mem::size_of::<usize>()
            && h.zshrs_version.as_str() == env!("CARGO_PKG_VERSION")
    }

    fn lookup(&self, path: &str) -> Option<&ArchivedScriptEntry> {
        self.shard().entries.get(path)
    }

    fn entry_count(&self) -> usize {
        self.shard().entries.len()
    }
}

/// One buffered write: what `try_save_bytes` queues until the next flush.
#[derive(Clone)]
pub struct PendingPut {
    path: String,
    mtime_secs: i64,
    mtime_nsecs: i64,
    env_fingerprint: u64,
    /// Kind-tagged blob, as stored.
    blob: Vec<u8>,
}

/// Writes buffered since the last flush; see [`try_flush_pending`].
static PENDING: Mutex<Vec<PendingPut>> = Mutex::new(Vec::new());

/// Process that queued the newest entry. A forked child inherits its parent's
/// buffer; only the process that queued an entry may write it, or every
/// external command a shell spawns would rewrite the shard.
static PENDING_PID: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);

/// Cache hits since the last flush: `(path, unix_secs)`. A hit leaves the shard
/// untouched (rewriting it per run would put a multi-megabyte write on every
/// startup), so the time of the run is appended to `scripts.runs` instead and
/// overlaid on the listing; see [`ScriptCache::overlay_runs`].
static RAN: Mutex<Vec<(String, i64)>> = Mutex::new(Vec::new());

/// Process that queued the newest entry of [`RAN`]; same fork rule as
/// [`PENDING_PID`].
static RAN_PID: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);

/// `scripts.runs` is rewritten deduplicated once it grows past this.
const RUNS_COMPACT_BYTES: u64 = 64 * 1024;

/// Past this many buffered entries `try_save_bytes` flushes by itself, so a
/// long non-interactive run cannot hold an unbounded amount of bytecode.
const PENDING_FLUSH_MAX: usize = 128;

/// Shard cache keyed by canonical script path. One per shard file.
pub struct ScriptCache {
    /// `path` field.
    path: PathBuf,
    /// `lock_path` field.
    lock_path: PathBuf,
    /// `mmap` field.
    mmap: Mutex<Option<MmappedShard>>,
}

impl ScriptCache {
    /// `open` — see implementation.
    pub fn open(path: &Path) -> std::io::Result<Self> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let parent = path.parent().unwrap_or_else(|| Path::new("/tmp"));
        let lock_path = parent.join(format!(
            "{}.lock",
            path.file_name()
                .and_then(|s| s.to_str())
                .unwrap_or("scripts.rkyv")
        ));
        Ok(Self {
            path: path.to_path_buf(),
            lock_path,
            mmap: Mutex::new(None),
        })
    }

    fn ensure_mmap(&self) {
        let mut guard = self.mmap.lock();
        // One `stat`: remap when another process replaced the shard (or it
        // vanished) since this mapping was taken. `put` and `clear` in THIS
        // process invalidate directly; a write from a script run elsewhere
        // only shows up here.
        let on_disk = std::fs::metadata(&self.path).ok().map(|m| file_identity(&m));
        let stale = match (guard.as_ref(), on_disk) {
            (Some(m), Some(id)) => m.identity != id,
            (Some(_), None) => true,
            (None, _) => false,
        };
        if stale {
            *guard = None;
        }
        if guard.is_none() {
            *guard = MmappedShard::open(&self.path);
        }
    }

    fn invalidate_mmap(&self) {
        let mut guard = self.mmap.lock();
        *guard = None;
    }

    /// Cache lookup. Returns `None` on miss, mtime mismatch, version drift, or
    /// zshrs binary newer than the cached entry.
    pub fn get(
        &self,
        path: &str,
        mtime_secs: i64,
        mtime_nsecs: i64,
        env_fingerprint: u64,
    ) -> Option<Vec<u8>> {
        self.ensure_mmap();
        let guard = self.mmap.lock();
        let shard = guard.as_ref()?;
        if !shard.header_ok() {
            return None;
        }
        let entry = shard.lookup(path)?;

        let entry_mtime_s: i64 = entry.mtime_secs.into();
        let entry_mtime_ns: i64 = entry.mtime_nsecs.into();
        if entry_mtime_s != mtime_secs || entry_mtime_ns != mtime_nsecs {
            return None;
        }
        let entry_fp: u64 = entry.env_fingerprint.into();
        if entry_fp != env_fingerprint {
            return None;
        }

        // Was this chunk emitted by the binary that is running right now?
        //
        // EXACT equality on (mtime, len), the same test `autoload_cache`
        // already applies to its chunks. The previous `cached < running`
        // comparison only rejected a cache written by an OLDER build, so a
        // binary whose mtime went BACKWARDS — an earlier build restored over
        // a later one, a `cp` that preserves timestamps, a checkout of a
        // previously-built target dir — silently replayed the newer build's
        // bytecode. Proven by touching the binary's mtime into the past and
        // watching the shard still hit. The one-second granularity of the
        // stored mtime has the same effect when a rebuild lands in the same
        // second as the cache write, which is why the length is compared too.
        match current_binary_identity() {
            Some((bin_mtime, bin_len)) => {
                let cached_bin_mtime: i64 = entry.binary_mtime_at_cache.into();
                let cached_bin_len: u64 = entry.binary_len_at_cache.into();
                if cached_bin_mtime != bin_mtime || cached_bin_len != bin_len {
                    return None;
                }
            }
            // No `current_exe()` — nothing can be proven, so nothing is used.
            None => return None,
        }

        Some(entry.chunk_blob.as_slice().to_vec())
    }

    /// Insert / replace an entry. Serializes the whole shard and atomic-renames.
    pub fn put(
        &self,
        path: &str,
        mtime_secs: i64,
        mtime_nsecs: i64,
        env_fingerprint: u64,
        chunk_blob: Vec<u8>,
    ) -> Result<(), String> {
        let _lock = match acquire_lock(&self.lock_path) {
            Some(l) => l,
            None => return Ok(()),
        };

        let mut shard = match read_owned_shard(&self.path) {
            Some(s)
                if s.header.zshrs_version == env!("CARGO_PKG_VERSION")
                    && s.header.pointer_width as usize == std::mem::size_of::<usize>()
                    && s.header.format_version == SHARD_FORMAT_VERSION =>
            {
                s
            }
            _ => fresh_shard(),
        };

        self.fold_runs(&mut shard);
        let (bin_mtime, bin_len) = current_binary_identity().unwrap_or((0, 0));
        let entry = ScriptEntry {
            mtime_secs,
            mtime_nsecs,
            binary_mtime_at_cache: bin_mtime,
            binary_len_at_cache: bin_len,
            cached_at_secs: now_secs(),
            env_fingerprint,
            chunk_blob,
        };
        shard.entries.insert(path.to_string(), entry);
        shard.header.built_at_secs = now_secs() as u64;

        write_shard_atomic(&self.path, &shard)?;
        let _ = std::fs::remove_file(self.runs_path());
        self.invalidate_mmap();
        Ok(())
    }

    /// Insert / replace several entries with ONE lock, read and rewrite of the
    /// shard. `put` costs a full shard rewrite and fsync per call, so a shell
    /// that sources a hundred files for the first time would pay that a hundred
    /// times; the sourced-file path buffers and flushes here instead.
    pub fn put_many(&self, batch: Vec<PendingPut>) -> Result<(), String> {
        if batch.is_empty() {
            return Ok(());
        }
        let _lock = match acquire_lock(&self.lock_path) {
            Some(l) => l,
            None => return Ok(()),
        };
        let mut shard = match read_owned_shard(&self.path) {
            Some(s)
                if s.header.zshrs_version == env!("CARGO_PKG_VERSION")
                    && s.header.pointer_width as usize == std::mem::size_of::<usize>()
                    && s.header.format_version == SHARD_FORMAT_VERSION =>
            {
                s
            }
            _ => fresh_shard(),
        };
        self.fold_runs(&mut shard);
        let (bin_mtime, bin_len) = current_binary_identity().unwrap_or((0, 0));
        for p in batch {
            shard.entries.insert(
                p.path,
                ScriptEntry {
                    mtime_secs: p.mtime_secs,
                    mtime_nsecs: p.mtime_nsecs,
                    binary_mtime_at_cache: bin_mtime,
                    binary_len_at_cache: bin_len,
                    cached_at_secs: now_secs(),
                    env_fingerprint: p.env_fingerprint,
                    chunk_blob: p.blob,
                },
            );
        }
        shard.header.built_at_secs = now_secs() as u64;
        write_shard_atomic(&self.path, &shard)?;
        let _ = std::fs::remove_file(self.runs_path());
        self.invalidate_mmap();
        Ok(())
    }

    /// `(count    /// `(count, total_blob_bytes)` snapshot.
    pub fn stats(&self) -> (i64, i64) {
        self.ensure_mmap();
        let guard = self.mmap.lock();
        let (mut count, mut bytes) = guard.as_ref().map_or((0, 0), |shard| {
            let bytes: i64 = shard
                .shard()
                .entries
                .values()
                .map(|e| e.chunk_blob.len() as i64)
                .sum();
            (shard.entry_count() as i64, bytes)
        });
        // Buffered, not yet in the shard (see `list_scripts`).
        let mut seen = std::collections::HashSet::new();
        for p in PENDING.lock().iter() {
            let in_shard = guard
                .as_ref()
                .is_some_and(|s| s.shard().entries.get(p.path.as_str()).is_some());
            if !in_shard && seen.insert(p.path.clone()) {
                count += 1;
                bytes += p.blob.len() as i64;
            }
        }
        (count, bytes)
    }

    /// `(path, chunk_kb, version, cached_at_localstr)` per entry,
    /// sorted by `cached_at` desc.
    pub fn list_scripts(&self) -> Vec<(String, f64, String, String)> {
        self.ensure_mmap();
        let guard = self.mmap.lock();
        let v = guard
            .as_ref()
            .map(|s| s.shard().header.zshrs_version.as_str().to_string())
            .unwrap_or_else(|| env!("CARGO_PKG_VERSION").to_string());
        let mut out: Vec<(String, f64, String, String, i64)> = guard
            .as_ref()
            .map(|shard| {
                shard
                    .shard()
                    .entries
                    .iter()
                    .map(|(k, e)| {
                        let chunk_kb = e.chunk_blob.len() as f64 / 1024.0;
                        let cached_at: i64 = e.cached_at_secs.into();
                        let ts = format_local_ts(cached_at);
                        (k.as_str().to_string(), chunk_kb, v.clone(), ts, cached_at)
                    })
                    .collect()
            })
            .unwrap_or_default();
        self.overlay_runs(&mut out);
        // Entries buffered in this process (or inherited by a forked `$(...)`)
        // are not in the shard yet; they are the newest word on their path.
        let now = now_secs();
        for p in PENDING.lock().iter() {
            out.retain(|row| row.0 != p.path);
            out.push((
                p.path.clone(),
                p.blob.len() as f64 / 1024.0,
                v.clone(),
                format_local_ts(now),
                now,
            ));
        }
        out.sort_by_key(|x| std::cmp::Reverse(x.4));
        out.into_iter()
            .map(|(p, ck, ver, ts, _)| (p, ck, ver, ts))
            .collect()
    }

    /// `scripts.runs`, the append-only log of cache hits (`unix_secs\tpath`).
    fn runs_path(&self) -> PathBuf {
        self.path.with_file_name("scripts.runs")
    }

    /// Raise each row's time to the newest logged run of its path, so a script
    /// replayed from the cache lists as run now rather than as stored then.
    fn overlay_runs(&self, rows: &mut [(String, f64, String, String, i64)]) {
        let mut ran = read_runs(&self.runs_path());
        for (path, ts) in RAN.lock().iter() {
            let t = ran.entry(path.clone()).or_insert(*ts);
            *t = (*t).max(*ts);
        }
        for row in rows.iter_mut() {
            if let Some(&ts) = ran.get(&row.0).filter(|&&ts| ts > row.4) {
                row.3 = format_local_ts(ts);
                row.4 = ts;
            }
        }
    }

    /// Carry logged runs into the entries about to be rewritten, so the log can
    /// be dropped with the old shard. Called with the shard lock held.
    fn fold_runs(&self, shard: &mut ScriptShard) {
        for (path, ts) in read_runs(&self.runs_path()) {
            if let Some(e) = shard.entries.get_mut(&path) {
                e.cached_at_secs = e.cached_at_secs.max(ts);
            }
        }
    }

    /// `scripts.db`, the read-only SQLite mirror kept beside the shard. Never
    /// read on a cache lookup or while running code — `dbview` only.
    fn mirror_path(&self) -> PathBuf {
        self.path.with_file_name("scripts.db")
    }

    /// Bring `scripts.db` in line with the shard. The mirror is derived data:
    /// it is rebuilt whole whenever the shard's identity (inode, length, mtime)
    /// differs from the one recorded in its `meta` table, so writers of the
    /// shard never touch SQLite and a missing or stale file heals itself.
    fn refresh_mirror(&self, conn: &rusqlite::Connection) -> rusqlite::Result<()> {
        use rusqlite::OptionalExtension;
        conn.execute_batch(
            "CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
             CREATE TABLE IF NOT EXISTS scripts (
                 path               TEXT PRIMARY KEY,
                 kind               TEXT NOT NULL,
                 events             INTEGER NOT NULL,
                 bytes              INTEGER NOT NULL,
                 version            TEXT NOT NULL,
                 cached_at          INTEGER NOT NULL,
                 source_mtime_secs  INTEGER NOT NULL,
                 source_mtime_nsecs INTEGER NOT NULL,
                 env_fingerprint    TEXT NOT NULL,
                 binary_len         INTEGER NOT NULL
             );
             CREATE INDEX IF NOT EXISTS scripts_cached_at ON scripts(cached_at DESC);",
        )?;
        let on_disk = std::fs::metadata(&self.path).ok().map(|m| {
            let (ino, len, mtime_ns) = file_identity(&m);
            format!("{ino}:{len}:{mtime_ns}")
        });
        let recorded: Option<String> = conn
            .query_row("SELECT value FROM meta WHERE key = 'shard'", [], |r| r.get(0))
            .optional()?;
        if recorded == on_disk {
            return Ok(());
        }
        let tx = conn.unchecked_transaction()?;
        tx.execute("DELETE FROM scripts", [])?;
        if let Some(shard) = read_owned_shard(&self.path) {
            let mut insert = tx.prepare(
                "INSERT OR REPLACE INTO scripts
                 (path, kind, events, bytes, version, cached_at,
                  source_mtime_secs, source_mtime_nsecs, env_fingerprint, binary_len)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
            )?;
            for (path, e) in &shard.entries {
                let (kind, events) = match e.chunk_blob.first() {
                    // bincode of a `Vec<Chunk>` opens with its u64 length.
                    Some(1) if e.chunk_blob.len() >= 9 => {
                        let n = u64::from_le_bytes(e.chunk_blob[1..9].try_into().unwrap());
                        ("events", n as i64)
                    }
                    Some(0) => ("script", 1),
                    _ => ("unknown", 0),
                };
                insert.execute(rusqlite::params![
                    path,
                    kind,
                    events,
                    e.chunk_blob.len() as i64,
                    shard.header.zshrs_version,
                    e.cached_at_secs,
                    e.mtime_secs,
                    e.mtime_nsecs,
                    format!("{:016x}", e.env_fingerprint),
                    e.binary_len_at_cache as i64,
                ])?;
            }
        }
        match on_disk {
            Some(id) => tx.execute("INSERT OR REPLACE INTO meta (key, value) VALUES ('shard', ?1)", [id])?,
            None => tx.execute("DELETE FROM meta WHERE key = 'shard'", [])?,
        };
        tx.commit()
    }

    /// `(path, chunk_kb, version, cached_at_localstr)` for every cached script
    /// whose path contains `pattern`, newest first — read from `scripts.db`
    /// after refreshing it, plus entries still buffered in this process.
    /// `None` when the mirror cannot be opened; callers fall back to
    /// [`Self::list_scripts`].
    pub fn mirror_rows(&self, pattern: &str) -> Option<Vec<(String, f64, String, String)>> {
        let _lowfd = crate::lowfd::LowFdGuard::new();
        let conn = rusqlite::Connection::open(self.mirror_path()).ok()?;
        conn.busy_timeout(std::time::Duration::from_secs(5)).ok()?;
        let _ = conn.pragma_update(None, "journal_mode", "WAL");
        crate::lowfd::register_internal_fds();
        self.refresh_mirror(&conn).ok()?;
        let mut stmt = conn
            .prepare(
                "SELECT path, bytes, version, cached_at FROM scripts
                 WHERE instr(path, ?1) > 0 ORDER BY cached_at DESC, path",
            )
            .ok()?;
        let mut out: Vec<(String, f64, String, String, i64)> = stmt
            .query_map([pattern], |r| {
                let path: String = r.get(0)?;
                let bytes: i64 = r.get(1)?;
                let version: String = r.get(2)?;
                let cached_at: i64 = r.get(3)?;
                Ok((path, bytes as f64 / 1024.0, version, format_local_ts(cached_at), cached_at))
            })
            .ok()?
            .filter_map(Result::ok)
            .collect();
        self.overlay_runs(&mut out);
        // Buffered, not yet in the shard (see `list_scripts`).
        let now = now_secs();
        for p in PENDING.lock().iter().filter(|p| p.path.contains(pattern)) {
            out.retain(|row| row.0 != p.path);
            out.push((
                p.path.clone(),
                p.blob.len() as f64 / 1024.0,
                env!("CARGO_PKG_VERSION").to_string(),
                format_local_ts(now),
                now,
            ));
        }
        out.sort_by_key(|x| std::cmp::Reverse(x.4));
        Some(out.into_iter().map(|(p, kb, v, ts, _)| (p, kb, v, ts)).collect())
    }

    /// Drop entries whose source file vanished or whose mtime changed.
    pub fn evict_stale(&self) -> usize {
        let _lock = match acquire_lock(&self.lock_path) {
            Some(l) => l,
            None => return 0,
        };
        let mut shard = match read_owned_shard(&self.path) {
            Some(s) => s,
            None => return 0,
        };
        let before = shard.entries.len();
        shard.entries.retain(|p, e| match file_mtime(Path::new(p)) {
            Some((s, ns)) => s == e.mtime_secs && ns == e.mtime_nsecs,
            None => false,
        });
        let evicted = before - shard.entries.len();
        if evicted > 0 {
            let _ = write_shard_atomic(&self.path, &shard);
            self.invalidate_mmap();
        }
        evicted
    }
    /// `clear` — see implementation.
    pub fn clear(&self) -> std::io::Result<()> {
        let _lock = acquire_lock(&self.lock_path);
        let res = match std::fs::remove_file(&self.path) {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(e) => Err(e),
        };
        let _ = std::fs::remove_file(self.runs_path());
        self.invalidate_mmap();
        res
    }
}

/// `(inode, len, mtime_ns)` — what distinguishes one atomically-renamed shard
/// from the next.
fn file_identity(m: &std::fs::Metadata) -> (u64, u64, i64) {
    (m.ino(), m.len(), m.mtime() * 1_000_000_000 + m.mtime_nsec())
}

fn acquire_lock(path: &Path) -> Option<nix::fcntl::Flock<File>> {
    let f = File::options()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(path)
        .ok()?;
    nix::fcntl::Flock::lock(f, nix::fcntl::FlockArg::LockExclusive).ok()
}

fn fresh_shard() -> ScriptShard {
    ScriptShard {
        header: ShardHeader {
            magic: SHARD_MAGIC,
            format_version: SHARD_FORMAT_VERSION,
            zshrs_version: env!("CARGO_PKG_VERSION").to_string(),
            pointer_width: std::mem::size_of::<usize>() as u32,
            built_at_secs: now_secs() as u64,
        },
        entries: HashMap::new(),
    }
}

fn read_owned_shard(path: &Path) -> Option<ScriptShard> {
    let bytes = std::fs::read(path).ok()?;
    let archived = rkyv::check_archived_root::<ScriptShard>(&bytes[..]).ok()?;
    archived.deserialize(&mut rkyv::Infallible).ok()
}

fn write_shard_atomic(path: &Path, shard: &ScriptShard) -> Result<(), String> {
    let bytes = rkyv::to_bytes::<_, 4096>(shard).map_err(|e| format!("rkyv serialize: {}", e))?;
    // Shared with `autoload_cache`: same temp-then-rename scheme, and
    // the same obligation to unlink the temp on a failed write and to
    // reap temps abandoned by processes that died mid-write.
    crate::atomic_write::write_bytes_atomic(path, &bytes)
}

fn now_secs() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

fn format_local_ts(secs: i64) -> String {
    let dt = chrono::DateTime::<chrono::Local>::from(
        UNIX_EPOCH + std::time::Duration::from_secs(secs.max(0) as u64),
    );
    dt.format("%Y-%m-%d %H:%M:%S").to_string()
}
/// `file_mtime` — see implementation.
pub fn file_mtime(path: &Path) -> Option<(i64, i64)> {
    let meta = std::fs::metadata(path).ok()?;
    Some((meta.mtime(), meta.mtime_nsec()))
}

/// `(mtime, len)` of the running zshrs binary — the identity a cached chunk
/// is stamped with, mirroring `autoload_cache::current_binary_identity`.
fn current_binary_identity() -> Option<(i64, u64)> {
    static BIN_ID: OnceLock<Option<(i64, u64)>> = OnceLock::new();
    *BIN_ID.get_or_init(|| {
        let exe = std::env::current_exe().ok()?;
        let meta = std::fs::metadata(&exe).ok()?;
        Some((meta.mtime(), meta.len()))
    })
}

/// Default shard path: `$ZSHRS_HOME/scripts.rkyv` (default
/// `~/.zshrs/scripts.rkyv`).
///
/// This was the one cache that ignored `$ZSHRS_HOME`. Every other
/// store honours it — `autoload_cache::default_cache_path`,
/// `compsys::cache::default_cache_path`, the daemon's `CachePaths`
/// (daemon/paths.rs) — so a test or a session pointed at an isolated
/// home still read and WROTE the real `~/.zshrs/scripts.rkyv`,
/// which is both a leak out of the isolation and a writer the
/// isolated run never accounted for.
pub fn default_cache_path() -> PathBuf {
    let root = if let Some(custom) = std::env::var_os("ZSHRS_HOME") {
        PathBuf::from(custom)
    } else {
        dirs::home_dir()
            .unwrap_or_else(|| PathBuf::from("/tmp"))
            .join(".zshrs")
    };
    root.join("scripts.rkyv")
}

/// Process-local disable flag set by parity-mode flags (`--zsh` etc.)
/// in bins/zshrs.rs. Preferred over `ZSHRS_CACHE=0` in env so the
/// env var doesn't leak into `${(k)parameters}` and inflate the
/// param count vs reference zsh.
pub static CACHE_DISABLED: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

/// `ZSHRS_CACHE=0|false|no` (env) or `CACHE_DISABLED=true` (process-
/// local) disables the cache entirely.
pub fn cache_enabled() -> bool {
    if CACHE_DISABLED.load(std::sync::atomic::Ordering::Relaxed) {
        return false;
    }
    !matches!(
        std::env::var("ZSHRS_CACHE").as_deref(),
        Ok("0") | Ok("false") | Ok("no")
    )
}

/// Process-wide `ScriptCache` rooted at `default_cache_path()`. `None` when the
/// cache is disabled or the path could not be opened.
pub static CACHE: once_cell::sync::Lazy<Option<ScriptCache>> = once_cell::sync::Lazy::new(|| {
    if !cache_enabled() {
        return None;
    }
    ScriptCache::open(&default_cache_path()).ok()
});

/// A program being recorded: the chunks of the events a file's loop has
/// compiled so far. Lives in a thread-local stack (not in the loop's frame) so
/// `zexit` can finish it — a script that ends in `exit` never returns to the
/// loop, and used to leave no cache entry at all.
struct Capture {
    path: PathBuf,
    env_fp: u64,
    /// `ALIAS_GEN` when recording began: each event stores how far the alias
    /// tables had moved from here when it was lexed (see [`GuardedEvent`]).
    alias_base: u64,
    /// A script run by `zshrs FILE` starts from a fresh, deterministic state;
    /// a sourced file runs inside whatever session sourced it.
    toplevel: bool,
    /// The process that began recording. A forked subshell that exits holds a
    /// copy of its parent's frames; it must not store them as finished.
    pid: u32,
    events: Vec<GuardedEvent>,
}

/// One executed event of a recorded file: its compiled chunk, plus what a
/// replay needs to know that the chunk is still the program the lexer would
/// produce now.
///
/// A chunk is lexed text, and lexing depends on the alias tables as the
/// earlier events left them. `alias_delta` is how many alias-table mutations
/// had happened since the file began when this event was lexed; a replay that
/// has made a different number by the time it reaches the event (a conditional
/// `alias` took the other branch) is no longer running the recorded program,
/// and carries on from `start`, the byte offset of the event in the file, by
/// lexing the rest of it afresh.
#[derive(serde::Serialize, serde::Deserialize)]
pub struct GuardedEvent {
    pub chunk: fusevm::Chunk,
    pub alias_delta: u64,
    pub start: u64,
}

thread_local! {
    static CAPTURES: std::cell::RefCell<Vec<Capture>> = const { std::cell::RefCell::new(Vec::new()) };
}

/// Start recording the events of `path`, which the caller is about to run.
pub fn capture_begin(path: &Path, env_fp: u64, toplevel: bool) {
    // Resolved now, not at store time: a script that `cd`s leaves a relative
    // `path` pointing nowhere, and the entry was silently dropped.
    let cap = Capture {
        path: path.canonicalize().unwrap_or_else(|_| path.to_path_buf()),
        env_fp,
        alias_base: env_generation().0,
        toplevel,
        pid: std::process::id(),
        events: Vec::new(),
    };
    CAPTURES.with(|c| c.borrow_mut().push(cap));
}

/// Append the chunk of the event about to run to the innermost recording.
/// `start` is the event's byte offset in the text the loop is reading. Called
/// BEFORE the event runs, so an `exit` inside it still finds it, and with the
/// alias tables exactly as the event was lexed under.
pub fn capture_push(chunk: &fusevm::Chunk, start: usize) {
    let now = env_generation().0;
    CAPTURES.with(|c| {
        if let Some(top) = c.borrow_mut().last_mut() {
            top.events.push(GuardedEvent {
                chunk: chunk.clone(),
                alias_delta: now.wrapping_sub(top.alias_base),
                start: start as u64,
            });
        }
    });
}

/// Finish the innermost recording. `complete` is the caller's verdict that the
/// file ran to its end or to a deliberate `return` / `exit` — not to a parse
/// error or a failed event.
pub fn capture_commit(complete: bool) {
    let Some(cap) = CAPTURES.with(|c| c.borrow_mut().pop()) else {
        return;
    };
    store_capture(cap, complete);
}

/// `zexit` is committed: every file still being recorded stops here, at the
/// event that called `exit`, and a replay stops at the same event. Skipped in a
/// forked child (see [`Capture::pid`]) and when an error flag is live.
pub fn capture_commit_all_on_exit(errored: bool) {
    let caps: Vec<Capture> = CAPTURES.with(|c| std::mem::take(&mut *c.borrow_mut()));
    for cap in caps.into_iter().rev() {
        store_capture(cap, !errored);
    }
}

fn store_capture(cap: Capture, complete: bool) {
    // A file that defines aliases is stored like any other: each event
    // carries the alias-table position it was lexed under (`GuardedEvent`),
    // and a replay that finds the tables elsewhere finishes the file from
    // source. Option changes are replayed as the events that make them.
    if !complete || cap.events.is_empty() || cap.pid != std::process::id() {
        return;
    }
    let _ = cap.toplevel;
    if let Ok(blob) = bincode::serialize(&cap.events) {
        let _ = try_save_bytes(&cap.path, BlobKind::Guarded, cap.env_fp, &blob);
    }
}

/// What a cached blob holds. A top-level script is one chunk; a sourced file is
/// the sequence of per-event chunks the `loop()` of c:Src/init.c:155-220
/// compiled, each lexed under the state the previous ones left behind.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum BlobKind {
    /// bincode of one `fusevm::Chunk`.
    Script = 0,
    /// bincode of a `Vec<fusevm::Chunk>`, one per executed event. No longer
    /// written; a stored blob of this kind is simply a miss.
    Events = 1,
    /// bincode of a `Vec<GuardedEvent>`: the executed events with the
    /// alias-table position each was lexed under and its offset in the file.
    Guarded = 2,
}

/// Hash of everything the lexer and compiler read from the live shell: the
/// alias tables, the option store and the emulation mode. Two runs that start
/// from equal fingerprints lex the same bytes into the same chunks.
///
/// The alias tables are large and change rarely, so their hash is reused until
/// `ALIAS_GEN` moves; the option store is a few hundred bytes and is hashed
/// whenever `opts_cache::generation` moved since the last call.
pub fn env_fingerprint() -> u64 {
    use std::sync::atomic::Ordering;
    static MEMO: Mutex<Option<(u64, u64, u64, u64)>> = Mutex::new(None);
    let alias_gen = crate::ported::hashtable::ALIAS_GEN.load(Ordering::Relaxed);
    let opt_gen = crate::opts_cache::generation();
    let mut memo = MEMO.lock();
    if let Some((ag, og, alias_hash, opt_hash)) = *memo {
        if ag == alias_gen && og == opt_gen {
            return fnv_mix(alias_hash, opt_hash);
        }
    }
    let alias_hash = hash_alias_tables();
    let opt_hash = hash_option_store();
    *memo = Some((alias_gen, opt_gen, alias_hash, opt_hash));
    fnv_mix(alias_hash, opt_hash)
}

/// `(ALIAS_GEN, option generation)` — compared before and after a run to learn
/// whether the run itself changed lexer-visible state.
pub fn env_generation() -> (u64, u64) {
    (
        crate::ported::hashtable::ALIAS_GEN.load(std::sync::atomic::Ordering::Relaxed),
        crate::opts_cache::generation(),
    )
}

fn fnv_mix(a: u64, b: u64) -> u64 {
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    for byte in a.to_le_bytes().into_iter().chain(b.to_le_bytes()) {
        h ^= u64::from(byte);
        h = h.wrapping_mul(0x0000_0100_0000_01b3);
    }
    h
}

fn fnv_bytes(mut h: u64, bytes: &[u8]) -> u64 {
    for &byte in bytes {
        h ^= u64::from(byte);
        h = h.wrapping_mul(0x0000_0100_0000_01b3);
    }
    // Field separator, so ("ab","c") and ("a","bc") hash apart.
    h ^= 0xff;
    h.wrapping_mul(0x0000_0100_0000_01b3)
}

fn hash_alias_tables() -> u64 {
    use crate::ported::hashtable::{aliastab_lock, sufaliastab_lock};
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    for table in [aliastab_lock(), sufaliastab_lock()] {
        let Ok(t) = table.read() else { continue };
        // Bucket order is deterministic for a given insertion history, but the
        // fingerprint must not depend on history — sort by name.
        let mut rows: Vec<(&String, &crate::ported::zsh_h::alias)> = t.iter().collect();
        rows.sort_by(|a, b| a.0.cmp(b.0));
        for (name, a) in rows {
            h = fnv_bytes(h, name.as_bytes());
            h = fnv_bytes(h, a.text.as_bytes());
            h = fnv_bytes(h, &a.node.flags.to_le_bytes());
        }
        h = fnv_bytes(h, b"|");
    }
    h
}

fn hash_option_store() -> u64 {
    let mut rows: Vec<(String, bool)> = crate::ported::options::opt_state_snapshot()
        .into_iter()
        .collect();
    rows.sort();
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    for (name, on) in rows {
        h = fnv_bytes(h, name.as_bytes());
        h = fnv_bytes(h, &[on as u8]);
    }
    let emul = crate::ported::options::emulation.load(std::sync::atomic::Ordering::Relaxed);
    fnv_bytes(h, &emul.to_le_bytes())
}

/// Newest logged run time per path in `scripts.runs`; empty when absent.
fn read_runs(path: &Path) -> HashMap<String, i64> {
    let mut out = HashMap::new();
    let text = std::fs::read(path).map(|b| String::from_utf8_lossy(&b).into_owned()).unwrap_or_default();
    for line in text.lines() {
        let Some((ts, p)) = line.split_once('\t') else { continue };
        let Ok(ts) = ts.parse::<i64>() else { continue };
        let t = out.entry(p.to_string()).or_insert(ts);
        *t = (*t).max(ts);
    }
    out
}

/// Append the buffered hits to `scripts.runs` in one write, then deduplicate
/// the file when it has grown past [`RUNS_COMPACT_BYTES`].
fn append_runs(path: &Path, batch: &[(String, i64)]) {
    let lines: String = batch.iter().map(|(p, ts)| format!("{ts}\t{p}\n")).collect();
    let appended = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .and_then(|mut f| f.write_all(lines.as_bytes()));
    if appended.is_err() || std::fs::metadata(path).map_or(0, |m| m.len()) < RUNS_COMPACT_BYTES {
        return;
    }
    let deduped: String = read_runs(path).into_iter().map(|(p, ts)| format!("{ts}\t{p}\n")).collect();
    let _ = crate::atomic_write::write_bytes_atomic(path, deduped.as_bytes());
}

/// Try to load a cached program by source path. Returns `None` on any miss,
/// including a different lexer-state fingerprint or a blob of another kind.
pub fn try_load_bytes(path: &Path, kind: BlobKind, env_fp: u64) -> Option<Vec<u8>> {
    let cache = CACHE.as_ref()?;
    let canonical = path.canonicalize().ok()?;
    let path_str = canonical.to_string_lossy();
    let (mtime_s, mtime_ns) = file_mtime(&canonical)?;
    // A file sourced twice in one session: the first run's entry is still
    // buffered, and it is the newest word on the path.
    let buffered = PENDING.lock().iter().rev().find_map(|p| {
        (p.path == path_str && p.mtime_secs == mtime_s && p.mtime_nsecs == mtime_ns && p.env_fingerprint == env_fp)
            .then(|| p.blob.clone())
    });
    let blob = match buffered {
        Some(b) => b,
        None => cache.get(&path_str, mtime_s, mtime_ns, env_fp)?,
    };
    let (tag, body) = blob.split_first()?;
    if *tag != kind as u8 {
        return None;
    }
    RAN_PID.store(std::process::id(), std::sync::atomic::Ordering::Relaxed);
    RAN.lock().push((path_str.into_owned(), now_secs()));
    Some(body.to_vec())
}

/// Store the bincode-encoded program for a script path. `env_fp` is the
/// fingerprint taken BEFORE the run: the state the lexer started from, not the
/// state the file left behind. Best-effort — cache disabled / canonicalize
/// failure / mtime stat failure all return `Ok(())` silently so the caller can
/// fire-and-forget.
pub fn try_save_bytes(
    path: &Path,
    kind: BlobKind,
    env_fp: u64,
    chunk_blob: &[u8],
) -> Result<(), String> {
    let Some(cache) = CACHE.as_ref() else {
        return Ok(());
    };
    let canonical = match path.canonicalize() {
        Ok(p) => p,
        Err(_) => return Ok(()),
    };
    let path_str = canonical.to_string_lossy();
    let (mtime_s, mtime_ns) = match file_mtime(&canonical) {
        Some(m) => m,
        None => return Ok(()),
    };
    let mut tagged = Vec::with_capacity(chunk_blob.len() + 1);
    tagged.push(kind as u8);
    tagged.extend_from_slice(chunk_blob);
    PENDING_PID.store(std::process::id(), std::sync::atomic::Ordering::Relaxed);
    let over = {
        let mut pending = PENDING.lock();
        pending.push(PendingPut {
            path: path_str.into_owned(),
            mtime_secs: mtime_s,
            mtime_nsecs: mtime_ns,
            env_fingerprint: env_fp,
            blob: tagged,
        });
        pending.len() >= PENDING_FLUSH_MAX
    };
    if over {
        try_flush_pending();
    }
    Ok(())
}

/// Write out everything `try_save_bytes` buffered, in one shard rewrite. Called
/// at the prompt, from `zexit` and from the `atexit` hook, alongside the
/// autoload and deparse caches. Cheap when nothing is buffered.
pub fn try_flush_pending() {
    // Hits first: they are logged beside the shard, never into it, so they
    // need neither its lock nor a rewrite.
    if RAN_PID.load(std::sync::atomic::Ordering::Relaxed) == std::process::id() {
        let ran = std::mem::take(&mut *RAN.lock());
        if let (false, Some(cache)) = (ran.is_empty(), CACHE.as_ref()) {
            append_runs(&cache.runs_path(), &ran);
        }
    }
    // A forked child leaves its inherited copy alone: the parent writes it, and
    // a `dbview scripts` in the child still lists it from the buffer.
    if PENDING_PID.load(std::sync::atomic::Ordering::Relaxed) != std::process::id() {
        return;
    }
    let batch = std::mem::take(&mut *PENDING.lock());
    if batch.is_empty() {
        return;
    }
    if let Some(cache) = CACHE.as_ref() {
        let _ = cache.put_many(batch);
    }
}
/// `stats` — see implementation.
pub fn stats() -> Option<(i64, i64)> {
    CACHE.as_ref().map(|c| c.stats())
}
/// `evict_stale` — see implementation.
pub fn evict_stale() -> usize {
    CACHE.as_ref().map(|c| c.evict_stale()).unwrap_or(0)
}
/// `clear` — see implementation.
pub fn clear() -> bool {
    CACHE.as_ref().map(|c| c.clear().is_ok()).unwrap_or(false)
}

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

    #[test]
    fn round_trip() {
        let _g = crate::test_util::global_state_lock();
        let dir = tempdir().unwrap();
        let cache_path = dir.path().join("scripts.rkyv");
        let cache = ScriptCache::open(&cache_path).unwrap();

        let script_path = dir.path().join("test.zsh");
        std::fs::write(&script_path, "echo hi").unwrap();

        let (mtime_s, mtime_ns) = file_mtime(&script_path).unwrap();
        let path_str = script_path.to_string_lossy().to_string();

        let blob = vec![1u8, 2, 3, 4, 5];
        cache
            .put(&path_str, mtime_s, mtime_ns, 0, blob.clone())
            .unwrap();

        let loaded = cache.get(&path_str, mtime_s, mtime_ns, 0).unwrap();
        assert_eq!(loaded, blob);

        let (count, _bytes) = cache.stats();
        assert_eq!(count, 1);
    }

    /// `dbview scripts` in a long-lived shell reads through a mapping taken
    /// earlier. A script run by ANOTHER process replaces the shard by rename,
    /// which this process's mapping never sees on its own: the listing froze
    /// at whatever the shard held when it was first read.
    #[test]
    fn a_write_from_another_process_shows_up_in_a_mapped_reader() {
        let _g = crate::test_util::global_state_lock();
        let dir = tempdir().unwrap();
        let cache_path = dir.path().join("scripts.rkyv");
        let reader = ScriptCache::open(&cache_path).unwrap();
        let writer = ScriptCache::open(&cache_path).unwrap();

        let first = dir.path().join("first.zsh");
        let second = dir.path().join("second.zsh");
        std::fs::write(&first, "echo 1").unwrap();
        std::fs::write(&second, "echo 2").unwrap();
        let (s1, n1) = file_mtime(&first).unwrap();
        let (s2, n2) = file_mtime(&second).unwrap();

        writer.put(&first.to_string_lossy(), s1, n1, 0, vec![1]).unwrap();
        assert_eq!(reader.list_scripts().len(), 1, "reader maps the shard");

        writer.put(&second.to_string_lossy(), s2, n2, 0, vec![2]).unwrap();
        let rows = reader.list_scripts();
        assert_eq!(rows.len(), 2, "the replaced shard must be remapped: {rows:?}");
        assert_eq!(reader.stats().0, 2);
    }

    /// `dbview scripts` reads `scripts.db`, which is derived from the shard and
    /// rebuilt whenever the shard's identity moved: new entries appear, a
    /// replaced entry shows its new state, and a deleted shard empties it.
    #[test]
    fn the_sqlite_mirror_follows_the_shard() {
        let _g = crate::test_util::global_state_lock();
        let dir = tempdir().unwrap();
        let cache = ScriptCache::open(&dir.path().join("scripts.rkyv")).unwrap();
        let one = dir.path().join("one.zsh");
        let two = dir.path().join("two.zsh");
        std::fs::write(&one, "echo 1").unwrap();
        std::fs::write(&two, "echo 2").unwrap();
        let (s1, n1) = file_mtime(&one).unwrap();
        let (s2, n2) = file_mtime(&two).unwrap();

        assert_eq!(cache.mirror_rows("").unwrap().len(), 0, "empty before any write");

        cache.put(&one.to_string_lossy(), s1, n1, 0, vec![1u8]).unwrap();
        let rows = cache.mirror_rows("").unwrap();
        assert_eq!(rows.len(), 1);
        assert!(rows[0].0.ends_with("one.zsh"));
        assert!(dir.path().join("scripts.db").is_file());

        cache.put(&two.to_string_lossy(), s2, n2, 0, vec![2u8; 4096]).unwrap();
        assert_eq!(cache.mirror_rows("").unwrap().len(), 2);
        let only_two = cache.mirror_rows("two.zsh").unwrap();
        assert_eq!(only_two.len(), 1);
        assert!(only_two[0].1 > 3.9, "kb is the blob size: {only_two:?}");

        cache.clear().unwrap();
        assert_eq!(cache.mirror_rows("").unwrap().len(), 0, "deleted shard empties the mirror");
    }

    /// A script replayed from the cache is a hit, which leaves the shard
    /// alone; `dbview scripts` must still list it as run at the hit's time,
    /// and the log must be carried into the shard (not lost) when it is
    /// next rewritten.
    #[test]
    fn a_cache_hit_moves_the_listed_time_and_survives_a_rewrite() {
        let _g = crate::test_util::global_state_lock();
        let dir = tempdir().unwrap();
        let cache = ScriptCache::open(&dir.path().join("scripts.rkyv")).unwrap();
        let one = dir.path().join("one.zsh");
        let two = dir.path().join("two.zsh");
        std::fs::write(&one, "echo 1").unwrap();
        std::fs::write(&two, "echo 2").unwrap();
        let (s1, n1) = file_mtime(&one).unwrap();
        let (s2, n2) = file_mtime(&two).unwrap();
        let one_key = one.to_string_lossy().to_string();
        cache.put(&one_key, s1, n1, 0, vec![1u8]).unwrap();
        cache.put(&two.to_string_lossy(), s2, n2, 0, vec![2u8]).unwrap();

        let ran_at = now_secs() + 1000;
        append_runs(&cache.runs_path(), &[(one_key.clone(), ran_at)]);

        let by_path = |rows: Vec<(String, f64, String, String)>| {
            rows.into_iter().map(|r| (r.0, r.3)).collect::<Vec<_>>()
        };
        let listed = by_path(cache.list_scripts());
        assert!(listed[0].0.ends_with("one.zsh"), "newest run first: {listed:?}");
        assert_eq!(listed[0].1, format_local_ts(ran_at));
        let mirrored = by_path(cache.mirror_rows("").unwrap());
        assert_eq!(mirrored[0], listed[0]);

        // A later store folds the logged run into the entry and drops the log.
        let three = dir.path().join("three.zsh");
        std::fs::write(&three, "echo 3").unwrap();
        let (s3, n3) = file_mtime(&three).unwrap();
        cache.put(&three.to_string_lossy(), s3, n3, 0, vec![3u8]).unwrap();
        assert!(!cache.runs_path().exists());
        let after = by_path(cache.list_scripts());
        assert!(after[0].0.ends_with("one.zsh"), "{after:?}");
        assert_eq!(after[0].1, format_local_ts(ran_at));
    }

    #[test]
    fn mtime_invalidation() {
        let _g = crate::test_util::global_state_lock();
        let dir = tempdir().unwrap();
        let cache_path = dir.path().join("scripts.rkyv");
        let cache = ScriptCache::open(&cache_path).unwrap();

        let script_path = dir.path().join("test.zsh");
        std::fs::write(&script_path, "echo hi").unwrap();

        let (mtime_s, mtime_ns) = file_mtime(&script_path).unwrap();
        let path_str = script_path.to_string_lossy().to_string();
        cache.put(&path_str, mtime_s, mtime_ns, 0, vec![9u8]).unwrap();

        assert!(cache.get(&path_str, mtime_s + 1, mtime_ns, 0).is_none());
    }

    /// Bytecode is not portable between builds, so an entry must be refused
    /// whichever way the recorded binary timestamp points. The old
    /// `cached < running` test only caught the "cache written by an older
    /// build" direction, which meant a binary whose mtime moved BACKWARDS
    /// (an earlier build restored over a later one, a `cp -p`, a checkout of
    /// a previously-built target dir) replayed the newer build's chunks.
    #[test]
    fn an_entry_from_another_binary_is_never_served() {
        let _g = crate::test_util::global_state_lock();
        let dir = tempdir().unwrap();
        let cache_path = dir.path().join("scripts.rkyv");
        let cache = ScriptCache::open(&cache_path).unwrap();

        let script_path = dir.path().join("test.zsh");
        std::fs::write(&script_path, "echo hi").unwrap();
        let (mtime_s, mtime_ns) = file_mtime(&script_path).unwrap();
        let path_str = script_path.to_string_lossy().to_string();
        cache.put(&path_str, mtime_s, mtime_ns, 0, vec![9u8]).unwrap();
        assert_eq!(
            cache.get(&path_str, mtime_s, mtime_ns, 0),
            Some(vec![9u8]),
            "the emitting binary must hit its own entry",
        );

        // Restamp the entry as if a NEWER build had produced it — the
        // direction the old comparison let through.
        let mut shard = read_owned_shard(&cache_path).expect("shard readable");
        shard
            .entries
            .get_mut(&path_str)
            .expect("entry present")
            .binary_mtime_at_cache += 10_000;
        write_shard_atomic(&cache_path, &shard).unwrap();
        let reopened = ScriptCache::open(&cache_path).unwrap();
        assert!(
            reopened.get(&path_str, mtime_s, mtime_ns, 0).is_none(),
            "a chunk from a newer build was accepted",
        );

        // Same length, same-second mtime, different build: only the length
        // term can reject this one.
        let mut shard = read_owned_shard(&cache_path).expect("shard readable");
        let entry = shard.entries.get_mut(&path_str).expect("entry present");
        entry.binary_mtime_at_cache -= 10_000;
        entry.binary_len_at_cache += 1;
        write_shard_atomic(&cache_path, &shard).unwrap();
        let reopened = ScriptCache::open(&cache_path).unwrap();
        assert!(
            reopened.get(&path_str, mtime_s, mtime_ns, 0).is_none(),
            "a chunk from a same-second build of a different size was accepted",
        );
    }

    #[test]
    fn second_put_replaces_first() {
        let _g = crate::test_util::global_state_lock();
        let dir = tempdir().unwrap();
        let cache_path = dir.path().join("scripts.rkyv");
        let cache = ScriptCache::open(&cache_path).unwrap();

        let p1 = dir.path().join("a.zsh");
        let p2 = dir.path().join("b.zsh");
        std::fs::write(&p1, "1").unwrap();
        std::fs::write(&p2, "2").unwrap();

        let (m1s, m1n) = file_mtime(&p1).unwrap();
        let (m2s, m2n) = file_mtime(&p2).unwrap();

        cache
            .put(&p1.to_string_lossy(), m1s, m1n, 0, vec![1u8])
            .unwrap();
        cache
            .put(&p2.to_string_lossy(), m2s, m2n, 0, vec![2u8])
            .unwrap();

        let (count, _) = cache.stats();
        assert_eq!(count, 2);
        assert!(cache.get(&p1.to_string_lossy(), m1s, m1n, 0).is_some());
        assert!(cache.get(&p2.to_string_lossy(), m2s, m2n, 0).is_some());
    }

    #[test]
    fn corrupt_file_returns_no_mmap() {
        let _g = crate::test_util::global_state_lock();
        let dir = tempdir().unwrap();
        let cache_path = dir.path().join("scripts.rkyv");
        std::fs::write(&cache_path, b"this is not a valid rkyv archive").unwrap();
        let cache = ScriptCache::open(&cache_path).unwrap();
        assert!(cache.get("/nope", 0, 0, 0).is_none());
    }

    #[test]
    fn clear_removes_file() {
        let _g = crate::test_util::global_state_lock();
        let dir = tempdir().unwrap();
        let cache_path = dir.path().join("scripts.rkyv");
        let cache = ScriptCache::open(&cache_path).unwrap();

        let script_path = dir.path().join("test.zsh");
        std::fs::write(&script_path, "echo hi").unwrap();
        let (mtime_s, mtime_ns) = file_mtime(&script_path).unwrap();
        cache
            .put(&script_path.to_string_lossy(), mtime_s, mtime_ns, 0, vec![7u8])
            .unwrap();
        assert!(cache_path.exists());

        cache.clear().unwrap();
        assert!(!cache_path.exists());
    }

    // ========================================================
    // now_secs — monotonic-ish wall-clock
    // ========================================================

    #[test]
    fn now_secs_is_positive_and_within_realistic_range() {
        let _g = crate::test_util::global_state_lock();
        let n = now_secs();
        // Year 2020 = ~1.58e9 seconds. Year 2100 = ~4.1e9 seconds.
        assert!(
            (1_577_836_800..4_102_444_800).contains(&n),
            "now_secs out of plausible range: {}",
            n
        );
    }

    #[test]
    fn now_secs_does_not_go_backwards_in_quick_succession() {
        let _g = crate::test_util::global_state_lock();
        let a = now_secs();
        let b = now_secs();
        assert!(b >= a, "now_secs went backwards: {} -> {}", a, b);
    }

    // ========================================================
    // format_local_ts — human-readable timestamp
    // ========================================================

    #[test]
    fn format_local_ts_includes_year_and_punctuation() {
        let _g = crate::test_util::global_state_lock();
        // 2024-01-01 = 1704067200 UTC; local timezone shifts it but
        // the year prefix is stable regardless of TZ.
        let s = format_local_ts(1_704_067_200);
        assert!(s.starts_with("202"), "expected 21st century year: {}", s);
        assert!(s.contains('-'), "expected dash separator: {}", s);
        assert!(s.contains(':'), "expected colon separator: {}", s);
    }

    #[test]
    fn format_local_ts_length_matches_pattern() {
        let _g = crate::test_util::global_state_lock();
        let s = format_local_ts(1_700_000_000);
        // `YYYY-MM-DD HH:MM:SS` = 19 chars.
        assert_eq!(s.len(), 19, "unexpected width: {}", s);
    }

    #[test]
    fn format_local_ts_handles_zero_secs_via_clamp() {
        let _g = crate::test_util::global_state_lock();
        // 0 → 1970-01-01 in UTC, but local TZ may shift the date.
        let s = format_local_ts(0);
        assert_eq!(s.len(), 19);
        assert!(s.starts_with("19"), "expected 1970-ish year: {}", s);
    }

    #[test]
    fn format_local_ts_negative_clamped_to_zero() {
        let _g = crate::test_util::global_state_lock();
        // .max(0) prevents negative seconds reaching chrono.
        let s = format_local_ts(-1_000_000);
        assert_eq!(s.len(), 19);
    }

    // ========================================================
    // file_mtime — pure metadata sniff
    // ========================================================

    #[test]
    fn file_mtime_returns_some_for_real_file() {
        let _g = crate::test_util::global_state_lock();
        let dir = tempdir().unwrap();
        let p = dir.path().join("foo.zsh");
        std::fs::write(&p, b"x").unwrap();
        let (s, _ns) = file_mtime(&p).unwrap();
        assert!(s > 0);
    }

    #[test]
    fn file_mtime_returns_none_for_missing_path() {
        let _g = crate::test_util::global_state_lock();
        assert!(file_mtime(Path::new("/nonexistent/zshrs/script_cache_missing.bin")).is_none());
    }

    // ========================================================
    // default_cache_path / cache_enabled — config knobs
    // ========================================================

    #[test]
    fn default_cache_path_ends_in_scripts_rkyv() {
        let _g = crate::test_util::global_state_lock();
        let p = default_cache_path();
        assert_eq!(p.file_name().and_then(|s| s.to_str()), Some("scripts.rkyv"));
    }

    #[test]
    fn cache_enabled_true_when_env_unset() {
        let _g = crate::test_util::global_state_lock();
        let prev = std::env::var_os("ZSHRS_CACHE");
        std::env::remove_var("ZSHRS_CACHE");
        let on = cache_enabled();
        if let Some(v) = prev {
            std::env::set_var("ZSHRS_CACHE", v);
        }
        assert!(on, "cache should be enabled when ZSHRS_CACHE is unset");
    }

    #[test]
    fn cache_enabled_false_when_env_is_zero_false_or_no() {
        let _g = crate::test_util::global_state_lock();
        let prev = std::env::var_os("ZSHRS_CACHE");
        for v in ["0", "false", "no"] {
            std::env::set_var("ZSHRS_CACHE", v);
            assert!(!cache_enabled(), "ZSHRS_CACHE={} must disable cache", v);
        }
        if let Some(v) = prev {
            std::env::set_var("ZSHRS_CACHE", v);
        } else {
            std::env::remove_var("ZSHRS_CACHE");
        }
    }

    #[test]
    fn cache_enabled_true_for_other_env_values() {
        // Truthiness model: only `0|false|no` disable. Anything else
        // (including empty string) leaves the cache on.
        let _g = crate::test_util::global_state_lock();
        let prev = std::env::var_os("ZSHRS_CACHE");
        for v in ["1", "true", "yes", "on", ""] {
            std::env::set_var("ZSHRS_CACHE", v);
            assert!(
                cache_enabled(),
                "ZSHRS_CACHE={:?} must NOT disable cache",
                v
            );
        }
        if let Some(v) = prev {
            std::env::set_var("ZSHRS_CACHE", v);
        } else {
            std::env::remove_var("ZSHRS_CACHE");
        }
    }
}