Skip to main content

mnemo_engine/
store.rs

1//! The Mnemo database — ties the storage engine and the memory model together.
2//!
3//! Layout of a populated file:
4//!
5//! ```text
6//!   page 0           : Header (unencrypted)
7//!   pages 1..W       : write-ahead log region
8//!   pages W..        : encrypted record runs + catalog runs + index +
9//!                      snapshot manifest (append-only)
10//! ```
11//!
12//! Durability is provided by a **write-ahead log** ([`crate::wal`]). A `flush`
13//! is one transaction: record data pages are written copy-on-write to fresh
14//! pages, then the new catalog, ANN index, and header are logged to the WAL
15//! and committed with a single fsync. A checkpoint then folds the WAL into the
16//! home pages. A crash before the commit leaves the previous state intact; a
17//! crash after it is repaired by replaying the WAL on open.
18//!
19//! Because pages are only ever appended, every past transaction's runs survive
20//! on disk. Each flush also appends an entry to a **snapshot manifest**, so
21//! [`Mnemo::restore_to`] can reinstate any past state. Stale pages — and the
22//! history along with them — are reclaimed by [`Mnemo::compact_file`].
23
24use std::collections::HashMap;
25use std::fs::{File, OpenOptions};
26use std::io::{Read, Seek, SeekFrom, Write};
27use std::path::{Path, PathBuf};
28
29use fs4::fs_std::FileExt;
30use serde::{Deserialize, Serialize};
31use ulid::Ulid;
32
33use crate::crypto::{self, KdfParams};
34use crate::embed::{self, Embedder};
35use crate::error::{MnemoError, Result};
36use crate::format::{Header, FLAG_ENCRYPTED, PAGE_SIZE, PAYLOAD, VERSION, WRAPPED_DEK_LEN};
37use crate::index::{IndexConfig, IndexInfo, IvfPqIndex};
38use crate::memory::{self, Memory, MemoryType, Metric, Scope, ScoreWeights};
39use crate::pager::Pager;
40use crate::result_cache::{
41    self, BatchState, CacheBudget, CacheDirectoryEntry, CacheEntry, CacheFlushPolicy, CachePutOpts,
42    CacheStats, CachedValue, CounterState, DirectoryIndex, SemanticCachePutOpts,
43};
44use crate::wal;
45
46/// Default initial size of the write-ahead log region, in pages (64 KiB).
47/// The region grows automatically when a transaction's control plane outgrows
48/// it, so this is a *floor* on the fresh-file footprint, not a cap on
49/// transaction size. Was 64 pages (512 KiB) in v0.1.0 — lowered because that
50/// reservation dominated small-file size (~62% of a 31-memory dogfood file).
51const DEFAULT_WAL_PAGES: u64 = 8;
52
53/// Hard floor on the WAL reservation. A single header-page flush already
54/// needs one full DATA frame plus a COMMIT frame; two pages are the bare
55/// minimum that lets the first transaction commit without immediate growth.
56const MIN_WAL_PAGES: u64 = 2;
57
58/// Default cap on retained snapshot manifest entries. Each `flush` appends
59/// one entry; without a cap, long-running processes accumulate entries
60/// forever, growing per-flush manifest-serialize cost as O(total flushes)
61/// and the manifest run size on disk in lockstep. 256 keeps roughly a
62/// week of hourly snapshots while staying tiny (256 × ~82 bytes ≈ 21 KiB
63/// of manifest). Set `MnemoConfig::max_snapshots = 0` to disable the cap.
64pub const DEFAULT_MAX_SNAPSHOTS: usize = 256;
65
66/// Chunk size for the torn-header scan's streaming read of the file.
67/// We walk the file 1 MiB at a time looking for 8 KiB-aligned `MWAL`
68/// magic. There is **no** cap on how far a given candidate's frame
69/// chain may extend — a 1M-vector IVF+PQ relocation can push a single
70/// transaction well past 16 MiB — so each candidate is read frame by
71/// frame via [`wal::scan_committed_txn`] rather than slurped into a
72/// bounded buffer. (DUR-1 R-002.1 part 2.)
73const TORN_HEADER_SCAN_CHUNK: usize = 1 << 20; // 1 MiB
74
75/// Walk the file for every 8 KiB-aligned page whose first four bytes are
76/// `MWAL`, treating each as a candidate WAL-region start, and return the
77/// frames of the highest-`txn_id` committed transaction whose header
78/// frame fully validates under `passphrase`.
79///
80/// A candidate is accepted only if:
81/// * `wal::scan_committed_txn` returns a committed transaction (chained
82///   CRC verifies, single `txn_id`, `KIND_COMMIT` closes it);
83/// * that transaction contains a `(0, header_page)` frame;
84/// * the header parses with `wal_start == candidate_page_no` and
85///   `wal_seq == txn_id` (so a frame cannot forge a region it was not
86///   written into);
87/// * deriving the KEK from `passphrase + header.salt`, unwrapping the
88///   wrapped DEK, and validating the header seal all succeed.
89///
90/// Returns `Ok(None)` when the scan finds no valid candidate — the
91/// caller is expected to propagate the original parse error so
92/// non-mnemo files still report `BadMagic`.
93///
94/// DUR-1 (R-002.1 part 2): there is no fixed scan window — the whole
95/// file is scanned. Reading at page granularity (not byte granularity)
96/// is sufficient because every WAL region in the format is page-aligned
97/// and every WAL commit writes its first frame at the region's start.
98fn scan_header_recovery_candidates(
99    file: &mut File,
100    passphrase: &str,
101) -> Result<Option<Vec<wal::Frame>>> {
102    let file_len = file.seek(SeekFrom::End(0))?;
103    if file_len < PAGE_SIZE as u64 * 2 {
104        return Ok(None); // No room for even one WAL page past the header.
105    }
106
107    let mut best: Option<(u64, Vec<wal::Frame>)> = None;
108    let page_sz = PAGE_SIZE as u64;
109    let mut offset = page_sz; // Skip page 0.
110
111    // Stream through the file in 1 MiB chunks, scanning each chunk for
112    // page-aligned `MWAL` candidates. Each candidate's frame chain is
113    // then read incrementally via `wal::scan_committed_txn`, which has
114    // no window cap — a 1M-vector IVF+PQ relocation can span tens of
115    // MiB in one committed transaction.
116    let mut chunk = vec![0u8; TORN_HEADER_SCAN_CHUNK];
117    while offset < file_len {
118        let want = (TORN_HEADER_SCAN_CHUNK as u64).min(file_len - offset) as usize;
119        chunk.resize(want, 0);
120        file.seek(SeekFrom::Start(offset))?;
121        let mut filled = 0;
122        while filled < want {
123            match file.read(&mut chunk[filled..])? {
124                0 => break,
125                n => filled += n,
126            }
127        }
128        if filled == 0 {
129            break;
130        }
131        let mut inner = 0usize;
132        while inner + 4 <= filled {
133            if &chunk[inner..inner + 4] == wal::FRAME_MAGIC_BYTES {
134                let page_no = (offset + inner as u64) / page_sz;
135                if let Some((txn_id, frames)) = try_recover_candidate(file, page_no, passphrase)? {
136                    let keep = match &best {
137                        None => true,
138                        Some((id, _)) => txn_id > *id,
139                    };
140                    if keep {
141                        best = Some((txn_id, frames));
142                    }
143                }
144            }
145            inner += PAGE_SIZE;
146        }
147        offset += filled as u64;
148    }
149
150    Ok(best.map(|(_, f)| f))
151}
152
153/// Try to recover a committed transaction from `candidate_page_no`,
154/// returning `(txn_id, frames)` only if the candidate's
155/// `(0, header_page)` frame fully validates: the frame chain commits
156/// (via [`wal::scan_committed_txn`]'s streaming reader), the header
157/// parses with `wal_start == candidate_page_no` **and**
158/// `wal_seq == txn_id`, the DEK unwraps under `passphrase`, and the
159/// AES-GCM seal validates. DUR-1 (R-002.1 item 2) requires the
160/// `wal_seq == txn_id` check and no fixed scan window — the streaming
161/// reader handles transactions of arbitrary size.
162fn try_recover_candidate(
163    file: &mut File,
164    candidate_page_no: u64,
165    passphrase: &str,
166) -> Result<Option<(u64, Vec<wal::Frame>)>> {
167    let Some((txn_id, frames)) = wal::scan_committed_txn(file, candidate_page_no)? else {
168        return Ok(None);
169    };
170
171    // Must contain a header frame at page 0.
172    let header_bytes = match frames.iter().find(|(p, _)| *p == 0) {
173        Some((_, b)) => b,
174        None => return Ok(None),
175    };
176    if header_bytes.len() != PAGE_SIZE {
177        return Ok(None);
178    }
179    let mut hbuf = [0u8; PAGE_SIZE];
180    hbuf.copy_from_slice(header_bytes);
181    let candidate = match Header::from_page(&hbuf) {
182        Ok(h) => h,
183        Err(_) => return Ok(None),
184    };
185
186    // The frame's claimed region must match where we found it, and the
187    // header frame's `wal_seq` must match the committed transaction's
188    // `txn_id` — this is what pins the frame to the region it was
189    // written into and keeps a stale checkpointed-and-copied frame
190    // from being accepted as a new region's root.
191    if candidate.wal_start != candidate_page_no || candidate.wal_seq != txn_id {
192        return Ok(None);
193    }
194    // DEK unwrap under this candidate's own salt + KDF params.
195    let kdf = KdfParams {
196        m_cost: candidate.m_cost,
197        t_cost: candidate.t_cost,
198        p_cost: candidate.p_cost,
199    };
200    let kek = match crypto::derive_kek(passphrase.as_bytes(), &candidate.salt, kdf) {
201        Ok(k) => k,
202        Err(_) => return Ok(None),
203    };
204    let dek = match crypto::unwrap_dek(&kek, &candidate.dek_nonce, &candidate.wrapped_dek) {
205        Ok(d) => d,
206        Err(_) => return Ok(None),
207    };
208    if candidate.validate_seal(&dek).is_err() {
209        return Ok(None);
210    }
211
212    Ok(Some((txn_id, frames)))
213}
214
215/// Configuration for creating a new database.
216#[derive(Clone, Copy, Debug)]
217pub struct MnemoConfig {
218    /// Embedding dimensionality. Every stored vector must match this.
219    pub dimensions: usize,
220    /// Key-derivation parameters.
221    pub kdf: KdfParams,
222    /// Initial size of the WAL region in 8 KiB pages. Default is
223    /// [`DEFAULT_WAL_PAGES`] (8 pages = 64 KiB), which keeps fresh files
224    /// small; the region auto-grows when a transaction's control plane needs
225    /// more space, so raising this is only a hint for the expected steady-state
226    /// transaction size. Write-heavy databases that routinely flush large
227    /// catalogs or ANN indexes can set this higher (e.g. 64 = 512 KiB) to
228    /// avoid early grow events; tiny embedded uses can leave it at the
229    /// default. Clamped to a minimum of [`MIN_WAL_PAGES`] (2 pages).
230    pub wal_pages_initial: u64,
231    /// Maximum number of snapshot manifest entries to retain. Each `flush`
232    /// appends one entry; once the cap is hit, the *oldest* entry is dropped
233    /// from the in-memory manifest before the new one is added, so the
234    /// retained set is always the most-recent N. Defaults to
235    /// [`DEFAULT_MAX_SNAPSHOTS`] (256). Set to `0` to disable the cap and
236    /// retain every snapshot forever (the pre-v0.3 behavior).
237    ///
238    /// Pages referenced *only* by pruned snapshots stay on disk until the
239    /// next [`Mnemo::compact_file`], which reclaims them. Point-in-time
240    /// recovery via [`Mnemo::restore_to`] into a pruned `txn_id` returns
241    /// [`MnemoError::NotFound`].
242    ///
243    /// Apply to an already-open database with [`Mnemo::set_max_snapshots`].
244    pub max_snapshots: usize,
245}
246
247impl Default for MnemoConfig {
248    fn default() -> Self {
249        Self {
250            dimensions: 768,
251            kdf: KdfParams::secure(),
252            wal_pages_initial: DEFAULT_WAL_PAGES,
253            max_snapshots: DEFAULT_MAX_SNAPSHOTS,
254        }
255    }
256}
257
258/// Catalog entry: maps a memory ID to its page run on disk.
259///
260/// **Schema (v5+):** carries `accessed_at` and `access_count` so `recall`
261/// can update access stats by mutating the catalog entry rather than
262/// rewriting the entire record (Phase 2.1 of the improvement plan). The
263/// values in a memory's serialized record body become a stale snapshot
264/// after the first recall; [`Mnemo::read_memory`] re-populates them from
265/// the catalog so consumers always see the live values.
266#[derive(Serialize, Deserialize, Clone, Debug)]
267struct CatalogEntry {
268    /// ULID as a raw `u128`.
269    id: u128,
270    start_page: u64,
271    page_count: u32,
272    /// Exact serialized byte length of the record.
273    len: u32,
274    deleted: bool,
275    /// Unix seconds of the most recent recall hit (or write, for fresh entries).
276    accessed_at: i64,
277    /// How many times `recall` has surfaced this memory.
278    access_count: u32,
279}
280
281/// Frozen v4 catalog entry shape, used only by the v4→v5 migration path in
282/// [`Mnemo::open`]. Its layout matches the catalog encoding written by
283/// pre-v5 builds (5 positional fields). Do not change this struct.
284#[derive(Serialize, Deserialize, Clone, Debug)]
285struct CatalogEntryV4 {
286    id: u128,
287    start_page: u64,
288    page_count: u32,
289    len: u32,
290    deleted: bool,
291}
292
293/// Returned by [`Mnemo::prepare_for_flush`] and consumed by [`Mnemo::flush`].
294/// Carries the already-serialized control-plane bytes through the prelude
295/// so we don't re-serialize after the lease has been persisted.
296struct FlushPrelude {
297    cat_bytes: Option<Vec<u8>>,
298    idx_bytes: Option<Vec<u8>>,
299    /// Serialized result-cache directory bytes when `dirty_cache` is
300    /// set (Phase 10.1). `None` if the cache didn't change since the
301    /// last flush — in that case the header keeps its current
302    /// `cache_start` / `cache_pages` / `cache_len` and no fresh cache
303    /// run is written.
304    cache_bytes: Option<Vec<u8>>,
305}
306
307/// One entry in the append-only snapshot manifest. Because record, catalog,
308/// and index pages are only ever *appended*, the runs a past flush wrote are
309/// still on disk; a `Snapshot` is the set of pointers needed to reconstruct
310/// the database exactly as that flush left it.
311#[derive(Serialize, Deserialize, Clone, Debug)]
312struct Snapshot {
313    txn_id: u64,
314    created_at: i64,
315    catalog_start: u64,
316    catalog_pages: u64,
317    catalog_len: u64,
318    index_start: u64,
319    index_pages: u64,
320    index_len: u64,
321    memory_count: u64,
322}
323
324/// A restorable point in a database's history — see [`Mnemo::snapshots`].
325#[derive(Clone, Copy, Debug)]
326pub struct SnapshotInfo {
327    /// Id of the transaction that produced this snapshot (monotonic from 1).
328    pub txn_id: u64,
329    /// When the snapshot was committed (unix seconds).
330    pub created_at: i64,
331    /// Live memory count captured in the snapshot.
332    pub memory_count: u64,
333}
334
335/// A retrieval request for [`Mnemo::recall`].
336#[derive(Clone, Debug)]
337pub struct RecallRequest {
338    /// Query embedding.
339    pub query: Vec<f32>,
340    /// Maximum results to return.
341    pub top_k: usize,
342    /// Restrict to these memory types (`None` = all types).
343    pub memory_types: Option<Vec<MemoryType>>,
344    /// Restrict to a single agent's view: its own memories plus shared ones.
345    pub agent_id: Option<String>,
346    /// Similarity metric.
347    pub metric: Metric,
348    /// Multi-signal score weights.
349    pub weights: ScoreWeights,
350    /// Index override: partitions to probe (`None` = index default). Ignored
351    /// when no ANN index is present.
352    pub n_probe: Option<usize>,
353    /// Index override: candidates to rerank (`None` = index default). Ignored
354    /// when no ANN index is present.
355    pub n_rerank: Option<usize>,
356    /// Whether to update each result's `accessed_at` and `access_count` in
357    /// the catalog. Defaults to `true`, matching pre-v5 behavior.
358    ///
359    /// Set to `false` for **fully read-only recall** — useful for batch
360    /// scoring, dry-run scoring, or recalls run by introspection tooling
361    /// (e.g. `mnemo about` consumers) where you don't want the score's own
362    /// observation to perturb the database. With `false`, recall does not
363    /// dirty the catalog and the next `flush` is a no-op.
364    pub track_access: bool,
365}
366
367impl RecallRequest {
368    /// A request with sensible defaults for a query embedding.
369    pub fn new(query: Vec<f32>) -> Self {
370        Self {
371            query,
372            top_k: 10,
373            memory_types: None,
374            agent_id: None,
375            metric: Metric::Cosine,
376            weights: ScoreWeights::default(),
377            n_probe: None,
378            n_rerank: None,
379            track_access: true,
380        }
381    }
382
383    /// Build a request by embedding `query_text` with the supplied
384    /// embedder (Phase 3). Shorthand for `embedder.embed(&[text])` +
385    /// [`RecallRequest::new`]; same defaults as `new`. Fails if the
386    /// embedder returns an empty vector list or an embedding failure.
387    pub fn from_text(embedder: &dyn Embedder, query_text: &str) -> Result<Self> {
388        let mut vectors = embedder.embed(&[query_text])?;
389        let v = vectors.pop().ok_or_else(|| {
390            MnemoError::Invalid("embedder returned no vector for the query text".into())
391        })?;
392        Ok(Self::new(v))
393    }
394    /// Set the result cap.
395    pub fn top_k(mut self, k: usize) -> Self {
396        self.top_k = k;
397        self
398    }
399    /// Restrict to specific memory types.
400    pub fn types(mut self, t: Vec<MemoryType>) -> Self {
401        self.memory_types = Some(t);
402        self
403    }
404    /// Restrict to one agent's view.
405    pub fn agent(mut self, agent_id: impl Into<String>) -> Self {
406        self.agent_id = Some(agent_id.into());
407        self
408    }
409    /// Set the similarity metric (default: cosine).
410    pub fn metric(mut self, metric: Metric) -> Self {
411        self.metric = metric;
412        self
413    }
414    /// Replace the multi-signal score weights.
415    pub fn weights(mut self, weights: ScoreWeights) -> Self {
416        self.weights = weights;
417        self
418    }
419    /// Override the number of IVF partitions probed (accuracy/speed dial).
420    pub fn n_probe(mut self, n: usize) -> Self {
421        self.n_probe = Some(n);
422        self
423    }
424    /// Override the number of candidates reranked exactly (accuracy dial).
425    pub fn n_rerank(mut self, n: usize) -> Self {
426        self.n_rerank = Some(n);
427        self
428    }
429    /// Toggle whether recall updates `accessed_at` / `access_count` on
430    /// the returned memories' catalog entries (default `true`). Pass
431    /// `false` for a fully read-only recall — see the field doc.
432    pub fn track_access(mut self, track: bool) -> Self {
433        self.track_access = track;
434        self
435    }
436}
437
438/// One scored result from [`Mnemo::recall`].
439#[derive(Clone, Debug)]
440pub struct RecallResult {
441    /// The retrieved memory.
442    pub memory: Memory,
443    /// Combined multi-signal score.
444    pub score: f32,
445    /// The bare similarity component (before other signals).
446    pub similarity: f32,
447}
448
449/// Summary statistics for a database.
450#[derive(Clone, Debug)]
451pub struct Stats {
452    /// Live (non-deleted) memory count.
453    pub memories: usize,
454    /// Tombstoned entries awaiting compaction.
455    pub deleted: usize,
456    /// Embedding dimensionality.
457    pub dimensions: usize,
458    /// File size in bytes.
459    pub file_bytes: u64,
460    /// Distinct agent IDs present.
461    pub agents: Vec<String>,
462    /// Whether pages are encrypted (always true in v1).
463    pub encrypted: bool,
464    /// Creation time (unix seconds).
465    pub created_at: i64,
466    /// ANN index shape, if an index has been built.
467    pub index: Option<IndexInfo>,
468    /// Current size of the write-ahead log region, in 8 KiB pages.
469    pub wal_pages: u64,
470    /// Fsyncs issued by the most recent successful [`Mnemo::flush`] on
471    /// this handle. `0` on a fresh handle before any flush, or when
472    /// `flush()` short-circuited (nothing dirty). Phase 6 instrumentation
473    /// for the Phase 11.1 write-latency work — the planner wants this
474    /// before deciding whether cutting fsync count is a viable
475    /// durability-preserving fix.
476    pub fsyncs_last_flush: u32,
477}
478
479/// Result of a [`Mnemo::compact_file`] run.
480#[derive(Clone, Copy, Debug, Default)]
481pub struct CompactReport {
482    /// Live memories before compaction.
483    pub before: usize,
484    /// Live memories after compaction (expired ones dropped).
485    pub after: usize,
486    /// Live, non-expired cache entries before compaction. FIX-1
487    /// R-001.1: compaction now preserves the result cache instead of
488    /// silently dropping it.
489    pub cache_before: usize,
490    /// Live, non-expired cache entries after compaction (tombstoned
491    /// and TTL-expired ones dropped).
492    pub cache_after: usize,
493}
494
495/// An encrypted, single-file agent memory database.
496pub struct Mnemo {
497    pager: Pager,
498    header: Header,
499    catalog: Vec<CatalogEntry>,
500    index: HashMap<u128, usize>,
501    #[allow(dead_code)]
502    path: PathBuf,
503    dimensions: usize,
504    kdf: KdfParams,
505    /// Set whenever the catalog changes; drives whether `flush` rewrites it.
506    dirty_catalog: bool,
507    /// Optional IVF+PQ approximate-nearest-neighbour index.
508    ann: Option<IvfPqIndex>,
509    /// Set whenever the ANN index changes; drives whether `flush` rewrites it.
510    dirty_index: bool,
511    /// Append-only manifest of committed snapshots, oldest first. Capped
512    /// at [`Mnemo::max_snapshots`] entries on every flush — the oldest
513    /// entries are pruned first.
514    manifest: Vec<Snapshot>,
515    /// Maximum manifest entries to retain across flushes. `0` disables
516    /// the cap (retain forever). Defaults to [`DEFAULT_MAX_SNAPSHOTS`].
517    max_snapshots: usize,
518    /// If `true`, this handle was opened via [`Mnemo::open_read_only`] and
519    /// every mutating API returns [`MnemoError::ReadOnly`]. The underlying
520    /// file was opened without write permission and holds an OS shared
521    /// lock; the OS would reject a write() anyway, but gating in Rust
522    /// gives callers a clean, typed error.
523    read_only: bool,
524    /// Result-cache directory (Phase 10.1). See [`crate::result_cache`].
525    /// Loaded from `header.cache_start` on open; persisted on `flush`
526    /// via the same seal_run pipeline as catalog/index/manifest.
527    cache_dir: Vec<CacheDirectoryEntry>,
528    /// In-memory index over `cache_dir` for O(1) `(namespace, key_hash)`
529    /// lookups. Rebuilt on open + kept in sync on `cache_put`/`delete`.
530    cache_idx: DirectoryIndex,
531    /// Set whenever `cache_dir` changes; drives whether `flush` rewrites
532    /// the cache directory run.
533    dirty_cache: bool,
534    /// Per-namespace eviction budgets. Missing entries fall back to
535    /// [`CacheBudget::default`]. Not persisted (per-handle policy).
536    cache_budgets: std::collections::HashMap<String, CacheBudget>,
537    /// Flush policy for cache mutations. Memory writes ignore this and
538    /// always behave as `Strict`.
539    cache_flush_policy: CacheFlushPolicy,
540    /// Bookkeeping for `CacheFlushPolicy::Batched` — pending mutation
541    /// count and first-dirty timestamp.
542    cache_batch: BatchState,
543    /// In-memory hit/miss/eviction counters (reset on reopen).
544    cache_counters: CounterState,
545    /// Optional handle-scoped embedder (Phase 3, PR 10a). Registered
546    /// via [`Mnemo::set_embedder`]; consulted by [`Mnemo::remember_text`]
547    /// and [`RecallRequest::from_text`]. `None` means text-side APIs
548    /// return [`MnemoError::Invalid`] with a "no embedder" message.
549    /// Not persisted — the handle is re-registered after every open.
550    embedder: Option<std::sync::Arc<dyn Embedder>>,
551    /// Fsyncs issued by the most recent `flush()` call. Snapshotted
552    /// from the thread-local fsync counter in `lib.rs::fsync_counter`:
553    /// `flush()` reads it on entry, does its work (each `sync_all` the
554    /// pager / wal / recovery path makes bumps the counter), and
555    /// stores the delta here. `0` on a fresh handle before any flush,
556    /// or on a read-only handle where `flush()` short-circuits.
557    /// Exposed through [`Stats::fsyncs_last_flush`] for diagnostics —
558    /// the Phase 6 planner wants this before deciding whether cutting
559    /// fsync count is a viable write-latency win.
560    fsyncs_last_flush: u32,
561    /// FIX-1 R-001.8 part 2: lazy cache of the distinct agent IDs
562    /// across every live memory, built once on first `stats()` call
563    /// per handle. `remember()` adds to it in O(1); `delete()` can't
564    /// cheaply tell whether the removed memory was the last of its
565    /// agent (the agent lives in the encrypted body, not the catalog
566    /// — pre-v9 format), so it just invalidates the set and the next
567    /// `stats()` call rebuilds. Pre-FIX-1 every `stats()` call
568    /// decrypted every live memory to collect agents, turning an
569    /// otherwise-cheap diagnostic into an O(N) decrypt.
570    agent_set: Option<std::collections::BTreeSet<String>>,
571}
572
573/// Open a file and take an OS advisory lock, returning `Locked { path }`
574/// if the requested lock cannot be acquired. On drop of the returned
575/// [`File`], the OS releases the lock. `flock`/`LockFileEx` semantics
576/// (per file-descriptor on Unix, per-open-file-handle on Windows) mean
577/// the lock is scoped to this handle: cloning the file (or the returned
578/// `File`) shares the lock, but a fresh `open()` from any process gets a
579/// distinct handle and will collide.
580fn open_with_lock(path: &Path, write: bool, create: bool) -> Result<File> {
581    let mut opts = OpenOptions::new();
582    opts.read(true);
583    if write {
584        opts.write(true);
585    }
586    if create {
587        opts.create(true).truncate(true);
588    }
589    let file = opts.open(path)?;
590    // On success the returned `Result<()>` is `Ok(())`. On conflict:
591    //   * Unix `flock(2)` → `ErrorKind::WouldBlock` (EWOULDBLOCK / EAGAIN)
592    //   * Windows `LockFileEx` → raw OS code 33 (ERROR_LOCK_VIOLATION)
593    //     or 32 (ERROR_SHARING_VIOLATION), both surfaced by Rust as
594    //     `ErrorKind::Uncategorized` because `std::io::ErrorKind`
595    //     doesn't carry a portable "locked" variant.
596    // Anything else is a real I/O failure (permissions, invalid handle,
597    // ...) and gets bubbled up as `MnemoError::Io`.
598    let result = if write {
599        FileExt::try_lock_exclusive(&file)
600    } else {
601        FileExt::try_lock_shared(&file)
602    };
603    match result {
604        Ok(()) => Ok(file),
605        Err(e) if is_lock_conflict(&e) => Err(MnemoError::Locked {
606            path: path.to_path_buf(),
607        }),
608        Err(e) => Err(MnemoError::Io(e)),
609    }
610}
611
612/// Do two paths resolve to the same on-disk file? Used by
613/// [`Mnemo::export_encrypted`] to refuse overwriting the live
614/// database. On Unix we compare `(dev, ino)` via `MetadataExt`, which
615/// catches hard links and symlinks. On Windows we compare canonicalized
616/// paths, since there is no stable device+inode equivalent in `std`.
617/// Returns `None` if either side can't be stat'd / canonicalized —
618/// the caller treats that as "not the same file" so a missing `dest`
619/// (the normal export case) proceeds.
620fn paths_refer_to_same_file(a: &Path, b: &Path) -> Option<bool> {
621    #[cfg(unix)]
622    {
623        use std::os::unix::fs::MetadataExt;
624        let ma = std::fs::metadata(a).ok()?;
625        let mb = std::fs::metadata(b).ok()?;
626        Some(ma.dev() == mb.dev() && ma.ino() == mb.ino())
627    }
628    #[cfg(not(unix))]
629    {
630        let ca = std::fs::canonicalize(a).ok()?;
631        let cb = std::fs::canonicalize(b).ok()?;
632        Some(ca == cb)
633    }
634}
635
636/// A sibling temp-file path next to `dest`, suffixed with
637/// `.export-tmp`. Lives in `dest`'s own parent directory so the final
638/// [`std::fs::rename`] is same-filesystem (and therefore atomic on
639/// every platform mnemo supports).
640fn temp_sibling(dest: &Path) -> PathBuf {
641    let mut buf = dest.as_os_str().to_os_string();
642    buf.push(".export-tmp");
643    PathBuf::from(buf)
644}
645
646/// Was this I/O error the OS reporting a conflicting file lock, or a
647/// genuine I/O failure? Platform-conditional because
648/// [`std::io::ErrorKind`] carries no portable "resource-locked" variant
649/// as of Rust 1.75 — Windows surfaces `ERROR_LOCK_VIOLATION` /
650/// `ERROR_SHARING_VIOLATION` as `Uncategorized`.
651fn is_lock_conflict(e: &std::io::Error) -> bool {
652    if e.kind() == std::io::ErrorKind::WouldBlock {
653        return true;
654    }
655    #[cfg(windows)]
656    if let Some(code) = e.raw_os_error() {
657        // 33 = ERROR_LOCK_VIOLATION, 32 = ERROR_SHARING_VIOLATION —
658        // both mean "someone else holds an incompatible lock on this
659        // region of the file". See
660        // https://learn.microsoft.com/en-us/windows/win32/debug/system-error-codes--0-499-
661        if code == 33 || code == 32 {
662            return true;
663        }
664    }
665    false
666}
667
668/// Read a run of consecutive encrypted pages and concatenate their plaintext.
669fn read_run_bytes(pager: &mut Pager, start: u64, pages: u64, len: u64) -> Result<Vec<u8>> {
670    let mut buf = Vec::with_capacity(len as usize);
671    for i in 0..pages {
672        buf.extend_from_slice(&pager.read_page(start + i)?);
673    }
674    buf.truncate(len as usize);
675    Ok(buf)
676}
677
678/// Load a v5+ catalog run (an empty `pages` yields an empty catalog).
679fn load_catalog(pager: &mut Pager, start: u64, pages: u64, len: u64) -> Result<Vec<CatalogEntry>> {
680    if pages == 0 {
681        return Ok(Vec::new());
682    }
683    let buf = read_run_bytes(pager, start, pages, len)?;
684    rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
685}
686
687/// Load a **v4** catalog run using the frozen [`CatalogEntryV4`] shape.
688/// Used only by the v4→v5 migration path in [`Mnemo::open`].
689fn load_catalog_v4(
690    pager: &mut Pager,
691    start: u64,
692    pages: u64,
693    len: u64,
694) -> Result<Vec<CatalogEntryV4>> {
695    if pages == 0 {
696        return Ok(Vec::new());
697    }
698    let buf = read_run_bytes(pager, start, pages, len)?;
699    rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
700}
701
702/// Walk a v4 catalog and synthesize the v5 shape. For each live entry,
703/// reads the memory's serialized body and copies its `accessed_at` and
704/// `access_count` into the catalog row (the v4 record body still carries
705/// them — they're moving *from* the body *to* the catalog). Deleted
706/// entries skip the read and zero the fields.
707fn migrate_v4_catalog(pager: &mut Pager, v4: Vec<CatalogEntryV4>) -> Result<Vec<CatalogEntry>> {
708    let mut out = Vec::with_capacity(v4.len());
709    for e in &v4 {
710        let (accessed_at, access_count) = if e.deleted {
711            (0i64, 0u32)
712        } else {
713            let mut buf = Vec::with_capacity(e.len as usize);
714            for i in 0..e.page_count as u64 {
715                buf.extend_from_slice(&pager.read_page(e.start_page + i)?);
716            }
717            buf.truncate(e.len as usize);
718            let mem: Memory = rmp_serde::from_slice(&buf)
719                .map_err(|err| MnemoError::Serialize(err.to_string()))?;
720            (mem.accessed_at, mem.access_count)
721        };
722        out.push(CatalogEntry {
723            id: e.id,
724            start_page: e.start_page,
725            page_count: e.page_count,
726            len: e.len,
727            deleted: e.deleted,
728            accessed_at,
729            access_count,
730        });
731    }
732    Ok(out)
733}
734
735/// Load the snapshot manifest (an empty `pages` yields an empty manifest).
736fn load_manifest(pager: &mut Pager, start: u64, pages: u64, len: u64) -> Result<Vec<Snapshot>> {
737    if pages == 0 {
738        return Ok(Vec::new());
739    }
740    let buf = read_run_bytes(pager, start, pages, len)?;
741    rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
742}
743
744/// Load the result-cache directory (v8+). An empty `pages` yields an
745/// empty directory, which is the natural state for v7 files migrated
746/// to v8 (no cache entries yet) and for freshly created v8 files.
747fn load_cache_directory(
748    pager: &mut Pager,
749    start: u64,
750    pages: u64,
751    len: u64,
752) -> Result<Vec<CacheDirectoryEntry>> {
753    if pages == 0 {
754        return Ok(Vec::new());
755    }
756    let buf = read_run_bytes(pager, start, pages, len)?;
757    rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
758}
759
760/// Load an ANN index run, validating its dimensionality.
761fn load_index(
762    pager: &mut Pager,
763    start: u64,
764    pages: u64,
765    len: u64,
766    dims: usize,
767) -> Result<Option<IvfPqIndex>> {
768    if pages == 0 {
769        return Ok(None);
770    }
771    let buf = read_run_bytes(pager, start, pages, len)?;
772    let mut idx: IvfPqIndex =
773        rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
774    if idx.dims() != dims {
775        return Err(MnemoError::Invalid(
776            "ANN index dimensionality does not match the database".into(),
777        ));
778    }
779    idx.rebuild_assignment();
780    Ok(Some(idx))
781}
782
783/// True if a memory's metadata marks it as the canonical onboarding manifest
784/// (`metadata.topic == "manifest"`). Used by [`Mnemo::about`] to hoist the
785/// manifest to the top of the briefing regardless of importance ordering.
786fn is_manifest(m: &Memory) -> bool {
787    m.metadata
788        .get("topic")
789        .and_then(|v| v.as_str())
790        .map(|s| s.eq_ignore_ascii_case("manifest"))
791        .unwrap_or(false)
792}
793
794/// Build the ULID → catalog-slot lookup map.
795fn build_id_index(catalog: &[CatalogEntry]) -> HashMap<u128, usize> {
796    let mut m = HashMap::with_capacity(catalog.len());
797    for (i, e) in catalog.iter().enumerate() {
798        m.insert(e.id, i);
799    }
800    m
801}
802
803impl Mnemo {
804    /// Create a brand-new encrypted database at `path`.
805    pub fn create(
806        path: impl Into<PathBuf>,
807        passphrase: &str,
808        config: MnemoConfig,
809    ) -> Result<Mnemo> {
810        let path: PathBuf = path.into();
811        if config.dimensions == 0 {
812            return Err(MnemoError::Invalid("dimensions must be > 0".into()));
813        }
814        // Exclusive lock covers the write path; second writer sees `Locked`.
815        let file = open_with_lock(&path, /*write=*/ true, /*create=*/ true)?;
816
817        let salt = crypto::random_salt();
818        let dek = crypto::random_dek();
819        let kek = crypto::derive_kek(passphrase.as_bytes(), &salt, config.kdf)?;
820        let dek_nonce = crypto::random_nonce();
821        let wrapped = crypto::wrap_dek(&kek, &dek_nonce, &dek)?;
822        if wrapped.len() != WRAPPED_DEK_LEN {
823            return Err(MnemoError::Crypto("unexpected wrapped DEK length".into()));
824        }
825        let mut wrapped_dek = [0u8; WRAPPED_DEK_LEN];
826        wrapped_dek.copy_from_slice(&wrapped);
827
828        let mut header = Header {
829            version: VERSION,
830            page_size: PAGE_SIZE as u32,
831            flags: FLAG_ENCRYPTED,
832            dimensions: config.dimensions as u32,
833            created_at: memory::now_secs(),
834            write_counter: 0,
835            // Page 0 is the header; pages 1..=wal_pages are the WAL;
836            // record/catalog/index pages start after it. Initial WAL size is
837            // chosen per-config (default 8 pages = 64 KiB) and clamped to a
838            // sane floor so the first transaction can always commit.
839            next_page: 1 + config.wal_pages_initial.max(MIN_WAL_PAGES),
840            catalog_start: 0,
841            catalog_pages: 0,
842            catalog_len: 0,
843            vector_count: 0,
844            m_cost: config.kdf.m_cost,
845            t_cost: config.kdf.t_cost,
846            p_cost: config.kdf.p_cost,
847            salt,
848            dek_nonce,
849            wrapped_dek,
850            index_start: 0,
851            index_pages: 0,
852            index_len: 0,
853            wal_start: 1,
854            wal_pages: config.wal_pages_initial.max(MIN_WAL_PAGES),
855            wal_seq: 0,
856            manifest_start: 0,
857            manifest_pages: 0,
858            manifest_len: 0,
859            // Regenerated per write by `apply_seal`; `None` until we seal.
860            seal_nonce: None,
861            seal_tag: None,
862            // v8+ result-cache pointer — zero-initialized on create,
863            // populated only when `cache_put` runs at least once and
864            // the next `flush` seals a directory run.
865            cache_start: 0,
866            cache_pages: 0,
867            cache_len: 0,
868        };
869
870        let mut pager = Pager::new(file, dek, 0, VERSION);
871        let mut page = header.to_page();
872        header.apply_seal(&mut page, pager.dek())?;
873        pager.write_raw(0, &page)?;
874        pager.sync()?;
875
876        Ok(Mnemo {
877            pager,
878            header,
879            catalog: Vec::new(),
880            index: HashMap::new(),
881            path,
882            dimensions: config.dimensions,
883            kdf: config.kdf,
884            dirty_catalog: false,
885            ann: None,
886            dirty_index: false,
887            manifest: Vec::new(),
888            max_snapshots: config.max_snapshots,
889            read_only: false,
890            cache_dir: Vec::new(),
891            cache_idx: DirectoryIndex::new(),
892            dirty_cache: false,
893            cache_budgets: std::collections::HashMap::new(),
894            cache_flush_policy: CacheFlushPolicy::default(),
895            cache_batch: BatchState::default(),
896            cache_counters: CounterState::default(),
897            embedder: None,
898            fsyncs_last_flush: 0,
899            agent_set: None,
900        })
901    }
902
903    /// Open an existing database. A wrong passphrase fails cleanly with
904    /// [`MnemoError::WrongPassphrase`]. Takes an OS exclusive lock for
905    /// the lifetime of the returned handle; a second concurrent open
906    /// (from any process) returns [`MnemoError::Locked`].
907    ///
908    /// For a caller that only needs to read (and can tolerate the
909    /// pending-recovery / pending-migration cases documented on
910    /// [`Mnemo::open_read_only`]), prefer that entry point — it takes a
911    /// shared lock and coexists with other readers.
912    pub fn open(path: impl Into<PathBuf>, passphrase: &str) -> Result<Mnemo> {
913        let path: PathBuf = path.into();
914        let mut file = open_with_lock(&path, /*write=*/ true, /*create=*/ false)?;
915
916        let mut hbuf = [0u8; PAGE_SIZE];
917        file.seek(SeekFrom::Start(0))?;
918        file.read_exact(&mut hbuf)?;
919        let mut header = match Header::from_page(&hbuf) {
920            Ok(h) => h,
921            Err(e) => {
922                // DUR-1 (R-002.1 part 2+3): page 0 did not parse — any
923                // error is treated as a torn-header candidate, not just
924                // `HeaderChecksum`. The whole 294-byte header lives in
925                // the first 512-byte sector, so the realistic failure is
926                // the sector zeroed or garbled, which returns `BadMagic`
927                // or `UnsupportedVersion` *before* the CRC gate.
928                //
929                // Scan the file for valid committed header-frame WAL
930                // regions, pick the highest `txn_id`, and replay it. If
931                // nothing valid is found, return the *original* error so
932                // non-mnemo files still report `BadMagic`.
933                let scan = scan_header_recovery_candidates(&mut file, passphrase)?;
934                let frames = scan.ok_or(e)?;
935                for (page_no, bytes) in &frames {
936                    if bytes.len() != PAGE_SIZE {
937                        return Err(MnemoError::Invalid("WAL frame is not page-sized".into()));
938                    }
939                    file.seek(SeekFrom::Start(page_no * PAGE_SIZE as u64))?;
940                    file.write_all(bytes)?;
941                }
942                file.sync_all()?;
943                crate::bump_fsync();
944                file.seek(SeekFrom::Start(0))?;
945                file.read_exact(&mut hbuf)?;
946                Header::from_page(&hbuf)?
947            }
948        };
949
950        // Crash recovery: replay a committed-but-uncheckpointed transaction.
951        // Each frame is a finished page image; the header frame, replayed to
952        // page 0, supersedes the header just read.
953        if let Some(frames) = wal::recover(
954            &mut file,
955            header.wal_start,
956            header.wal_pages,
957            header.wal_seq,
958        )? {
959            for (page_no, bytes) in &frames {
960                if bytes.len() != PAGE_SIZE {
961                    return Err(MnemoError::Invalid("WAL frame is not page-sized".into()));
962                }
963                file.seek(SeekFrom::Start(page_no * PAGE_SIZE as u64))?;
964                file.write_all(bytes)?;
965            }
966            file.sync_all()?;
967            crate::bump_fsync();
968            // Re-read the now-current header from the replayed page 0.
969            file.seek(SeekFrom::Start(0))?;
970            file.read_exact(&mut hbuf)?;
971            header = Header::from_page(&hbuf)?;
972        }
973
974        let kdf = KdfParams {
975            m_cost: header.m_cost,
976            t_cost: header.t_cost,
977            p_cost: header.p_cost,
978        };
979        let kek = crypto::derive_kek(passphrase.as_bytes(), &header.salt, kdf)?;
980        let dek = crypto::unwrap_dek(&kek, &header.dek_nonce, &header.wrapped_dek)?;
981
982        // Validate the v7 header seal now that we have the DEK. Returns
983        // `MnemoError::HeaderTampered` if any of the sealed mutable fields
984        // were rewritten after the last legitimate flush. No-op for v6 and
985        // older — pre-v7 files don't carry a seal.
986        //
987        // DUR-1 (Planner decision 2026-11-01, Option B): a seal failure is
988        // **always** reported as tampering, never auto-healed. R-002.1
989        // part 3's seal-mismatch WAL fallback is withdrawn: the seal
990        // exists for tamper-evidence; silently loading a stale committed
991        // header from the WAL would erase that evidence. A real crash
992        // cannot produce "CRC valid + seal invalid" either, because the
993        // whole 294-byte header lies inside sector 0 which storage
994        // writes atomically — a torn sector breaks magic, version or
995        // CRC first, and *those* are healed by the scan above.
996        header.validate_seal(&dek)?;
997
998        // Build the pager at the file's *current* version so initial reads
999        // use the matching AAD scheme — `Pager::page_aad` returns empty AAD
1000        // for v4/v5 (no page binding) and `page_no.to_le_bytes()` for v6+.
1001        let on_disk_version = header.version;
1002        let mut pager = Pager::new(file, dek, header.write_counter, on_disk_version);
1003        let dimensions = header.dimensions as usize;
1004
1005        // === v4→v5 catalog migration =====================================
1006        // v6 reads directly; v5 reads directly; v4 replays the catalog into
1007        // the v5 shape by reading each memory's body to populate the new
1008        // `accessed_at` / `access_count` fields. All page reads here still
1009        // use the v4/v5 AAD scheme (none).
1010        let (catalog, migrated_from_v4) = if on_disk_version >= 5 {
1011            (
1012                load_catalog(
1013                    &mut pager,
1014                    header.catalog_start,
1015                    header.catalog_pages,
1016                    header.catalog_len,
1017                )?,
1018                false,
1019            )
1020        } else if on_disk_version == 4 {
1021            let v4 = load_catalog_v4(
1022                &mut pager,
1023                header.catalog_start,
1024                header.catalog_pages,
1025                header.catalog_len,
1026            )?;
1027            (migrate_v4_catalog(&mut pager, v4)?, true)
1028        } else {
1029            return Err(MnemoError::UnsupportedVersion(on_disk_version));
1030        };
1031        let index = build_id_index(&catalog);
1032        let ann = load_index(
1033            &mut pager,
1034            header.index_start,
1035            header.index_pages,
1036            header.index_len,
1037            dimensions,
1038        )?;
1039        let mut manifest = load_manifest(
1040            &mut pager,
1041            header.manifest_start,
1042            header.manifest_pages,
1043            header.manifest_len,
1044        )?;
1045
1046        // === v5→v6 page-encryption migration =============================
1047        // Pre-v6 data pages were encrypted under no AAD; v6 binds the home
1048        // `page_no` as AAD so a page-transplant attack fails authentication
1049        // on read. Re-stage every live record page through `write_page` so
1050        // the next flush re-encrypts it under v6 AAD at the same home slot.
1051        // Tombstoned entries skip — they're reclaimed by `compact_file`.
1052        // Old catalog / ANN / manifest pages stay as orphan ciphertext;
1053        // the next flush writes the new runs at fresh slots.
1054        let migrated_pages_to_v6 = on_disk_version < 6;
1055        if migrated_pages_to_v6 {
1056            for e in &catalog {
1057                if e.deleted {
1058                    continue;
1059                }
1060                for i in 0..e.page_count as u64 {
1061                    let payload = pager.read_page(e.start_page + i)?;
1062                    pager.write_page(e.start_page + i, &payload)?;
1063                }
1064            }
1065            pager.set_version(6);
1066        }
1067
1068        // === v6→v7 header-seal migration =================================
1069        // v7 adds an AES-GCM seal at the tail of the header page that
1070        // authenticates every mutable header field. Pre-v7 files have no
1071        // seal — we just bump the version in memory and mark `dirty_catalog`
1072        // so the next flush writes a sealed v7 header. Pages are untouched
1073        // (page crypto is unchanged from v6).
1074        let migrated_header_to_v7 = on_disk_version < 7;
1075        if migrated_header_to_v7 {
1076            pager.set_version(7);
1077        }
1078
1079        // === v7→v8 cache-directory migration =============================
1080        // v8 adds three u64 header fields (cache_start/pages/len) for the
1081        // result-cache directory (Phase 10.1). Pre-v8 files have those
1082        // three bytes as zero (the on-disk region beyond the v7 seal is
1083        // zero-initialized), which is the legal "empty cache directory"
1084        // state — no data migration required. We bump the version in
1085        // memory and mark dirty_catalog so the next flush writes a
1086        // v8-sealed header with the new AAD shape covering the cache
1087        // pointer. Pages are untouched.
1088        let migrated_header_to_v8 = on_disk_version < 8;
1089        if migrated_header_to_v8 {
1090            pager.set_version(VERSION);
1091        }
1092
1093        let migrated = migrated_from_v4
1094            || migrated_pages_to_v6
1095            || migrated_header_to_v7
1096            || migrated_header_to_v8;
1097        if migrated {
1098            // Stamp the version forward in memory; the next flush persists
1099            // it via the WAL-committed header frame (and the v7 seal).
1100            header.version = VERSION;
1101            // Snapshots written under an older format reference page runs
1102            // encrypted under the old AAD scheme; this build can't decrypt
1103            // them under the new pager. Drop the manifest so the next
1104            // flush records a fresh snapshot of the migrated state.
1105            // Point-in-time recovery into the pre-migration past is
1106            // sacrificed for migration simplicity. Old pages remain on
1107            // disk until `compact_file`. (For v6→v7 alone — which doesn't
1108            // change page crypto — the old snapshots would technically
1109            // still be readable, but dropping them uniformly keeps the
1110            // migration policy simple.)
1111            manifest.clear();
1112        }
1113
1114        // The ANN index pages on disk were sealed under the *old* AAD scheme
1115        // if we migrated pages to v6; force a rewrite at fresh v6-encrypted
1116        // slots on the next flush so a future reopen doesn't try to read
1117        // them under the new AAD and fail. (Catalog and manifest already
1118        // get rewritten via dirty_catalog and the manifest.clear() above.)
1119        let dirty_index = migrated_pages_to_v6 && ann.is_some();
1120
1121        // Load the cache directory (v8+ only). Pre-v8 files have
1122        // cache_pages == 0 by header layout, so this trivially yields
1123        // an empty directory without touching the disk.
1124        let cache_dir = load_cache_directory(
1125            &mut pager,
1126            header.cache_start,
1127            header.cache_pages,
1128            header.cache_len,
1129        )?;
1130        let cache_idx = DirectoryIndex::build(&cache_dir);
1131
1132        Ok(Mnemo {
1133            pager,
1134            header,
1135            catalog,
1136            index,
1137            path,
1138            dimensions,
1139            kdf,
1140            dirty_catalog: migrated,
1141            ann,
1142            dirty_index,
1143            manifest,
1144            // `Mnemo::open` has no [`MnemoConfig`] to read; default the cap
1145            // to [`DEFAULT_MAX_SNAPSHOTS`]. Override with
1146            // [`Mnemo::set_max_snapshots`] right after open if you need
1147            // unlimited retention or a different cap for this handle.
1148            max_snapshots: DEFAULT_MAX_SNAPSHOTS,
1149            read_only: false,
1150            cache_dir,
1151            cache_idx,
1152            dirty_cache: false,
1153            cache_budgets: std::collections::HashMap::new(),
1154            cache_flush_policy: CacheFlushPolicy::default(),
1155            cache_batch: BatchState::default(),
1156            cache_counters: CounterState::default(),
1157            embedder: None,
1158            fsyncs_last_flush: 0,
1159            agent_set: None,
1160        })
1161    }
1162
1163    /// Open an existing database in **read-only** mode. Takes an OS shared
1164    /// lock; multiple read-only handles can coexist, but any attempt to
1165    /// open the same file read-write elsewhere returns
1166    /// [`MnemoError::Locked`]. All mutating methods on the returned handle
1167    /// return [`MnemoError::ReadOnly`], including `recall` with
1168    /// `track_access(true)`.
1169    ///
1170    /// Two conditions block a read-only open and return
1171    /// [`MnemoError::NeedsWriteOpen`]:
1172    ///
1173    /// 1. The file has a committed-but-uncheckpointed WAL transaction that
1174    ///    would ordinarily be replayed on open. A read-only handle
1175    ///    cannot write the recovered pages back.
1176    /// 2. The on-disk format version is older than the current
1177    ///    `format::VERSION`, requiring an on-disk migration.
1178    ///
1179    /// In either case, open the file read-write once (which recovers /
1180    /// migrates in place), close the handle cleanly, and reopen
1181    /// read-only. This is documented in the README's Durability section.
1182    pub fn open_read_only(path: impl Into<PathBuf>, passphrase: &str) -> Result<Mnemo> {
1183        let path: PathBuf = path.into();
1184        let mut file = open_with_lock(&path, /*write=*/ false, /*create=*/ false)?;
1185
1186        let mut hbuf = [0u8; PAGE_SIZE];
1187        file.seek(SeekFrom::Start(0))?;
1188        file.read_exact(&mut hbuf)?;
1189        // A torn header on a read-only open cannot be healed (that requires
1190        // writing recovered frames back). Surface as `NeedsWriteOpen` so the
1191        // caller knows to reopen read-write once. DUR-1 (R-002.1 part 4):
1192        // any `Header::from_page` failure (`BadMagic`,
1193        // `UnsupportedVersion`, `HeaderChecksum`) qualifies — but only
1194        // when the file-wide scan actually finds a valid committed
1195        // header frame to recover from, so a non-mnemo file still
1196        // reports `BadMagic` cleanly. A `validate_seal` failure is NOT
1197        // routed here (Option B: tampering is never auto-healed).
1198        let header = match Header::from_page(&hbuf) {
1199            Ok(h) => h,
1200            Err(parse_err) => {
1201                let recoverable = scan_header_recovery_candidates(&mut file, passphrase)?;
1202                return Err(match (parse_err, recoverable) {
1203                    (_, Some(_)) => MnemoError::NeedsWriteOpen {
1204                        reason: "torn header page needs recovery from WAL",
1205                    },
1206                    (e, None) => e,
1207                });
1208            }
1209        };
1210
1211        // Refuse to migrate on read-only opens. v6→v7 is header-only
1212        // (no page writes required), but even that bumps `version` on
1213        // disk and re-seals the header — which needs write access. Keep
1214        // the policy uniform across all sub-version bumps.
1215        if header.version < VERSION {
1216            return Err(MnemoError::NeedsWriteOpen {
1217                reason: "pending format-version migration",
1218            });
1219        }
1220
1221        // Refuse if there is a committed-but-uncheckpointed transaction
1222        // waiting for `wal::recover` to fold it into home pages.
1223        if wal::recover(
1224            &mut file,
1225            header.wal_start,
1226            header.wal_pages,
1227            header.wal_seq,
1228        )?
1229        .is_some()
1230        {
1231            return Err(MnemoError::NeedsWriteOpen {
1232                reason: "pending WAL recovery",
1233            });
1234        }
1235
1236        let kdf = KdfParams {
1237            m_cost: header.m_cost,
1238            t_cost: header.t_cost,
1239            p_cost: header.p_cost,
1240        };
1241        let kek = crypto::derive_kek(passphrase.as_bytes(), &header.salt, kdf)?;
1242        let dek = crypto::unwrap_dek(&kek, &header.dek_nonce, &header.wrapped_dek)?;
1243        header.validate_seal(&dek)?;
1244
1245        let mut pager = Pager::new(file, dek, header.write_counter, header.version);
1246        let dimensions = header.dimensions as usize;
1247
1248        let catalog = load_catalog(
1249            &mut pager,
1250            header.catalog_start,
1251            header.catalog_pages,
1252            header.catalog_len,
1253        )?;
1254        let index = build_id_index(&catalog);
1255        let ann = load_index(
1256            &mut pager,
1257            header.index_start,
1258            header.index_pages,
1259            header.index_len,
1260            dimensions,
1261        )?;
1262        let manifest = load_manifest(
1263            &mut pager,
1264            header.manifest_start,
1265            header.manifest_pages,
1266            header.manifest_len,
1267        )?;
1268        let cache_dir = load_cache_directory(
1269            &mut pager,
1270            header.cache_start,
1271            header.cache_pages,
1272            header.cache_len,
1273        )?;
1274        let cache_idx = DirectoryIndex::build(&cache_dir);
1275
1276        Ok(Mnemo {
1277            pager,
1278            header,
1279            catalog,
1280            index,
1281            path,
1282            dimensions,
1283            kdf,
1284            // No mutation is possible; leave both dirty flags false so
1285            // `close()` is a genuine no-op even if we later add an
1286            // auto-flush on drop.
1287            dirty_catalog: false,
1288            ann,
1289            dirty_index: false,
1290            manifest,
1291            max_snapshots: DEFAULT_MAX_SNAPSHOTS,
1292            read_only: true,
1293            cache_dir,
1294            cache_idx,
1295            dirty_cache: false,
1296            cache_budgets: std::collections::HashMap::new(),
1297            cache_flush_policy: CacheFlushPolicy::default(),
1298            cache_batch: BatchState::default(),
1299            cache_counters: CounterState::default(),
1300            embedder: None,
1301            fsyncs_last_flush: 0,
1302            agent_set: None,
1303        })
1304    }
1305
1306    /// Override the manifest snapshot cap on this open handle. `0` disables
1307    /// the cap; any positive value keeps the most-recent `max` snapshots and
1308    /// drops the rest on the next flush. See [`MnemoConfig::max_snapshots`].
1309    pub fn set_max_snapshots(&mut self, max: usize) {
1310        self.max_snapshots = max;
1311    }
1312
1313    /// Embedding dimensionality this database expects.
1314    pub fn dimensions(&self) -> usize {
1315        self.dimensions
1316    }
1317
1318    /// Number of live (non-deleted) memories.
1319    pub fn len(&self) -> usize {
1320        self.catalog.iter().filter(|e| !e.deleted).count()
1321    }
1322
1323    /// True if the database holds no live memories.
1324    pub fn is_empty(&self) -> bool {
1325        self.len() == 0
1326    }
1327
1328    // --- internal page/record helpers -------------------------------------
1329
1330    fn write_record(&mut self, bytes: &[u8]) -> Result<(u64, u32)> {
1331        let pc = bytes.len().div_ceil(PAYLOAD).max(1);
1332        let start = self.header.next_page;
1333        self.header.next_page += pc as u64;
1334        for i in 0..pc {
1335            let lo = i * PAYLOAD;
1336            let hi = ((i + 1) * PAYLOAD).min(bytes.len());
1337            self.pager.write_page(start + i as u64, &bytes[lo..hi])?;
1338        }
1339        Ok((start, pc as u32))
1340    }
1341
1342    fn read_record(&mut self, e: &CatalogEntry) -> Result<Vec<u8>> {
1343        let mut buf = Vec::with_capacity(e.len as usize);
1344        for i in 0..e.page_count as u64 {
1345            buf.extend_from_slice(&self.pager.read_page(e.start_page + i)?);
1346        }
1347        buf.truncate(e.len as usize);
1348        Ok(buf)
1349    }
1350
1351    fn read_memory(&mut self, e: &CatalogEntry) -> Result<Memory> {
1352        let bytes = self.read_record(e)?;
1353        let mut m: Memory =
1354            rmp_serde::from_slice(&bytes).map_err(|err| MnemoError::Serialize(err.to_string()))?;
1355        // The catalog is the source of truth for access stats — the values
1356        // in the record body are a stale snapshot from when the record was
1357        // last written. Overwrite them here so consumers always see live
1358        // values regardless of when the record was last serialized.
1359        m.accessed_at = e.accessed_at;
1360        m.access_count = e.access_count;
1361        Ok(m)
1362    }
1363
1364    /// Serialize and store a memory (insert or overwrite by ID).
1365    fn put(&mut self, mut m: Memory) -> Result<Ulid> {
1366        if m.vector.len() != self.dimensions {
1367            return Err(MnemoError::DimensionMismatch {
1368                expected: self.dimensions,
1369                got: m.vector.len(),
1370            });
1371        }
1372        // Reject all-zero vectors (Phase 3 PR 10a; breaking vs 0.4.x).
1373        // A zero vector scores exactly 0.0 on every cosine query — it
1374        // can't contribute signal, but recency / importance can still
1375        // pull it into top-k as noise. We'd rather fail loudly at
1376        // `remember` than let a mis-embedded entry pollute recall
1377        // quality for the life of the database. Scaffold manifests now
1378        // use a sentinel unit vector so this check doesn't trip them.
1379        if embed::is_zero_vector(&m.vector) {
1380            return Err(MnemoError::Invalid(
1381                "all-zero vector refused: it scores 0.0 on every cosine query and \
1382                 pollutes recall; check your embedder returned a real embedding"
1383                    .into(),
1384            ));
1385        }
1386        if m.id == Ulid::nil() {
1387            m.id = Ulid::new();
1388        }
1389        let id_u: u128 = m.id.0;
1390        let bytes = rmp_serde::to_vec(&m).map_err(|e| MnemoError::Serialize(e.to_string()))?;
1391        let (start, pc) = self.write_record(&bytes)?;
1392        // For overwrites preserve the existing access stats; for inserts
1393        // seed them from the memory (which carries them in its body).
1394        let (prev_accessed, prev_count) = match self.index.get(&id_u).copied() {
1395            Some(idx) => (
1396                self.catalog[idx].accessed_at,
1397                self.catalog[idx].access_count,
1398            ),
1399            None => (m.accessed_at, m.access_count),
1400        };
1401        let entry = CatalogEntry {
1402            id: id_u,
1403            start_page: start,
1404            page_count: pc,
1405            len: bytes.len() as u32,
1406            deleted: false,
1407            accessed_at: prev_accessed,
1408            access_count: prev_count,
1409        };
1410        match self.index.get(&id_u).copied() {
1411            Some(idx) => self.catalog[idx] = entry,
1412            None => {
1413                self.index.insert(id_u, self.catalog.len());
1414                self.catalog.push(entry);
1415            }
1416        }
1417        self.dirty_catalog = true;
1418
1419        // Keep the ANN index complete: assign the (new or changed) vector to
1420        // its nearest partition. Centroids/codebook stay fixed until rebuild.
1421        if let Some(ann) = &mut self.ann {
1422            ann.add(id_u, &m.vector);
1423            self.dirty_index = true;
1424        }
1425
1426        // FIX-1 R-001.8 part 2: keep the lazy agent set in sync. If
1427        // it hasn't been built yet (no `stats()` call this session),
1428        // skip — the first `stats()` will build it from scratch. If
1429        // it has, add this memory's agent_id. An overwrite with the
1430        // same id and same agent is a no-op (BTreeSet::insert returns
1431        // false); an overwrite with a *changed* agent_id adds the new
1432        // agent but doesn't drop the old (we can't tell without a
1433        // decrypt whether the old agent still has live memories).
1434        // That's conservative in the right direction — `stats()` may
1435        // over-report by one agent until the next delete invalidates.
1436        if let Some(set) = &mut self.agent_set {
1437            set.insert(m.agent_id.clone());
1438        }
1439        Ok(m.id)
1440    }
1441
1442    // --- public CRUD ------------------------------------------------------
1443
1444    /// Register a text embedder for this handle. Enables
1445    /// [`Mnemo::remember_text`] and the text-side `RecallRequest`
1446    /// convenience path. Does two checks before storing the embedder:
1447    ///
1448    /// 1. **Dimension check.** The embedder's `identity().dimensions`
1449    ///    must equal [`Mnemo::dimensions`] (the width this database
1450    ///    was created with). Mismatch returns
1451    ///    [`MnemoError::DimensionMismatch`].
1452    /// 2. **Manifest identity check.** If the scaffold / onboarding
1453    ///    manifest already records an embedder — the canonical shape
1454    ///    is `metadata.embedder = { name, dimensions, normalize }`,
1455    ///    with the legacy flat `embedder_name` / `embedder_dimensions`
1456    ///    / `embedder_normalize` keys accepted as an alias — the new
1457    ///    embedder's identity must match exactly. A different name —
1458    ///    even at the same dimensions — returns
1459    ///    [`MnemoError::Invalid`] with
1460    ///    [`EmbedderIdentity::MISMATCH_MSG`] in the message. If the
1461    ///    manifest has no embedder metadata yet, this call upserts
1462    ///    the nested object into the first onboarding/manifest memory
1463    ///    (write + flush so a subsequent reopen sees the identity).
1464    ///
1465    /// **Read-only handles.** `set_embedder` on an `open_read_only`
1466    /// handle performs both checks and adopts the embedder in memory;
1467    /// it never writes. Readers (CLI `--text`, agent-side `recall_text`)
1468    /// can therefore use the text path next to a writer that holds
1469    /// the exclusive lock.
1470    ///
1471    /// On both failure paths the handle's embedder slot is left
1472    /// unchanged. On success the embedder is wrapped in an `Arc<dyn
1473    /// Embedder>` and consulted by every text-side call made through
1474    /// this handle. The slot is per-handle and not persisted — a
1475    /// fresh open must re-register (same identity, so the manifest
1476    /// check round-trips cleanly).
1477    pub fn set_embedder<E: Embedder + 'static>(&mut self, embedder: E) -> Result<()> {
1478        let identity = embedder.identity();
1479        if identity.dimensions != self.dimensions {
1480            return Err(MnemoError::DimensionMismatch {
1481                expected: self.dimensions,
1482                got: identity.dimensions,
1483            });
1484        }
1485
1486        // Find a scaffold/onboarding manifest to inspect (or upsert
1487        // into). `about()` already filters + sorts the onboarding set.
1488        let onboarding = self.about()?;
1489        let mut upsert_target: Option<usize> = None;
1490        for (i, m) in onboarding.iter().enumerate() {
1491            if let Some(saw) = embed::read_identity_from_metadata(&m.metadata) {
1492                if saw != identity {
1493                    return Err(embed::identity_mismatch_err(&saw, &identity));
1494                }
1495                // Match — nothing to persist; adopt + done. Works the
1496                // same on read-only and read-write handles.
1497                self.embedder = Some(std::sync::Arc::new(embedder));
1498                return Ok(());
1499            }
1500            if upsert_target.is_none() {
1501                upsert_target = Some(i);
1502            }
1503        }
1504
1505        // No existing identity on the manifest. On a read-write handle
1506        // upsert into the first onboarding memory so later opens see
1507        // it; on a read-only handle skip the write (readers must not
1508        // mutate the file), but still adopt the embedder in memory.
1509        if !self.read_only {
1510            if let Some(idx) = upsert_target {
1511                let mut m = onboarding[idx].clone();
1512                embed::write_identity_into_metadata(&mut m.metadata, &identity);
1513                // Reuse the memory's existing id so this is an in-place
1514                // update of the scaffold manifest, not a new entry.
1515                self.put(m)?;
1516                self.flush()?;
1517            }
1518        }
1519
1520        self.embedder = Some(std::sync::Arc::new(embedder));
1521        Ok(())
1522    }
1523
1524    /// Borrow the currently-registered embedder, if any. Returns
1525    /// `None` when no embedder has been set via
1526    /// [`Mnemo::set_embedder`]. Bindings use this to drive their own
1527    /// text-side convenience methods without duplicating the identity
1528    /// check.
1529    pub fn embedder(&self) -> Option<&dyn Embedder> {
1530        self.embedder.as_deref()
1531    }
1532
1533    /// Shared handle to the registered embedder, cloned so the caller
1534    /// can drop the borrow on this `Mnemo` before invoking `embed()`.
1535    /// FIX-1 / PY-1 (R-001.11): the Python binding needs this so an
1536    /// embedder callback that re-enters `Mnemo` (e.g. a Python
1537    /// embedder that calls `db.dimensions()` or `db.cache_get(...)`)
1538    /// doesn't hit PyO3's "Already borrowed" — we clone the Arc under
1539    /// a short `&self` borrow, release that borrow, then invoke
1540    /// `embed()` with no `&mut Mnemo` held.
1541    pub fn embedder_arc(&self) -> Option<std::sync::Arc<dyn Embedder>> {
1542        self.embedder.clone()
1543    }
1544
1545    /// Embed `content` with the registered embedder and `remember` the
1546    /// result. Returns the new memory's ULID. Convenience path for
1547    /// callers who have a string, want it stored, and don't need any
1548    /// of the per-memory options [`Memory`] exposes (agent, importance,
1549    /// metadata, ttl, scope) — build a [`Memory`] by hand and call
1550    /// [`Mnemo::remember`] for those.
1551    ///
1552    /// Fails with [`MnemoError::Invalid`] if no embedder has been
1553    /// registered on this handle (call [`Mnemo::set_embedder`] first);
1554    /// with [`MnemoError::DimensionMismatch`] if the embedder returned
1555    /// the wrong width for this database.
1556    pub fn remember_text(&mut self, content: &str, memory_type: MemoryType) -> Result<Ulid> {
1557        if self.read_only {
1558            return Err(MnemoError::ReadOnly);
1559        }
1560        let embedder = self.embedder.as_ref().ok_or_else(|| {
1561            MnemoError::Invalid(
1562                "no embedder registered on this handle; call set_embedder first".into(),
1563            )
1564        })?;
1565        let mut vectors = embedder.embed(&[content])?;
1566        let vector = vectors.pop().ok_or_else(|| {
1567            MnemoError::Invalid("embedder returned no vector for the input text".into())
1568        })?;
1569        if vector.len() != self.dimensions {
1570            return Err(MnemoError::DimensionMismatch {
1571                expected: self.dimensions,
1572                got: vector.len(),
1573            });
1574        }
1575        let memory = Memory::new(content.to_string(), memory_type, vector);
1576        self.remember(memory)
1577    }
1578
1579    /// Store a memory. Returns its ULID. If the memory's ID is nil a fresh
1580    /// one is assigned; an existing ID overwrites in place.
1581    pub fn remember(&mut self, memory: Memory) -> Result<Ulid> {
1582        if self.read_only {
1583            return Err(MnemoError::ReadOnly);
1584        }
1585        self.put(memory)
1586    }
1587
1588    /// Fetch a memory by ID.
1589    pub fn get(&mut self, id: &Ulid) -> Result<Memory> {
1590        let idx = *self
1591            .index
1592            .get(&id.0)
1593            .ok_or_else(|| MnemoError::NotFound(id.to_string()))?;
1594        let entry = self.catalog[idx].clone();
1595        if entry.deleted {
1596            return Err(MnemoError::NotFound(id.to_string()));
1597        }
1598        self.read_memory(&entry)
1599    }
1600
1601    /// Soft-delete a memory (tombstoned; space reclaimed by `compact`).
1602    pub fn delete(&mut self, id: &Ulid) -> Result<()> {
1603        if self.read_only {
1604            return Err(MnemoError::ReadOnly);
1605        }
1606        let idx = *self
1607            .index
1608            .get(&id.0)
1609            .ok_or_else(|| MnemoError::NotFound(id.to_string()))?;
1610        if !self.catalog[idx].deleted {
1611            self.catalog[idx].deleted = true;
1612            self.dirty_catalog = true;
1613        }
1614        if let Some(ann) = &mut self.ann {
1615            ann.remove(id.0);
1616            self.dirty_index = true;
1617        }
1618        // FIX-1 R-001.8 part 2: we can't tell whether this delete
1619        // removed the last memory of its agent without a decrypt
1620        // (the catalog entry doesn't carry `agent_id` in format v8),
1621        // so just invalidate the lazy set. The next `stats()` call
1622        // rebuilds from live memories.
1623        self.agent_set = None;
1624        Ok(())
1625    }
1626
1627    /// Return every live memory (used by tooling and compaction).
1628    pub fn memories(&mut self) -> Result<Vec<Memory>> {
1629        let entries: Vec<CatalogEntry> = self
1630            .catalog
1631            .iter()
1632            .filter(|e| !e.deleted)
1633            .cloned()
1634            .collect();
1635        let mut out = Vec::with_capacity(entries.len());
1636        for e in &entries {
1637            out.push(self.read_memory(e)?);
1638        }
1639        Ok(out)
1640    }
1641
1642    /// Return the database's **self-describing onboarding memories** — the
1643    /// ones tagged `metadata.area = "onboarding"`. This is the engine-level
1644    /// surface for the *single-file philosophy*: an agent who receives a
1645    /// `.mnemo` file (and its passphrase) can call this to learn what the
1646    /// file is, which embedder it expects, the recommended agent id, and any
1647    /// other conventions the file's author chose to record — all without
1648    /// needing any external documentation.
1649    ///
1650    /// Ordering: the canonical manifest (tag `metadata.topic = "manifest"`)
1651    /// always comes first; everything else follows in `importance` descending,
1652    /// then `created_at` ascending for deterministic results.
1653    pub fn about(&mut self) -> Result<Vec<Memory>> {
1654        let mut out: Vec<Memory> = self
1655            .memories()?
1656            .into_iter()
1657            .filter(|m| {
1658                m.metadata
1659                    .get("area")
1660                    .and_then(|v| v.as_str())
1661                    .map(|s| s.eq_ignore_ascii_case("onboarding"))
1662                    .unwrap_or(false)
1663            })
1664            .collect();
1665        out.sort_by(|a, b| {
1666            let a_manifest = is_manifest(a);
1667            let b_manifest = is_manifest(b);
1668            // Manifest topic wins first; then importance desc; then created_at asc.
1669            b_manifest
1670                .cmp(&a_manifest)
1671                .then_with(|| b.importance.total_cmp(&a.importance))
1672                .then_with(|| a.created_at.cmp(&b.created_at))
1673        });
1674        Ok(out)
1675    }
1676
1677    // --- retrieval --------------------------------------------------------
1678
1679    /// Phase-1 brute-force search: rank live memories by raw similarity only.
1680    /// Read-only — does not update access statistics.
1681    pub fn search(
1682        &mut self,
1683        query: &[f32],
1684        top_k: usize,
1685        metric: Metric,
1686    ) -> Result<Vec<(Memory, f32)>> {
1687        if query.len() != self.dimensions {
1688            return Err(MnemoError::DimensionMismatch {
1689                expected: self.dimensions,
1690                got: query.len(),
1691            });
1692        }
1693        let now = memory::now_secs();
1694        let entries: Vec<CatalogEntry> = self
1695            .catalog
1696            .iter()
1697            .filter(|e| !e.deleted)
1698            .cloned()
1699            .collect();
1700
1701        let mut scored: Vec<(Memory, f32)> = Vec::new();
1702        for e in &entries {
1703            let m = self.read_memory(e)?;
1704            if m.is_expired(now) {
1705                continue;
1706            }
1707            let sim = memory::similarity(metric, query, &m.vector);
1708            scored.push((m, sim));
1709        }
1710        scored.sort_by(|a, b| b.1.total_cmp(&a.1));
1711        scored.truncate(top_k);
1712        Ok(scored)
1713    }
1714
1715    /// Phase-5 multi-signal recall: rank by
1716    /// `α·similarity + β·recency + γ·importance + δ·ln(freq)`, with type and
1717    /// agent filtering. Updates `accessed_at` / `access_count` on the
1718    /// returned memories (persisted on the next `flush`).
1719    ///
1720    /// When an ANN index has been built ([`Mnemo::build_index`]), recall runs
1721    /// the tiered IVF→PQ→rerank pipeline: it scores only the similarity-nearest
1722    /// `n_rerank` candidates rather than every memory. Without an index it
1723    /// scores every live memory exactly.
1724    pub fn recall(&mut self, req: &RecallRequest) -> Result<Vec<RecallResult>> {
1725        if req.query.len() != self.dimensions {
1726            return Err(MnemoError::DimensionMismatch {
1727                expected: self.dimensions,
1728                got: req.query.len(),
1729            });
1730        }
1731        // Access-tracking bumps `accessed_at` / `access_count` in the
1732        // catalog, which is a mutation. On a read-only handle the caller
1733        // must pass `track_access(false)` (via CLI `--no-track`).
1734        if self.read_only && req.track_access {
1735            return Err(MnemoError::ReadOnly);
1736        }
1737        let now = memory::now_secs();
1738
1739        // Candidate set: ANN-narrowed when an index exists, else everything.
1740        let entries: Vec<CatalogEntry> = if let Some(ann) = &self.ann {
1741            let ids = ann.query(&req.query, req.n_probe, req.n_rerank);
1742            ids.iter()
1743                .filter_map(|id| self.index.get(id).copied())
1744                .map(|i| self.catalog[i].clone())
1745                .filter(|e| !e.deleted)
1746                .collect()
1747        } else {
1748            self.catalog
1749                .iter()
1750                .filter(|e| !e.deleted)
1751                .cloned()
1752                .collect()
1753        };
1754
1755        // Scoring v2 (shipped v0.7.0). Single-pass filter + score:
1756        //   * `age_secs = now - m.created_at`, so a recall doesn't bump
1757        //     its own results' recency on later queries.
1758        //   * `ScoreWeights::score` applies the similarity-gated,
1759        //     importance-centered, saturating-usage formula documented
1760        //     on `ScoreWeights`.
1761        let mut scored: Vec<RecallResult> = Vec::with_capacity(entries.len());
1762        for e in &entries {
1763            let m = self.read_memory(e)?;
1764            if m.is_expired(now) {
1765                continue;
1766            }
1767            if let Some(types) = &req.memory_types {
1768                if !types.contains(&m.memory_type) {
1769                    continue;
1770                }
1771            }
1772            if let Some(agent) = &req.agent_id {
1773                let visible = m.agent_id == *agent || m.scope == Scope::Shared;
1774                if !visible {
1775                    continue;
1776                }
1777            }
1778            let sim = memory::similarity(req.metric, &req.query, &m.vector);
1779            let age_secs = (now - m.created_at) as f32;
1780            let score = req
1781                .weights
1782                .score(sim, age_secs, m.importance, m.access_count);
1783            scored.push(RecallResult {
1784                memory: m,
1785                score,
1786                similarity: sim,
1787            });
1788        }
1789        scored.sort_by(|a, b| b.score.total_cmp(&a.score));
1790        scored.truncate(req.top_k);
1791
1792        // Update access statistics for everything we surfaced — but only
1793        // mutate the catalog entry, not the full record on disk. Pre-v5
1794        // this loop called `self.put(r.memory.clone())`, which serialized
1795        // and rewrote every result (vector included) to fresh pages,
1796        // turning a top-K recall into K full record rewrites. Now the
1797        // catalog is the source of truth for these two fields and the
1798        // record body stays untouched until the next real edit.
1799        if req.track_access {
1800            for r in &mut scored {
1801                if let Some(&idx) = self.index.get(&r.memory.id.0) {
1802                    let entry = &mut self.catalog[idx];
1803                    entry.accessed_at = now;
1804                    entry.access_count = entry.access_count.saturating_add(1);
1805                    // Propagate the just-updated values onto the returned
1806                    // Memory so the caller sees the post-recall state.
1807                    r.memory.accessed_at = entry.accessed_at;
1808                    r.memory.access_count = entry.access_count;
1809                }
1810            }
1811            // Mark the catalog dirty so the next flush persists these
1812            // bumps. The records themselves are not dirty.
1813            if !scored.is_empty() {
1814                self.dirty_catalog = true;
1815            }
1816        }
1817        Ok(scored)
1818    }
1819
1820    // --- approximate index ------------------------------------------------
1821
1822    /// Build an IVF+PQ approximate-nearest-neighbour index over every live
1823    /// memory, using default tuning. After this, [`Mnemo::recall`] runs the
1824    /// tiered pipeline instead of an exact scan. Persisted on the next
1825    /// `flush`. Returns a snapshot of the index shape.
1826    pub fn build_index(&mut self) -> Result<IndexInfo> {
1827        self.build_index_with(IndexConfig::default())
1828    }
1829
1830    /// Build the ANN index with explicit tuning.
1831    pub fn build_index_with(&mut self, cfg: IndexConfig) -> Result<IndexInfo> {
1832        if self.read_only {
1833            return Err(MnemoError::ReadOnly);
1834        }
1835        let mems = self.memories()?;
1836        if mems.is_empty() {
1837            return Err(MnemoError::Invalid(
1838                "cannot build an index over an empty database".into(),
1839            ));
1840        }
1841        let items: Vec<(u128, &[f32])> =
1842            mems.iter().map(|m| (m.id.0, m.vector.as_slice())).collect();
1843        let idx = IvfPqIndex::build(self.dimensions, &items, cfg)?;
1844        let info = idx.info();
1845        self.ann = Some(idx);
1846        self.dirty_index = true;
1847        Ok(info)
1848    }
1849
1850    /// Rebuild the ANN index from scratch — re-clusters centroids and retrains
1851    /// the PQ codebook, undoing the cluster drift that accumulates as memories
1852    /// are inserted against fixed centroids.
1853    pub fn rebuild_index(&mut self) -> Result<IndexInfo> {
1854        // ANN-1.1: re-derive every parameter (partitions, subspaces,
1855        // n_probe, n_rerank) for the current corpus size by passing
1856        // `IndexConfig::default()`. Previously this freezed the
1857        // stored `(n_probe, n_rerank)` from the current index,
1858        // defeating the point of rebuilding on a grown corpus. If
1859        // you really want to pin specific values, call
1860        // `build_index_with(IndexConfig { … })` directly.
1861        self.build_index_with(IndexConfig::default())
1862    }
1863
1864    /// Drop the ANN index; recall reverts to exact scans. Persisted on flush.
1865    /// No-op on a read-only handle (signature is infallible — callers of
1866    /// `build_index` / `flush` on a read-only handle already get a typed
1867    /// `ReadOnly` error, which is the enforceable boundary).
1868    pub fn drop_index(&mut self) {
1869        if self.read_only {
1870            return;
1871        }
1872        if self.ann.is_some() {
1873            self.ann = None;
1874            self.dirty_index = true;
1875        }
1876    }
1877
1878    /// Whether an ANN index is currently loaded.
1879    pub fn has_index(&self) -> bool {
1880        self.ann.is_some()
1881    }
1882
1883    // --- durability & maintenance ----------------------------------------
1884
1885    /// Seal a serialized buffer into a fresh run of encrypted home pages and
1886    /// append a WAL frame for each. Returns the run's `(start_page, pages)`.
1887    /// The pages are *not* written to their home locations here — they go to
1888    /// the WAL and are folded in later by [`Mnemo::checkpoint`].
1889    fn seal_run(&mut self, bytes: &[u8], frames: &mut Vec<wal::Frame>) -> Result<(u64, u32)> {
1890        let pc = bytes.len().div_ceil(PAYLOAD).max(1);
1891        let start = self.header.next_page;
1892        self.header.next_page += pc as u64;
1893        for i in 0..pc {
1894            let lo = i * PAYLOAD;
1895            let hi = ((i + 1) * PAYLOAD).min(bytes.len());
1896            let page_no = start + i as u64;
1897            let sealed = self.pager.seal_page(page_no, &bytes[lo..hi])?;
1898            frames.push((page_no, sealed.to_vec()));
1899        }
1900        Ok((start, pc as u32))
1901    }
1902
1903    /// Grow the WAL region if `frame_pages` worth of frames would not fit.
1904    /// Called at the top of `flush`, where the WAL is always spent (every
1905    /// flush ends with a checkpoint), so relocating it cannot strand a
1906    /// committed transaction.
1907    fn ensure_wal_capacity(&mut self, frame_pages: usize) -> Result<()> {
1908        let need = wal::txn_byte_len(frame_pages, PAGE_SIZE);
1909        if need <= self.header.wal_pages * PAGE_SIZE as u64 {
1910            return Ok(());
1911        }
1912        // Allocate a new, larger region at the file tail (~1.5x headroom).
1913        let req = need.div_ceil(PAGE_SIZE as u64);
1914        let new_pages = (req + req / 2 + 4).max(DEFAULT_WAL_PAGES);
1915        let old_wal_start = self.header.wal_start;
1916        let new_start = self.header.next_page;
1917        self.header.next_page += new_pages;
1918        self.header.wal_start = new_start;
1919        self.header.wal_pages = new_pages;
1920        self.header.wal_seq += 1;
1921        // DUR-1 (R-002.1): WAL-log the header change into the *new* region
1922        // first, checkpoint page 0, then invalidate the old region. A crash
1923        // anywhere here either keeps the old state (if page 0 wasn't yet
1924        // overwritten) or recovers to the new state via the torn-header
1925        // scan in `Mnemo::open`, which prefers the highest-txn-id valid
1926        // committed header frame.
1927        self.relocate_header_through_wal(old_wal_start)
1928    }
1929
1930    /// Shared helper for every post-`create` page-0 commit: WAL-log the
1931    /// new header into the current WAL region, checkpoint page 0, and
1932    /// invalidate `old_wal_start` (the previous region, or the current
1933    /// region when it hasn't moved). The caller is expected to have
1934    /// mutated `self.header` to its target values *including a bumped
1935    /// `wal_seq`* (one more than the last on-disk `wal_seq`); this
1936    /// routine applies the seal, writes the frame + commit, folds page 0,
1937    /// then zeroes the old region's first frame header.
1938    ///
1939    /// If `old_wal_start == self.header.wal_start`, the invalidation is
1940    /// skipped — the region didn't move, so overwriting its first frame
1941    /// would destroy the just-committed one.
1942    fn relocate_header_through_wal(&mut self, old_wal_start: u64) -> Result<()> {
1943        self.header.write_counter = self.pager.write_counter;
1944        let mut page = self.header.to_page();
1945        self.header.apply_seal(&mut page, self.pager.dek())?;
1946        let txn_id = self.header.wal_seq;
1947        let frames = vec![(0u64, page.to_vec())];
1948        wal::commit(
1949            self.pager.file_mut(),
1950            self.header.wal_start,
1951            self.header.wal_pages,
1952            txn_id,
1953            &frames,
1954        )?;
1955        self.pager.write_raw(0, &page)?;
1956        self.pager.sync()?;
1957        if old_wal_start != self.header.wal_start {
1958            wal::invalidate_region(self.pager.file_mut(), old_wal_start)?;
1959        }
1960        Ok(())
1961    }
1962
1963    /// Fold a committed transaction's WAL frames into their home pages.
1964    fn checkpoint(&mut self, frames: &[wal::Frame]) -> Result<()> {
1965        for (page_no, bytes) in frames {
1966            let mut img = [0u8; PAGE_SIZE];
1967            img.copy_from_slice(bytes);
1968            self.pager.write_sealed(*page_no, &img)?;
1969        }
1970        self.pager.sync()?;
1971        Ok(())
1972    }
1973
1974    /// Persist all pending changes as one **write-ahead-logged transaction**.
1975    ///
1976    /// Sequence:
1977    ///
1978    /// 1. **Prepare.** Serialize the control plane (catalog, ANN index),
1979    ///    grow the WAL if needed, and **lease** counter and page slots —
1980    ///    bump `header.write_counter` and `header.next_page` by upper
1981    ///    bounds on this transaction's consumption and persist that
1982    ///    *leased* header to disk before any encrypted page is written.
1983    /// 2. **Data pages.** [`Pager::flush`] writes dirty record pages
1984    ///    copy-on-write under fresh nonces and fsyncs them. The orphan-page
1985    ///    nonce-reuse window (see [Phase 1.1 of the improvement plan])
1986    ///    is closed because the leased header on disk already records a
1987    ///    `write_counter` past anything this step can produce.
1988    /// 3. **Control plane.** Seal the new catalog / ANN index / snapshot
1989    ///    manifest into fresh home page runs.
1990    /// 4. **Commit.** Log all of step 3's frames plus the final header
1991    ///    frame into the WAL and fsync — the durability point.
1992    /// 5. **Checkpoint.** Fold the WAL into the home pages.
1993    ///
1994    /// A crash before step 4 leaves the previous state intact. A crash
1995    /// after step 4 is repaired by [`Mnemo::open`] replaying the WAL.
1996    /// Safe to call repeatedly.
1997    pub fn flush(&mut self) -> Result<()> {
1998        if self.read_only {
1999            return Err(MnemoError::ReadOnly);
2000        }
2001        // Snapshot the thread-local fsync counter. Every `sync_all` the
2002        // pager / wal / checkpoint path makes between here and the end
2003        // of this function bumps the same counter; `delta` lands on
2004        // `self.fsyncs_last_flush` so `Stats::fsyncs_last_flush` can
2005        // report it. The planner wants this to decide whether cutting
2006        // fsync count is a viable write-latency win before we touch
2007        // the write path's durability model.
2008        let fsync_before = crate::fsync_count();
2009        if !self.dirty_catalog && !self.dirty_index && !self.dirty_cache {
2010            // No control-plane changes pending. Because every `put` that
2011            // dirties a data page also dirties the catalog, and every
2012            // `cache_put` that dirties a data page also dirties
2013            // `dirty_cache`, there can't be dirty data pages either —
2014            // nothing to do.
2015            self.fsyncs_last_flush = 0;
2016            return Ok(());
2017        }
2018
2019        // FIX-1 R-001.8 part 3: snapshot the memory/index-dirty flags
2020        // before step 3b decides whether to push a restore point.
2021        // Only flushes that change the memory catalog or ANN index
2022        // need a snapshot — a cache-only flush (`dirty_cache` set,
2023        // `dirty_catalog` + `dirty_index` both false) must not push,
2024        // because snapshots don't capture cache state and every push
2025        // counts against `max_snapshots` (default 256), so a cache-
2026        // heavy workload would otherwise evict every real memory
2027        // restore point within ~256 cache writes.
2028        let is_memory_or_index_flush = self.dirty_catalog || self.dirty_index;
2029
2030        // 1. Serialize control plane, ensure WAL capacity, lease counter +
2031        //    page slots, and persist the leased header *before* any data
2032        //    page hits the disk.
2033        let ctx = self.prepare_for_flush()?;
2034
2035        // 2. Record (vector) data pages: copy-on-write to fresh pages, fsynced.
2036        //    Safe: the lease guarantees that even if we crash here, reopen
2037        //    will see a `write_counter` past every nonce this step uses.
2038        self.pager.flush()?;
2039
2040        // 3. Seal the catalog / index control pages into fresh home runs.
2041        let mut frames: Vec<wal::Frame> = Vec::new();
2042        if let Some(bytes) = &ctx.cat_bytes {
2043            let (start, pages) = self.seal_run(bytes, &mut frames)?;
2044            self.header.catalog_start = start;
2045            self.header.catalog_pages = pages as u64;
2046            self.header.catalog_len = bytes.len() as u64;
2047        }
2048        if self.dirty_index {
2049            match &ctx.idx_bytes {
2050                Some(bytes) => {
2051                    let (start, pages) = self.seal_run(bytes, &mut frames)?;
2052                    self.header.index_start = start;
2053                    self.header.index_pages = pages as u64;
2054                    self.header.index_len = bytes.len() as u64;
2055                }
2056                None => {
2057                    self.header.index_start = 0;
2058                    self.header.index_pages = 0;
2059                    self.header.index_len = 0;
2060                }
2061            }
2062        }
2063        // v8+ result-cache directory (Phase 10.1). Seal the serialized
2064        // directory into a fresh page run and update the pointer in the
2065        // header. The seal AAD already covers cache_start/pages/len for
2066        // v8+ (see `header_seal_aad`), so tampering with the pointer
2067        // trips the header seal on the next open.
2068        if self.dirty_cache {
2069            match &ctx.cache_bytes {
2070                Some(bytes) if !self.cache_dir.is_empty() => {
2071                    let (start, pages) = self.seal_run(bytes, &mut frames)?;
2072                    self.header.cache_start = start;
2073                    self.header.cache_pages = pages as u64;
2074                    self.header.cache_len = bytes.len() as u64;
2075                }
2076                _ => {
2077                    // Either the directory serialized to nothing (all
2078                    // entries deleted after tombstone-only migration) or
2079                    // it's empty — collapse the pointer to zero.
2080                    self.header.cache_start = 0;
2081                    self.header.cache_pages = 0;
2082                    self.header.cache_len = 0;
2083                }
2084            }
2085        }
2086
2087        // 3b. Record this transaction as a restorable snapshot. The manifest
2088        //     update is staged in a local and adopted only once the commit
2089        //     below succeeds, so a failed flush leaves no phantom entry.
2090        //
2091        //     The cap (`self.max_snapshots`) is applied to the staged local
2092        //     before the new entry lands, so after appending the manifest
2093        //     holds at most `max_snapshots` entries — the most-recent N.
2094        //     `max_snapshots == 0` disables the cap. Pages referenced only
2095        //     by pruned snapshots stay on disk until `compact_file`.
2096        //
2097        //     FIX-1 R-001.8 part 3: skip the snapshot push on a
2098        //     cache-only flush. Snapshots don't capture the cache
2099        //     anyway, and every push counts against `max_snapshots`
2100        //     (default 256) + rewrites the manifest run — so a
2101        //     cache-heavy workload would otherwise evict every real
2102        //     memory restore point within ~256 cache writes. On a
2103        //     cache-only flush we reuse the current manifest
2104        //     verbatim; the manifest run already lives at
2105        //     `self.header.manifest_{start,pages,len}` (set by the
2106        //     last memory-touching flush, or at `create` time).
2107        let live = self.len() as u64;
2108        let txn_id = self.header.wal_seq + 1;
2109        let mut manifest = self.manifest.clone();
2110        if is_memory_or_index_flush {
2111            if self.max_snapshots > 0 {
2112                // Drop the oldest entries so there's room for the new one.
2113                while manifest.len() >= self.max_snapshots {
2114                    manifest.remove(0);
2115                }
2116            }
2117            manifest.push(Snapshot {
2118                txn_id,
2119                created_at: memory::now_secs(),
2120                catalog_start: self.header.catalog_start,
2121                catalog_pages: self.header.catalog_pages,
2122                catalog_len: self.header.catalog_len,
2123                index_start: self.header.index_start,
2124                index_pages: self.header.index_pages,
2125                index_len: self.header.index_len,
2126                memory_count: live,
2127            });
2128            let man_bytes =
2129                rmp_serde::to_vec(&manifest).map_err(|e| MnemoError::Serialize(e.to_string()))?;
2130            let (m_start, m_pages) = self.seal_run(&man_bytes, &mut frames)?;
2131            self.header.manifest_start = m_start;
2132            self.header.manifest_pages = m_pages as u64;
2133            self.header.manifest_len = man_bytes.len() as u64;
2134        }
2135
2136        // 3c. The header is the transaction's final frame; stamp the new id
2137        //     and the *actual* (post-flush) write counter, which is <= the
2138        //     leased value persisted in `prepare_for_flush`. This is the
2139        //     value that ultimately replaces the leased header on checkpoint.
2140        self.header.vector_count = live;
2141        self.header.write_counter = self.pager.write_counter;
2142        self.header.wal_seq = txn_id;
2143        let mut hpage = self.header.to_page();
2144        self.header.apply_seal(&mut hpage, self.pager.dek())?;
2145        frames.push((0, hpage.to_vec()));
2146
2147        // 4. COMMIT — log the transaction and fsync. Nothing at a home page
2148        //    other than the leased header has changed yet; this single
2149        //    fsync is what makes the rest durable.
2150        let (wal_start, wal_pages) = (self.header.wal_start, self.header.wal_pages);
2151        wal::commit(self.pager.file_mut(), wal_start, wal_pages, txn_id, &frames)?;
2152
2153        // 5. Checkpoint — fold the WAL into the home pages.
2154        self.checkpoint(&frames)?;
2155
2156        self.manifest = manifest;
2157        self.dirty_catalog = false;
2158        self.dirty_index = false;
2159        self.dirty_cache = false;
2160        // Batched policy: successful flush consumes the pending batch.
2161        self.cache_batch.reset();
2162        // Record how many fsyncs this flush issued, for Stats reporting.
2163        // Thread-local counter guarantees we see only this handle's own
2164        // sync_all calls between the snapshot above and this read.
2165        self.fsyncs_last_flush =
2166            (crate::fsync_count().wrapping_sub(fsync_before)).min(u32::MAX as u64) as u32;
2167        Ok(())
2168    }
2169
2170    /// Pre-flush prelude shared by [`Mnemo::flush`] and the
2171    /// `__crash_partial_flush_for_testing` hook.
2172    ///
2173    /// Serializes the control plane (catalog, ANN index, cache directory)
2174    /// and grows the WAL if needed. That's it: this used to persist a
2175    /// leased header too (SEC-1 obsoleted it; see DUR-1 R-002.1 note in
2176    /// the body for the full rationale), but after DUR-1 the only page-0
2177    /// writes outside `create` go through the WAL via
2178    /// [`Mnemo::relocate_header_through_wal`].
2179    ///
2180    /// The historical pre-stamping used to close the nonce-reuse window
2181    /// flagged in Phase 1.1 of the improvement plan: a crash anywhere
2182    /// between here and the WAL commit left a file whose on-disk header
2183    /// recorded a `write_counter` past every nonce the rest of this
2184    /// transaction could produce, so the next reopen started from a
2185    /// counter that guaranteed uniqueness against any orphan data pages
2186    /// still on disk. SEC-1's per-session random nonce prefix gives the
2187    /// same guarantee without needing a persisted counter, which is why
2188    /// this routine no longer writes page 0.
2189    fn prepare_for_flush(&mut self) -> Result<FlushPrelude> {
2190        // Serialize.
2191        let cat_bytes = if self.dirty_catalog {
2192            Some(
2193                rmp_serde::to_vec(&self.catalog)
2194                    .map_err(|e| MnemoError::Serialize(e.to_string()))?,
2195            )
2196        } else {
2197            None
2198        };
2199        let idx_bytes: Option<Vec<u8>> = if self.dirty_index {
2200            match &self.ann {
2201                Some(ann) => {
2202                    Some(rmp_serde::to_vec(ann).map_err(|e| MnemoError::Serialize(e.to_string()))?)
2203                }
2204                None => None,
2205            }
2206        } else {
2207            None
2208        };
2209        // Serialize the result-cache directory when it's dirty. An
2210        // empty directory serializes to a small header + 0-element
2211        // array; still cheap enough that we don't special-case it.
2212        let cache_bytes: Option<Vec<u8>> = if self.dirty_cache {
2213            Some(
2214                rmp_serde::to_vec(&self.cache_dir)
2215                    .map_err(|e| MnemoError::Serialize(e.to_string()))?,
2216            )
2217        } else {
2218            None
2219        };
2220
2221        let cat_pc = cat_bytes
2222            .as_ref()
2223            .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
2224        let idx_pc = idx_bytes
2225            .as_ref()
2226            .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
2227        let cache_pc = cache_bytes
2228            .as_ref()
2229            .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
2230        // Upper bound on the manifest run: one extra entry, <=82 bytes each
2231        // (rmp_serde encodes a Snapshot as a 9-element fixarray of i64/u64,
2232        // worst case 81 bytes plus a 1-byte array header).
2233        let man_upper = 9 + (self.manifest.len() + 1) * 82;
2234        let man_pc_est = man_upper.div_ceil(PAYLOAD).max(1);
2235
2236        // Grow WAL if needed (DUR-1: relocation is WAL-logged and does
2237        // its own fsyncs). The cache directory rides in the same
2238        // transaction so its page count contributes to the reservation.
2239        self.ensure_wal_capacity(cat_pc + idx_pc + cache_pc + man_pc_est + 1)?;
2240
2241        // DUR-1 (R-002.1): the pre-flush "leased header" write that used
2242        // to live here is gone. It existed to close the Phase 1.1 nonce
2243        // reuse window (reopen must see a `write_counter` past any nonce
2244        // this flush could produce before data pages hit disk). SEC-1
2245        // superseded that: nonces now carry a per-session random prefix
2246        // and `write_counter` is no longer a nonce input, so the leased
2247        // write bought us nothing but an extra fsync per flush. Orphan
2248        // data pages past `header.next_page` on a crashed flush are
2249        // still safe — the next session's random prefix makes any later
2250        // write at the same page_no uniquely nonced regardless of
2251        // `write_counter`.
2252
2253        Ok(FlushPrelude {
2254            cat_bytes,
2255            idx_bytes,
2256            cache_bytes,
2257        })
2258    }
2259
2260    /// Flush and close. Equivalent to `flush()` on a read-write handle; on
2261    /// a read-only handle it's a no-op (no dirty state to persist). The
2262    /// file is released on drop, which also drops the OS advisory lock.
2263    pub fn close(&mut self) -> Result<()> {
2264        if self.read_only {
2265            return Ok(());
2266        }
2267        self.flush()
2268    }
2269
2270    /// **TEST ONLY — do not use.** Reproduces a crash inside
2271    /// [`Mnemo::flush`] *after* the prelude has leased counter+page slots
2272    /// and persisted the leased header, but *before* the WAL is committed.
2273    /// Runs `prepare_for_flush` + `pager.flush` and returns — leaving
2274    /// data pages and a leased header on disk, with no commit frame.
2275    ///
2276    /// Used by `tests/integration.rs::nonce_unique_after_crashed_data_flush`
2277    /// to verify that this window (Phase 1.1 of the improvement plan) does
2278    /// not enable AES-GCM nonce reuse. Calling this in production code
2279    /// strands data pages; never expose it from a binding.
2280    #[doc(hidden)]
2281    pub fn __crash_partial_flush_for_testing(&mut self) -> Result<()> {
2282        if self.read_only {
2283            return Err(MnemoError::ReadOnly);
2284        }
2285        if !self.dirty_catalog && !self.dirty_index {
2286            return Ok(());
2287        }
2288        let _prelude = self.prepare_for_flush()?;
2289        self.pager.flush()
2290    }
2291
2292    /// Bound the in-memory page cache to `pages` decrypted pages.
2293    ///
2294    /// The cache holds decrypted page payloads to speed repeated reads. By
2295    /// default it is capped at 8192 pages (~64 MiB); lower the cap to trade
2296    /// hit rate for a smaller footprint, or raise it for a hotter cache. The
2297    /// cap governs *clean* pages — pages with un-flushed writes are always
2298    /// retained until [`Mnemo::flush`], regardless of the cap.
2299    pub fn set_cache_capacity(&mut self, pages: usize) {
2300        self.pager.set_cache_capacity(pages);
2301    }
2302
2303    /// Page-cache occupancy: `(pages_cached, capacity)`. Renamed from
2304    /// `cache_stats` in v0.4.0: the result-cache API (Phase 10.1) took
2305    /// that name for its per-namespace [`CacheStats`] summary, and the
2306    /// page-cache pair is the older/less-user-facing of the two, so it
2307    /// gets the more-specific name.
2308    pub fn page_cache_stats(&self) -> (usize, usize) {
2309        self.pager.cache_stats()
2310    }
2311
2312    // === Result cache (Phase 10.1 + 10.3) ============================
2313
2314    /// Insert or overwrite a cache entry at `(namespace, key)`.
2315    ///
2316    /// The engine SHA-256-hashes `key` — callers pass the raw input
2317    /// string. Persists to disk on the next `flush()` under the current
2318    /// flush policy (see [`Mnemo::set_cache_flush_policy`]); in
2319    /// [`CacheFlushPolicy::Batched`] the engine may call `flush()`
2320    /// internally when a threshold trips.
2321    ///
2322    /// On overflow of the namespace's [`CacheBudget`] (default 10,000
2323    /// entries / 64 MiB) the LRU entry is tombstoned before the new
2324    /// insert. Tombstones are reclaimed by
2325    /// [`Mnemo::compact_file`], not the runtime flush.
2326    ///
2327    /// Returns [`MnemoError::ReadOnly`] on a read-only handle.
2328    pub fn cache_put(
2329        &mut self,
2330        namespace: &str,
2331        key: &str,
2332        value: &[u8],
2333        opts: CachePutOpts,
2334    ) -> Result<()> {
2335        if self.read_only {
2336            return Err(MnemoError::ReadOnly);
2337        }
2338        let key_hash = result_cache::hash_key(key);
2339        let key_preview = result_cache::key_preview(key);
2340        let now = memory::now_secs();
2341        let size_bytes = value.len() as u64;
2342
2343        let entry = CacheEntry {
2344            namespace: namespace.to_string(),
2345            key_hash,
2346            key_preview: key_preview.clone(),
2347            value: value.to_vec(),
2348            content_type: opts.content_type.clone(),
2349            created_at: now,
2350            accessed_at: now,
2351            access_count: 0,
2352            ttl_secs: opts.ttl_secs,
2353            cost_hint_ms: opts.cost_hint_ms,
2354            size_bytes,
2355            vector: None,
2356            model: None,
2357        };
2358        let bytes = rmp_serde::to_vec(&entry).map_err(|e| MnemoError::Serialize(e.to_string()))?;
2359        let (start_page, page_count) = self.write_record(&bytes)?;
2360
2361        // Tombstone any prior entry for this key. Its pages remain on
2362        // disk until `compact_file`, exactly like memory-record
2363        // overwrites — the tombstone just hides it from lookups.
2364        if let Some(prev_pos) = self.cache_idx.get(namespace, &key_hash) {
2365            self.cache_dir[prev_pos].deleted = true;
2366            self.cache_idx.remove(namespace, &key_hash);
2367        }
2368
2369        let dir_entry = CacheDirectoryEntry {
2370            namespace: namespace.to_string(),
2371            key_hash,
2372            start_page,
2373            page_count,
2374            len: bytes.len() as u32,
2375            deleted: false,
2376            accessed_at: now,
2377            access_count: 0,
2378            size_bytes,
2379            ttl_secs: opts.ttl_secs,
2380            created_at: now,
2381            vector: None,
2382            model: None,
2383        };
2384        let pos = self.cache_dir.len();
2385        self.cache_dir.push(dir_entry);
2386        self.cache_idx.insert(namespace, key_hash, pos);
2387
2388        // Enforce budget — evict LRU tombstones until the namespace
2389        // fits under both caps.
2390        self.evict_to_budget(namespace)?;
2391
2392        self.dirty_cache = true;
2393        self.cache_batch.record_dirty();
2394
2395        // Batched policy: auto-flush if a threshold tripped. Successful
2396        // `flush()` calls `cache_batch.reset()` for us.
2397        if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
2398            self.flush()?;
2399        }
2400        Ok(())
2401    }
2402
2403    /// Look up `(namespace, key)`. Returns `None` on miss (either the
2404    /// key was never inserted or the entry is TTL-expired). On hit
2405    /// updates catalog-side access stats (mirrors the recall v5-trick);
2406    /// on a read-only handle the stats update is silently skipped.
2407    ///
2408    /// A miss is recorded as a miss in this handle's counters even if
2409    /// the key was TTL-expired — from the caller's perspective the two
2410    /// are indistinguishable.
2411    pub fn cache_get(&mut self, namespace: &str, key: &str) -> Result<Option<CachedValue>> {
2412        // FIX-1 R-001.8 part 4: under `Batched`, honor `max_age` on
2413        // reads too. Pre-FIX-1 the age check only fired on `cache_put`,
2414        // so a workload that wrote a few entries and then switched to
2415        // pure reads could sit with its pending batch past the
2416        // nominal max_age indefinitely. `max_dirty` is already
2417        // maintained per-write; this adds the symmetric time-bound
2418        // check on the read path. Only runs on read-write handles —
2419        // flush() returns ReadOnly for shared-lock readers.
2420        if !self.read_only && self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
2421            self.flush()?;
2422        }
2423        let key_hash = result_cache::hash_key(key);
2424        let pos = match self.cache_idx.get(namespace, &key_hash) {
2425            Some(p) => p,
2426            None => {
2427                self.cache_counters.misses += 1;
2428                return Ok(None);
2429            }
2430        };
2431        let now = memory::now_secs();
2432        // Snapshot the directory entry fields we need before any
2433        // mutation — the borrow checker gets grumpy otherwise.
2434        let (start_page, page_count, len, ttl_secs, expired, created_at) = {
2435            let e = &self.cache_dir[pos];
2436            (
2437                e.start_page,
2438                e.page_count,
2439                e.len,
2440                e.ttl_secs,
2441                e.is_expired(now),
2442                e.created_at,
2443            )
2444        };
2445        if expired {
2446            // Treat expired as a miss for the caller. Don't tombstone
2447            // here — cache_purge does that. Just skip.
2448            self.cache_counters.misses += 1;
2449            return Ok(None);
2450        }
2451        // Decrypt and decode the record body.
2452        let mut buf = Vec::with_capacity(len as usize);
2453        for i in 0..page_count as u64 {
2454            buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
2455        }
2456        buf.truncate(len as usize);
2457        let entry: CacheEntry =
2458            rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
2459
2460        // Bump access stats catalog-only (v5 trick — no full record
2461        // rewrite on a hit). Skipped on read-only handles so a shared-lock
2462        // reader can't dirty the catalog.
2463        if !self.read_only {
2464            self.cache_dir[pos].accessed_at = now;
2465            self.cache_dir[pos].access_count = self.cache_dir[pos].access_count.saturating_add(1);
2466            self.dirty_cache = true;
2467        }
2468        self.cache_counters.hits += 1;
2469        // On a mutating handle: access_count reflects the just-bumped
2470        // value. On a read-only handle: the on-disk directory value is
2471        // unchanged, so returning it verbatim is honest (the caller's
2472        // reads did not, and cannot, alter the persisted counter).
2473        //
2474        // FIX-1 R-002.2 part 1: `accessed_at` must also come from the
2475        // directory on a read-only handle, not from `entry.accessed_at`
2476        // in the record body. The record body is written once at insert
2477        // time and never updated on a hit — every live `accessed_at`
2478        // update lands in the directory (see the `!self.read_only`
2479        // branch above). The pre-FIX-1 code returned
2480        // `entry.accessed_at`, which is really the entry's **creation
2481        // time** by then, so a read-only reader saw "last accessed =
2482        // when it was inserted" even after many successful hits by
2483        // concurrent writers.
2484        let access_count = self.cache_dir[pos].access_count;
2485        let accessed_at = if self.read_only {
2486            self.cache_dir[pos].accessed_at
2487        } else {
2488            now
2489        };
2490        Ok(Some(CachedValue {
2491            value: entry.value,
2492            content_type: entry.content_type,
2493            created_at,
2494            accessed_at,
2495            access_count,
2496            ttl_secs,
2497        }))
2498    }
2499
2500    /// Tombstone the entry at `(namespace, key)`. Silently succeeds if
2501    /// no such entry exists. Space is reclaimed by `compact_file`.
2502    pub fn cache_delete(&mut self, namespace: &str, key: &str) -> Result<()> {
2503        if self.read_only {
2504            return Err(MnemoError::ReadOnly);
2505        }
2506        let key_hash = result_cache::hash_key(key);
2507        if let Some(pos) = self.cache_idx.get(namespace, &key_hash) {
2508            self.cache_dir[pos].deleted = true;
2509            self.cache_idx.remove(namespace, &key_hash);
2510            self.dirty_cache = true;
2511            self.cache_batch.record_dirty();
2512            if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
2513                self.flush()?;
2514            }
2515        }
2516        Ok(())
2517    }
2518
2519    /// Purge cache entries. If `namespace` is `Some`, restrict to that
2520    /// namespace; otherwise purge across the whole cache. If
2521    /// `expired_only` is `true`, only TTL-expired entries are removed;
2522    /// otherwise every live entry in the target scope is tombstoned.
2523    /// Returns the number of entries newly tombstoned.
2524    pub fn cache_purge(&mut self, namespace: Option<&str>, expired_only: bool) -> Result<usize> {
2525        if self.read_only {
2526            return Err(MnemoError::ReadOnly);
2527        }
2528        let now = memory::now_secs();
2529        let mut purged = 0usize;
2530        // Iterate the directory by index; mark matching entries as
2531        // deleted and pull them out of the in-memory index.
2532        for pos in 0..self.cache_dir.len() {
2533            if self.cache_dir[pos].deleted {
2534                continue;
2535            }
2536            if let Some(ns) = namespace {
2537                if self.cache_dir[pos].namespace != ns {
2538                    continue;
2539                }
2540            }
2541            if expired_only && !self.cache_dir[pos].is_expired(now) {
2542                continue;
2543            }
2544            let key_hash = self.cache_dir[pos].key_hash;
2545            let ns = self.cache_dir[pos].namespace.clone();
2546            self.cache_dir[pos].deleted = true;
2547            self.cache_idx.remove(&ns, &key_hash);
2548            purged += 1;
2549        }
2550        if purged > 0 {
2551            self.dirty_cache = true;
2552            self.cache_batch.record_dirty();
2553            if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
2554                self.flush()?;
2555            }
2556        }
2557        Ok(purged)
2558    }
2559
2560    /// Statistics for `namespace` (or the whole cache if `None`).
2561    /// `hits`/`misses`/`evictions` are per-handle in-memory counters
2562    /// that reset on reopen; `entries`/`bytes` reflect the durable
2563    /// on-disk state (live entries only, tombstones excluded).
2564    pub fn cache_stats(&self, namespace: Option<&str>) -> CacheStats {
2565        let mut entries = 0usize;
2566        let mut bytes = 0u64;
2567        for e in &self.cache_dir {
2568            if e.deleted {
2569                continue;
2570            }
2571            if let Some(ns) = namespace {
2572                if e.namespace != ns {
2573                    continue;
2574                }
2575            }
2576            entries += 1;
2577            bytes += e.size_bytes;
2578        }
2579        let hits = self.cache_counters.hits;
2580        let misses = self.cache_counters.misses;
2581        let total = hits + misses;
2582        let hit_rate = if total > 0 {
2583            hits as f64 / total as f64
2584        } else {
2585            0.0
2586        };
2587        CacheStats {
2588            entries,
2589            bytes,
2590            hits,
2591            misses,
2592            hit_rate,
2593            evictions: self.cache_counters.evictions,
2594        }
2595    }
2596
2597    /// Set (or update) the eviction budget for `namespace`. Applies to
2598    /// subsequent `cache_put` calls; does not retroactively evict
2599    /// entries that already sit under a looser cap.
2600    pub fn set_cache_budget(&mut self, namespace: &str, budget: CacheBudget) {
2601        self.cache_budgets.insert(namespace.to_string(), budget);
2602    }
2603
2604    /// Set the per-handle cache flush policy. Defaults to
2605    /// [`CacheFlushPolicy::Strict`]. Memory writes are never batched;
2606    /// this affects only cache mutations.
2607    pub fn set_cache_flush_policy(&mut self, policy: CacheFlushPolicy) {
2608        self.cache_flush_policy = policy;
2609        // Switching mid-run: if we're moving to Strict and the batch is
2610        // non-empty, the pending count still counts toward dirty_cache
2611        // which the next flush() will consume.
2612    }
2613
2614    /// Evict LRU live entries in `namespace` until it fits under both
2615    /// budget caps. Called from `cache_put` after the insert; the new
2616    /// entry is included in the count so we always leave room.
2617    fn evict_to_budget(&mut self, namespace: &str) -> Result<()> {
2618        let budget = self
2619            .cache_budgets
2620            .get(namespace)
2621            .copied()
2622            .unwrap_or_default();
2623        loop {
2624            // Sum current live count + bytes in the namespace.
2625            let mut live_count = 0usize;
2626            let mut live_bytes = 0u64;
2627            for e in &self.cache_dir {
2628                if !e.deleted && e.namespace == namespace {
2629                    live_count += 1;
2630                    live_bytes += e.size_bytes;
2631                }
2632            }
2633            if live_count <= budget.max_entries && live_bytes <= budget.max_bytes {
2634                return Ok(());
2635            }
2636            // Find the LRU (oldest accessed_at) live entry.
2637            let victim_pos = self
2638                .cache_dir
2639                .iter()
2640                .enumerate()
2641                .filter(|(_, e)| !e.deleted && e.namespace == namespace)
2642                .min_by_key(|(_, e)| e.accessed_at)
2643                .map(|(i, _)| i);
2644            let Some(pos) = victim_pos else {
2645                // Nothing to evict but still over budget — the single
2646                // just-inserted entry itself is bigger than
2647                // max_bytes. Nothing we can safely do here without
2648                // rejecting the put; leave it and return.
2649                return Ok(());
2650            };
2651            let key_hash = self.cache_dir[pos].key_hash;
2652            let ns = self.cache_dir[pos].namespace.clone();
2653            self.cache_dir[pos].deleted = true;
2654            self.cache_idx.remove(&ns, &key_hash);
2655            self.cache_counters.evictions += 1;
2656        }
2657    }
2658
2659    // === Semantic cache (Phase 10.2) ==================================
2660
2661    /// Insert or overwrite a **semantic** cache entry — same as
2662    /// [`Mnemo::cache_put`] but with a required embedding vector and
2663    /// model identifier. The vector is stored on the directory entry
2664    /// so [`Mnemo::cache_get_semantic`] can score without decrypting
2665    /// record bodies.
2666    ///
2667    /// Vector dimensionality is validated against the database's
2668    /// configured `dimensions` — a mismatch returns
2669    /// [`MnemoError::DimensionMismatch`], the same signal `remember`
2670    /// gives for the memory catalog.
2671    ///
2672    /// The `model` string is matched **exactly** on lookup — a
2673    /// different model's cache is invisible to a given query. Pick
2674    /// canonical identifiers (`"text-embedding-3-small"`,
2675    /// `"bge-large-en-v1.5"`, ...) and stick to them.
2676    pub fn cache_put_semantic(
2677        &mut self,
2678        namespace: &str,
2679        key: &str,
2680        vector: Vec<f32>,
2681        value: &[u8],
2682        opts: SemanticCachePutOpts,
2683    ) -> Result<()> {
2684        if self.read_only {
2685            return Err(MnemoError::ReadOnly);
2686        }
2687        if vector.len() != self.dimensions {
2688            return Err(MnemoError::DimensionMismatch {
2689                expected: self.dimensions,
2690                got: vector.len(),
2691            });
2692        }
2693        let key_hash = result_cache::hash_key(key);
2694        let key_preview = result_cache::key_preview(key);
2695        let now = memory::now_secs();
2696        let size_bytes = value.len() as u64;
2697
2698        let entry = CacheEntry {
2699            namespace: namespace.to_string(),
2700            key_hash,
2701            key_preview: key_preview.clone(),
2702            value: value.to_vec(),
2703            content_type: opts.content_type.clone(),
2704            created_at: now,
2705            accessed_at: now,
2706            access_count: 0,
2707            ttl_secs: opts.ttl_secs,
2708            cost_hint_ms: opts.cost_hint_ms,
2709            size_bytes,
2710            vector: Some(vector.clone()),
2711            model: Some(opts.model.clone()),
2712        };
2713        let bytes = rmp_serde::to_vec(&entry).map_err(|e| MnemoError::Serialize(e.to_string()))?;
2714        let (start_page, page_count) = self.write_record(&bytes)?;
2715
2716        if let Some(prev_pos) = self.cache_idx.get(namespace, &key_hash) {
2717            self.cache_dir[prev_pos].deleted = true;
2718            self.cache_idx.remove(namespace, &key_hash);
2719        }
2720
2721        let dir_entry = CacheDirectoryEntry {
2722            namespace: namespace.to_string(),
2723            key_hash,
2724            start_page,
2725            page_count,
2726            len: bytes.len() as u32,
2727            deleted: false,
2728            accessed_at: now,
2729            access_count: 0,
2730            size_bytes,
2731            ttl_secs: opts.ttl_secs,
2732            created_at: now,
2733            vector: Some(vector),
2734            model: Some(opts.model),
2735        };
2736        let pos = self.cache_dir.len();
2737        self.cache_dir.push(dir_entry);
2738        self.cache_idx.insert(namespace, key_hash, pos);
2739
2740        self.evict_to_budget(namespace)?;
2741
2742        self.dirty_cache = true;
2743        self.cache_batch.record_dirty();
2744        if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
2745            self.flush()?;
2746        }
2747        Ok(())
2748    }
2749
2750    /// Semantic recall over the cache: score every live entry in
2751    /// `namespace` whose `model` matches against `query_vector` using
2752    /// cosine similarity; return the top-1 hit and its similarity iff
2753    /// `sim >= threshold`. `None` otherwise.
2754    ///
2755    /// Uses [`crate::result_cache::DEFAULT_SEMANTIC_THRESHOLD`] if you
2756    /// want a sensible default — pass it explicitly (`0.97` today).
2757    ///
2758    /// Exact-key entries (those without a vector) are transparently
2759    /// skipped, as are entries whose model doesn't match — a
2760    /// different-model hit would be a semantic bug, not a win.
2761    /// TTL-expired entries are also skipped.
2762    ///
2763    /// Access-stat updates on hit follow the same policy as
2764    /// [`Mnemo::cache_get`]: bumped on read-write handles, silently
2765    /// skipped on read-only. Hits and misses are recorded in the
2766    /// per-handle counters just like exact-key lookups.
2767    pub fn cache_get_semantic(
2768        &mut self,
2769        namespace: &str,
2770        query_vector: &[f32],
2771        threshold: f32,
2772        model: &str,
2773    ) -> Result<Option<(CachedValue, f32)>> {
2774        if query_vector.len() != self.dimensions {
2775            return Err(MnemoError::DimensionMismatch {
2776                expected: self.dimensions,
2777                got: query_vector.len(),
2778            });
2779        }
2780        // FIX-1 R-001.8 part 4: same Batched max_age check as
2781        // `cache_get`. See the comment there.
2782        if !self.read_only && self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
2783            self.flush()?;
2784        }
2785        let now = memory::now_secs();
2786
2787        // Scan namespace live entries, filter by model, compute
2788        // cosine, keep the best. Exact-key entries have `vector: None`
2789        // and are skipped by the filter.
2790        let mut best: Option<(usize, f32)> = None;
2791        for (i, e) in self.cache_dir.iter().enumerate() {
2792            if e.deleted || e.namespace != namespace || e.is_expired(now) {
2793                continue;
2794            }
2795            let Some(ref v) = e.vector else { continue };
2796            let Some(ref m) = e.model else { continue };
2797            if m != model {
2798                continue;
2799            }
2800            let sim = memory::cosine(query_vector, v);
2801            if best.map(|(_, s)| sim > s).unwrap_or(true) {
2802                best = Some((i, sim));
2803            }
2804        }
2805
2806        let Some((pos, sim)) = best else {
2807            self.cache_counters.misses += 1;
2808            return Ok(None);
2809        };
2810        if sim < threshold {
2811            self.cache_counters.misses += 1;
2812            return Ok(None);
2813        }
2814
2815        // Extract fields before any mutation to keep the borrow checker
2816        // happy.
2817        let (start_page, page_count, len, ttl_secs, created_at) = {
2818            let e = &self.cache_dir[pos];
2819            (e.start_page, e.page_count, e.len, e.ttl_secs, e.created_at)
2820        };
2821
2822        // Decrypt and decode the record body.
2823        let mut buf = Vec::with_capacity(len as usize);
2824        for i in 0..page_count as u64 {
2825            buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
2826        }
2827        buf.truncate(len as usize);
2828        let entry: CacheEntry =
2829            rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
2830
2831        if !self.read_only {
2832            self.cache_dir[pos].accessed_at = now;
2833            self.cache_dir[pos].access_count = self.cache_dir[pos].access_count.saturating_add(1);
2834            self.dirty_cache = true;
2835        }
2836        self.cache_counters.hits += 1;
2837        let access_count = self.cache_dir[pos].access_count;
2838        // FIX-1 R-002.2 part 1: same read-only `accessed_at` fix as
2839        // `cache_get`. The directory carries the live value; the record
2840        // body only holds the insert-time snapshot.
2841        let accessed_at = if self.read_only {
2842            self.cache_dir[pos].accessed_at
2843        } else {
2844            now
2845        };
2846        Ok(Some((
2847            CachedValue {
2848                value: entry.value,
2849                content_type: entry.content_type,
2850                created_at,
2851                accessed_at,
2852                access_count,
2853                ttl_secs,
2854            },
2855            sim,
2856        )))
2857    }
2858
2859    /// Begin a conversation [`Session`](crate::Session) for `agent_id`.
2860    ///
2861    /// The session borrows the database for its lifetime, records turns as
2862    /// working memory, and consolidates them into episodic memory when closed.
2863    pub fn session(&mut self, agent_id: impl Into<String>) -> crate::session::Session<'_> {
2864        crate::session::Session::new(self, agent_id.into())
2865    }
2866
2867    // --- snapshots & point-in-time recovery ------------------------------
2868
2869    /// Every committed transaction, oldest first — the restore points
2870    /// available to [`Mnemo::restore_to`] and [`Mnemo::restore_to_time`].
2871    ///
2872    /// Each `flush` appends one snapshot. Because the storage engine is
2873    /// append-only, the pages a past flush wrote are still on disk, so any
2874    /// listed snapshot can be reinstated exactly. The history reaches back to
2875    /// the last [`Mnemo::compact_file`], which reclaims space by collapsing
2876    /// it.
2877    pub fn snapshots(&self) -> Vec<SnapshotInfo> {
2878        let mut v: Vec<SnapshotInfo> = self
2879            .manifest
2880            .iter()
2881            .map(|s| SnapshotInfo {
2882                txn_id: s.txn_id,
2883                created_at: s.created_at,
2884                memory_count: s.memory_count,
2885            })
2886            .collect();
2887        v.sort_by_key(|s| s.txn_id);
2888        v
2889    }
2890
2891    /// Load a past snapshot's state and commit it as a new transaction.
2892    fn apply_snapshot(&mut self, snap: &Snapshot) -> Result<()> {
2893        let catalog = load_catalog(
2894            &mut self.pager,
2895            snap.catalog_start,
2896            snap.catalog_pages,
2897            snap.catalog_len,
2898        )?;
2899        let ann = load_index(
2900            &mut self.pager,
2901            snap.index_start,
2902            snap.index_pages,
2903            snap.index_len,
2904            self.dimensions,
2905        )?;
2906        self.index = build_id_index(&catalog);
2907        self.catalog = catalog;
2908        self.ann = ann;
2909        // Re-commit as a fresh transaction: crash-safe, and itself recorded
2910        // as a new snapshot so a restore can always be undone.
2911        self.dirty_catalog = true;
2912        self.dirty_index = true;
2913        self.flush()
2914    }
2915
2916    /// Restore the database to the snapshot produced by transaction `txn_id`.
2917    ///
2918    /// The restore is itself a new committed transaction (and a new
2919    /// snapshot), so it is crash-safe and reversible — restoring forward to a
2920    /// later snapshot afterwards works just as well.
2921    pub fn restore_to(&mut self, txn_id: u64) -> Result<SnapshotInfo> {
2922        if self.read_only {
2923            return Err(MnemoError::ReadOnly);
2924        }
2925        let snap = self
2926            .manifest
2927            .iter()
2928            .find(|s| s.txn_id == txn_id)
2929            .cloned()
2930            .ok_or_else(|| MnemoError::NotFound(format!("snapshot for transaction {txn_id}")))?;
2931        self.apply_snapshot(&snap)?;
2932        Ok(SnapshotInfo {
2933            txn_id: snap.txn_id,
2934            created_at: snap.created_at,
2935            memory_count: snap.memory_count,
2936        })
2937    }
2938
2939    /// Restore the database to the latest snapshot committed at or before
2940    /// `unix_secs`. Returns [`MnemoError::NotFound`] if no snapshot is that
2941    /// old. Like [`Mnemo::restore_to`], the restore is a new transaction.
2942    pub fn restore_to_time(&mut self, unix_secs: i64) -> Result<SnapshotInfo> {
2943        if self.read_only {
2944            return Err(MnemoError::ReadOnly);
2945        }
2946        let snap = self
2947            .manifest
2948            .iter()
2949            .filter(|s| s.created_at <= unix_secs)
2950            .max_by_key(|s| s.txn_id)
2951            .cloned()
2952            .ok_or_else(|| {
2953                MnemoError::NotFound(format!("no snapshot at or before time {unix_secs}"))
2954            })?;
2955        self.apply_snapshot(&snap)?;
2956        Ok(SnapshotInfo {
2957            txn_id: snap.txn_id,
2958            created_at: snap.created_at,
2959            memory_count: snap.memory_count,
2960        })
2961    }
2962
2963    /// Return the KDF (Argon2id) parameters this file currently uses to
2964    /// wrap its DEK. Needed by callers that want to [`rekey`] while
2965    /// keeping the file's original protection level (e.g. the Python
2966    /// binding's `rekey(fast=False)`, whose docstring promises exactly
2967    /// that). FIX-1 R-001.5.
2968    pub fn kdf_params(&self) -> KdfParams {
2969        self.kdf
2970    }
2971
2972    /// Flush any pending writes, then stream the (encrypted) file to
2973    /// `dest` atomically: writes to a sibling temp file in `dest`'s
2974    /// parent directory, `sync_all`s it, renames onto `dest`, and
2975    /// fsyncs the parent directory on Unix so a crash never leaves a
2976    /// partial export.
2977    ///
2978    /// Refuses when `dest` resolves to the file this handle has open
2979    /// — that case would otherwise `File::create`-truncate the live
2980    /// database to 0 bytes (`fs::copy` on Unix reads *after* the
2981    /// `O_TRUNC`, so a naive export-to-self destroys the source).
2982    /// Unix: compares `(dev, ino)` of both paths via `MetadataExt`,
2983    /// catching symlinks and hard links. Windows: compares
2984    /// canonicalized paths; a rename onto an open file also fails on
2985    /// Windows because of the mandatory lock the handle holds, so the
2986    /// rename step is a second line of defense.
2987    ///
2988    /// Reads the bytes from the already-open file handle, not by
2989    /// `fs::copy`. On Windows that matters: `LockFileEx` locks are
2990    /// mandatory, so a second handle opened by `fs::copy` would fail
2991    /// while this writer still holds the lock.
2992    ///
2993    /// FIX-1 R-001.5 + R-002.2 part 3.
2994    pub fn export_encrypted(&mut self, dest: impl AsRef<Path>) -> Result<()> {
2995        let dest = dest.as_ref();
2996
2997        // Step 1 — refuse export to the open file itself. We compare
2998        // by identity, not string, so symlinks and hard links to
2999        // `self.path` are caught too. The check runs before any flush
3000        // so a self-targeted export is a pure no-op.
3001        if paths_refer_to_same_file(&self.path, dest).unwrap_or(false) {
3002            return Err(MnemoError::Invalid(format!(
3003                "export_encrypted refuses to overwrite the open database at {}",
3004                dest.display()
3005            )));
3006        }
3007
3008        // Step 2 — flush, so on-disk bytes match the live state.
3009        if !self.read_only {
3010            self.flush()?;
3011        }
3012
3013        // Step 3 — stream from the open handle into a sibling temp.
3014        // Using the open handle (instead of a second open on `self.path`)
3015        // means a Windows mandatory lock can't block the copy.
3016        let parent = dest.parent().ok_or_else(|| {
3017            MnemoError::Invalid(format!(
3018                "export_encrypted: dest {} has no parent directory",
3019                dest.display()
3020            ))
3021        })?;
3022        let tmp = temp_sibling(dest);
3023        // Scope the temp handle so it closes before the rename on
3024        // Windows (which refuses to rename over an open file).
3025        {
3026            let src = self.pager.file_mut();
3027            src.seek(SeekFrom::Start(0))?;
3028            let mut out = OpenOptions::new()
3029                .write(true)
3030                .create(true)
3031                .truncate(true)
3032                .open(&tmp)?;
3033            let mut buf = vec![0u8; 1 << 20]; // 1 MiB
3034            loop {
3035                let n = src.read(&mut buf)?;
3036                if n == 0 {
3037                    break;
3038                }
3039                out.write_all(&buf[..n])?;
3040            }
3041            out.sync_all()?;
3042            crate::bump_fsync();
3043        }
3044
3045        // Step 4 — atomic rename onto `dest`.
3046        if let Err(e) = std::fs::rename(&tmp, dest) {
3047            // Best-effort cleanup; propagate the rename error.
3048            let _ = std::fs::remove_file(&tmp);
3049            return Err(MnemoError::Io(e));
3050        }
3051
3052        // Step 5 — fsync the parent directory on Unix so the rename
3053        // itself is durable. On Windows directory syncs aren't
3054        // supported; the rename above is atomic enough on NTFS.
3055        #[cfg(unix)]
3056        {
3057            if let Ok(dir) = File::open(parent) {
3058                let _ = dir.sync_all();
3059                crate::bump_fsync();
3060            }
3061        }
3062        #[cfg(not(unix))]
3063        {
3064            let _ = parent;
3065        }
3066
3067        Ok(())
3068    }
3069
3070    /// Change the passphrase. Cheap: re-derives the KEK and re-wraps the DEK;
3071    /// the encrypted pages are never rewritten.
3072    pub fn rekey(&mut self, new_passphrase: &str, kdf: KdfParams) -> Result<()> {
3073        if self.read_only {
3074            return Err(MnemoError::ReadOnly);
3075        }
3076        self.flush()?;
3077        let new_salt = crypto::random_salt();
3078        let new_kek = crypto::derive_kek(new_passphrase.as_bytes(), &new_salt, kdf)?;
3079        let new_nonce = crypto::random_nonce();
3080        let wrapped = crypto::wrap_dek(&new_kek, &new_nonce, self.pager.dek())?;
3081        if wrapped.len() != WRAPPED_DEK_LEN {
3082            return Err(MnemoError::Crypto("unexpected wrapped DEK length".into()));
3083        }
3084        let mut wrapped_dek = [0u8; WRAPPED_DEK_LEN];
3085        wrapped_dek.copy_from_slice(&wrapped);
3086
3087        self.header.salt = new_salt;
3088        self.header.dek_nonce = new_nonce;
3089        self.header.wrapped_dek = wrapped_dek;
3090        self.header.m_cost = kdf.m_cost;
3091        self.header.t_cost = kdf.t_cost;
3092        self.header.p_cost = kdf.p_cost;
3093        self.kdf = kdf;
3094
3095        // DUR-1 (R-002.1): WAL-log the new page 0 before touching it.
3096        // Allocate a fresh WAL region at the tail so that a crash during
3097        // the rekey leaves either the old region (old passphrase works)
3098        // or the new region (new passphrase works) fully committed —
3099        // overwriting the existing region would risk destroying both.
3100        let old_wal_start = self.header.wal_start;
3101        let new_pages = self.header.wal_pages;
3102        let new_start = self.header.next_page;
3103        self.header.next_page += new_pages;
3104        self.header.wal_start = new_start;
3105        self.header.wal_pages = new_pages;
3106        self.header.wal_seq += 1;
3107        self.relocate_header_through_wal(old_wal_start)
3108    }
3109
3110    /// Decrypt and validate every live record. Returns the count verified
3111    /// across memories **and** cache entries (FIX-1 R-001.1: now that the
3112    /// cache survives compaction, `verify()` has to walk it too — the
3113    /// pre-FIX-1 implementation only checked `self.catalog`, so a
3114    /// corrupted cache record went unnoticed).
3115    pub fn verify(&mut self) -> Result<usize> {
3116        let entries: Vec<CatalogEntry> = self
3117            .catalog
3118            .iter()
3119            .filter(|e| !e.deleted)
3120            .cloned()
3121            .collect();
3122        let mut n = entries.len();
3123        for e in &entries {
3124            let bytes = self.read_record(e)?;
3125            let _m: Memory = rmp_serde::from_slice(&bytes)
3126                .map_err(|err| MnemoError::Serialize(err.to_string()))?;
3127        }
3128
3129        // Also decode every live cache entry. Tombstoned entries are
3130        // skipped — their pages are orphan ciphertext awaiting
3131        // `compact_file`, same as deleted memories. Expired entries
3132        // are *still* on disk until compaction, so we verify them too.
3133        let cache_positions: Vec<(u64, u32, u32)> = self
3134            .cache_dir
3135            .iter()
3136            .filter(|e| !e.deleted)
3137            .map(|e| (e.start_page, e.page_count, e.len))
3138            .collect();
3139        for (start_page, page_count, len) in cache_positions {
3140            let mut buf = Vec::with_capacity(len as usize);
3141            for i in 0..page_count as u64 {
3142                buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
3143            }
3144            buf.truncate(len as usize);
3145            let _c: CacheEntry = rmp_serde::from_slice(&buf)
3146                .map_err(|err| MnemoError::Serialize(err.to_string()))?;
3147            n += 1;
3148        }
3149        Ok(n)
3150    }
3151
3152    /// Summary statistics for the open database.
3153    pub fn stats(&mut self) -> Result<Stats> {
3154        let deleted = self.catalog.iter().filter(|e| e.deleted).count();
3155        // FIX-1 R-001.8 part 2: build the agent set lazily on first
3156        // call per handle, then maintain in `put` / invalidate in
3157        // `delete`. Pre-FIX-1 every `stats()` call decrypted every
3158        // live memory (O(N) AES-GCM decrypts just to list a usually-
3159        // tiny set of distinct agent IDs). On a 1M-memory database
3160        // this was in the multi-second range on hardware AES.
3161        if self.agent_set.is_none() {
3162            let built: std::collections::BTreeSet<String> =
3163                self.memories()?.into_iter().map(|m| m.agent_id).collect();
3164            self.agent_set = Some(built);
3165        }
3166        let agents: Vec<String> = self
3167            .agent_set
3168            .as_ref()
3169            .expect("agent_set just filled above")
3170            .iter()
3171            .cloned()
3172            .collect();
3173        let file_bytes = self.header.next_page * PAGE_SIZE as u64;
3174        Ok(Stats {
3175            memories: self.len(),
3176            deleted,
3177            dimensions: self.dimensions,
3178            file_bytes,
3179            agents,
3180            encrypted: self.header.flags & FLAG_ENCRYPTED != 0,
3181            created_at: self.header.created_at,
3182            index: self.ann.as_ref().map(|a| a.info()),
3183            wal_pages: self.header.wal_pages,
3184            fsyncs_last_flush: self.fsyncs_last_flush,
3185        })
3186    }
3187
3188    /// Rewrite the file, dropping tombstoned and expired memories and
3189    /// reclaiming stale pages left by updates. Done out-of-place via a temp
3190    /// file and an atomic rename.
3191    ///
3192    /// FIX-1 R-001.1 + R-002.2 part 2: the result cache now survives
3193    /// compaction. Every live, non-expired directory entry is copied
3194    /// into the new file verbatim — `created_at`, `ttl_secs`,
3195    /// `cost_hint_ms`, `key_preview`, `size_bytes`, `content_type`,
3196    /// `vector` and `model` come from the record body; `accessed_at`
3197    /// and `access_count` come from the *directory* (the record body
3198    /// only holds the insert-time snapshot — the directory is the
3199    /// source of truth for the live counts, as of v5's recall trick).
3200    /// Tombstoned and TTL-expired entries are dropped. Pre-FIX-1
3201    /// compaction kept only the memory catalog and silently emptied
3202    /// the cache; the Planner's interim README caveat (c417d2e) called
3203    /// that out — this PR removes the surprise.
3204    pub fn compact_file(path: &str, passphrase: &str) -> Result<CompactReport> {
3205        let mut old = Mnemo::open(path, passphrase)?;
3206        let dims = old.dimensions;
3207        let kdf = old.kdf;
3208        let tmp = format!("{path}.compact-tmp");
3209
3210        let mut new = Mnemo::create(
3211            &tmp,
3212            passphrase,
3213            MnemoConfig {
3214                dimensions: dims,
3215                kdf,
3216                ..Default::default()
3217            },
3218        )?;
3219        // ANN-1.1: compaction is "re-tune for the current size" by
3220        // default. Pass IndexConfig::default() so partitions,
3221        // subspaces, n_probe and n_rerank are all re-derived from the
3222        // compacted vector count. Previously we froze the pre-compact
3223        // (n_probe, n_rerank), which left the new index on stale
3224        // numbers for the smaller post-compact corpus.
3225        let want_index = old.ann.is_some();
3226        let now = memory::now_secs();
3227        let all = old.memories()?;
3228        let before = all.len();
3229        let mut after = 0;
3230        for m in all {
3231            if m.is_expired(now) {
3232                continue;
3233            }
3234            new.remember(m)?;
3235            after += 1;
3236        }
3237        if want_index && after > 0 {
3238            new.build_index_with(IndexConfig::default())?;
3239        }
3240
3241        // Count + copy cache entries. We walk `old.cache_dir` by index
3242        // so each hop reads the directory's live `accessed_at` /
3243        // `access_count` directly rather than going through
3244        // `cache_get` (which would also treat read-only semantics,
3245        // tombstones, etc. — we want the raw, authoritative values).
3246        let mut cache_before = 0usize;
3247        let mut cache_after = 0usize;
3248        let positions: Vec<usize> = (0..old.cache_dir.len()).collect();
3249        for pos in positions {
3250            let keep = {
3251                let e = &old.cache_dir[pos];
3252                !e.deleted && !e.is_expired(now)
3253            };
3254            if keep {
3255                cache_before += 1;
3256                new.copy_cache_entry_from(&mut old, pos)?;
3257                cache_after += 1;
3258            }
3259        }
3260
3261        new.flush()?;
3262        drop(new);
3263        drop(old);
3264
3265        std::fs::rename(&tmp, path)?;
3266        Ok(CompactReport {
3267            before,
3268            after,
3269            cache_before,
3270            cache_after,
3271        })
3272    }
3273
3274    /// Copy one live cache entry from `old` at directory position
3275    /// `pos` into `self` (a freshly-created compaction target).
3276    /// Decrypts the old record body, substitutes the old *directory*'s
3277    /// `accessed_at` + `access_count` (the record body's copies are
3278    /// stale as of v5's directory-only hit path), re-encodes, writes
3279    /// a fresh record run, and pushes a matching directory entry.
3280    ///
3281    /// Caller is expected to have already filtered out tombstoned /
3282    /// expired entries and to call this on a read-write handle.
3283    pub(crate) fn copy_cache_entry_from(&mut self, old: &mut Mnemo, pos: usize) -> Result<()> {
3284        if self.read_only {
3285            return Err(MnemoError::ReadOnly);
3286        }
3287        // Snapshot the directory fields we need, so the borrow on
3288        // `old.cache_dir` ends before we touch `old.pager`.
3289        let (start_page, page_count, len, dir_accessed_at, dir_access_count) = {
3290            let e = &old.cache_dir[pos];
3291            (
3292                e.start_page,
3293                e.page_count,
3294                e.len,
3295                e.accessed_at,
3296                e.access_count,
3297            )
3298        };
3299
3300        // Read + decrypt the old record body.
3301        let mut buf = Vec::with_capacity(len as usize);
3302        for i in 0..page_count as u64 {
3303            buf.extend_from_slice(&old.pager.read_page(start_page + i)?);
3304        }
3305        buf.truncate(len as usize);
3306        let mut entry: CacheEntry =
3307            rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
3308
3309        // Overwrite the stale body copies with the directory's live
3310        // values before persisting. Everything else — `created_at`,
3311        // `ttl_secs`, `cost_hint_ms`, `key_preview`, `size_bytes`,
3312        // `content_type`, `value`, `vector`, `model` — stays as the
3313        // body already recorded it.
3314        entry.accessed_at = dir_accessed_at;
3315        entry.access_count = dir_access_count;
3316
3317        let bytes =
3318            rmp_serde::to_vec(&entry).map_err(|err| MnemoError::Serialize(err.to_string()))?;
3319        let (new_start_page, new_page_count) = self.write_record(&bytes)?;
3320
3321        let dir_entry = CacheDirectoryEntry {
3322            namespace: entry.namespace.clone(),
3323            key_hash: entry.key_hash,
3324            start_page: new_start_page,
3325            page_count: new_page_count,
3326            len: bytes.len() as u32,
3327            deleted: false,
3328            accessed_at: dir_accessed_at,
3329            access_count: dir_access_count,
3330            size_bytes: entry.size_bytes,
3331            ttl_secs: entry.ttl_secs,
3332            created_at: entry.created_at,
3333            vector: entry.vector.clone(),
3334            model: entry.model.clone(),
3335        };
3336        let new_pos = self.cache_dir.len();
3337        self.cache_idx
3338            .insert(&entry.namespace, entry.key_hash, new_pos);
3339        self.cache_dir.push(dir_entry);
3340        self.dirty_cache = true;
3341        Ok(())
3342    }
3343}