Skip to main content

mnemo/
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::error::{MnemoError, Result};
35use crate::format::{Header, FLAG_ENCRYPTED, PAGE_SIZE, PAYLOAD, VERSION, WRAPPED_DEK_LEN};
36use crate::index::{IndexConfig, IndexInfo, IvfPqIndex};
37use crate::memory::{self, Memory, MemoryType, Metric, Scope, ScoreWeights};
38use crate::pager::Pager;
39use crate::result_cache::{
40    self, BatchState, CacheBudget, CacheDirectoryEntry, CacheEntry, CacheFlushPolicy, CachePutOpts,
41    CacheStats, CachedValue, CounterState, DirectoryIndex, SemanticCachePutOpts,
42};
43use crate::wal;
44
45/// Default initial size of the write-ahead log region, in pages (64 KiB).
46/// The region grows automatically when a transaction's control plane outgrows
47/// it, so this is a *floor* on the fresh-file footprint, not a cap on
48/// transaction size. Was 64 pages (512 KiB) in v0.1.0 — lowered because that
49/// reservation dominated small-file size (~62% of a 31-memory dogfood file).
50const DEFAULT_WAL_PAGES: u64 = 8;
51
52/// Hard floor on the WAL reservation. A single header-page flush already
53/// needs one full DATA frame plus a COMMIT frame; two pages are the bare
54/// minimum that lets the first transaction commit without immediate growth.
55const MIN_WAL_PAGES: u64 = 2;
56
57/// Default cap on retained snapshot manifest entries. Each `flush` appends
58/// one entry; without a cap, long-running processes accumulate entries
59/// forever, growing per-flush manifest-serialize cost as O(total flushes)
60/// and the manifest run size on disk in lockstep. 256 keeps roughly a
61/// week of hourly snapshots while staying tiny (256 × ~82 bytes ≈ 21 KiB
62/// of manifest). Set `MnemoConfig::max_snapshots = 0` to disable the cap.
63pub const DEFAULT_MAX_SNAPSHOTS: usize = 256;
64
65/// Upper bound for the WAL scan when recovering a torn-header file. The
66/// real WAL region size lives in the header, but a torn header is exactly
67/// the case where we can't read it — so we scan a generous fixed window
68/// from the conventional WAL start, relying on `wal::recover` stopping at
69/// the first non-magic frame. 64 covers the historical pre-0.2 default
70/// of a 64-page reservation; users who configure larger WALs and then
71/// suffer a torn header at the same time will need offline recovery.
72const TORN_HEADER_SCAN_PAGES: u64 = 64;
73
74/// Configuration for creating a new database.
75#[derive(Clone, Copy, Debug)]
76pub struct MnemoConfig {
77    /// Embedding dimensionality. Every stored vector must match this.
78    pub dimensions: usize,
79    /// Key-derivation parameters.
80    pub kdf: KdfParams,
81    /// Initial size of the WAL region in 8 KiB pages. Default is
82    /// [`DEFAULT_WAL_PAGES`] (8 pages = 64 KiB), which keeps fresh files
83    /// small; the region auto-grows when a transaction's control plane needs
84    /// more space, so raising this is only a hint for the expected steady-state
85    /// transaction size. Write-heavy databases that routinely flush large
86    /// catalogs or ANN indexes can set this higher (e.g. 64 = 512 KiB) to
87    /// avoid early grow events; tiny embedded uses can leave it at the
88    /// default. Clamped to a minimum of [`MIN_WAL_PAGES`] (2 pages).
89    pub wal_pages_initial: u64,
90    /// Maximum number of snapshot manifest entries to retain. Each `flush`
91    /// appends one entry; once the cap is hit, the *oldest* entry is dropped
92    /// from the in-memory manifest before the new one is added, so the
93    /// retained set is always the most-recent N. Defaults to
94    /// [`DEFAULT_MAX_SNAPSHOTS`] (256). Set to `0` to disable the cap and
95    /// retain every snapshot forever (the pre-v0.3 behavior).
96    ///
97    /// Pages referenced *only* by pruned snapshots stay on disk until the
98    /// next [`Mnemo::compact_file`], which reclaims them. Point-in-time
99    /// recovery via [`Mnemo::restore_to`] into a pruned `txn_id` returns
100    /// [`MnemoError::NotFound`].
101    ///
102    /// Apply to an already-open database with [`Mnemo::set_max_snapshots`].
103    pub max_snapshots: usize,
104}
105
106impl Default for MnemoConfig {
107    fn default() -> Self {
108        Self {
109            dimensions: 768,
110            kdf: KdfParams::secure(),
111            wal_pages_initial: DEFAULT_WAL_PAGES,
112            max_snapshots: DEFAULT_MAX_SNAPSHOTS,
113        }
114    }
115}
116
117/// Catalog entry: maps a memory ID to its page run on disk.
118///
119/// **Schema (v5+):** carries `accessed_at` and `access_count` so `recall`
120/// can update access stats by mutating the catalog entry rather than
121/// rewriting the entire record (Phase 2.1 of the improvement plan). The
122/// values in a memory's serialized record body become a stale snapshot
123/// after the first recall; [`Mnemo::read_memory`] re-populates them from
124/// the catalog so consumers always see the live values.
125#[derive(Serialize, Deserialize, Clone, Debug)]
126struct CatalogEntry {
127    /// ULID as a raw `u128`.
128    id: u128,
129    start_page: u64,
130    page_count: u32,
131    /// Exact serialized byte length of the record.
132    len: u32,
133    deleted: bool,
134    /// Unix seconds of the most recent recall hit (or write, for fresh entries).
135    accessed_at: i64,
136    /// How many times `recall` has surfaced this memory.
137    access_count: u32,
138}
139
140/// Frozen v4 catalog entry shape, used only by the v4→v5 migration path in
141/// [`Mnemo::open`]. Its layout matches the catalog encoding written by
142/// pre-v5 builds (5 positional fields). Do not change this struct.
143#[derive(Serialize, Deserialize, Clone, Debug)]
144struct CatalogEntryV4 {
145    id: u128,
146    start_page: u64,
147    page_count: u32,
148    len: u32,
149    deleted: bool,
150}
151
152/// Returned by [`Mnemo::prepare_for_flush`] and consumed by [`Mnemo::flush`].
153/// Carries the already-serialized control-plane bytes through the prelude
154/// so we don't re-serialize after the lease has been persisted.
155struct FlushPrelude {
156    cat_bytes: Option<Vec<u8>>,
157    idx_bytes: Option<Vec<u8>>,
158    /// Serialized result-cache directory bytes when `dirty_cache` is
159    /// set (Phase 10.1). `None` if the cache didn't change since the
160    /// last flush — in that case the header keeps its current
161    /// `cache_start` / `cache_pages` / `cache_len` and no fresh cache
162    /// run is written.
163    cache_bytes: Option<Vec<u8>>,
164}
165
166/// One entry in the append-only snapshot manifest. Because record, catalog,
167/// and index pages are only ever *appended*, the runs a past flush wrote are
168/// still on disk; a `Snapshot` is the set of pointers needed to reconstruct
169/// the database exactly as that flush left it.
170#[derive(Serialize, Deserialize, Clone, Debug)]
171struct Snapshot {
172    txn_id: u64,
173    created_at: i64,
174    catalog_start: u64,
175    catalog_pages: u64,
176    catalog_len: u64,
177    index_start: u64,
178    index_pages: u64,
179    index_len: u64,
180    memory_count: u64,
181}
182
183/// A restorable point in a database's history — see [`Mnemo::snapshots`].
184#[derive(Clone, Copy, Debug)]
185pub struct SnapshotInfo {
186    /// Id of the transaction that produced this snapshot (monotonic from 1).
187    pub txn_id: u64,
188    /// When the snapshot was committed (unix seconds).
189    pub created_at: i64,
190    /// Live memory count captured in the snapshot.
191    pub memory_count: u64,
192}
193
194/// A retrieval request for [`Mnemo::recall`].
195#[derive(Clone, Debug)]
196pub struct RecallRequest {
197    /// Query embedding.
198    pub query: Vec<f32>,
199    /// Maximum results to return.
200    pub top_k: usize,
201    /// Restrict to these memory types (`None` = all types).
202    pub memory_types: Option<Vec<MemoryType>>,
203    /// Restrict to a single agent's view: its own memories plus shared ones.
204    pub agent_id: Option<String>,
205    /// Similarity metric.
206    pub metric: Metric,
207    /// Multi-signal score weights.
208    pub weights: ScoreWeights,
209    /// Index override: partitions to probe (`None` = index default). Ignored
210    /// when no ANN index is present.
211    pub n_probe: Option<usize>,
212    /// Index override: candidates to rerank (`None` = index default). Ignored
213    /// when no ANN index is present.
214    pub n_rerank: Option<usize>,
215    /// Whether to update each result's `accessed_at` and `access_count` in
216    /// the catalog. Defaults to `true`, matching pre-v5 behavior.
217    ///
218    /// Set to `false` for **fully read-only recall** — useful for batch
219    /// scoring, dry-run scoring, or recalls run by introspection tooling
220    /// (e.g. `mnemo about` consumers) where you don't want the score's own
221    /// observation to perturb the database. With `false`, recall does not
222    /// dirty the catalog and the next `flush` is a no-op.
223    pub track_access: bool,
224}
225
226impl RecallRequest {
227    /// A request with sensible defaults for a query embedding.
228    pub fn new(query: Vec<f32>) -> Self {
229        Self {
230            query,
231            top_k: 10,
232            memory_types: None,
233            agent_id: None,
234            metric: Metric::Cosine,
235            weights: ScoreWeights::default(),
236            n_probe: None,
237            n_rerank: None,
238            track_access: true,
239        }
240    }
241    /// Set the result cap.
242    pub fn top_k(mut self, k: usize) -> Self {
243        self.top_k = k;
244        self
245    }
246    /// Restrict to specific memory types.
247    pub fn types(mut self, t: Vec<MemoryType>) -> Self {
248        self.memory_types = Some(t);
249        self
250    }
251    /// Restrict to one agent's view.
252    pub fn agent(mut self, agent_id: impl Into<String>) -> Self {
253        self.agent_id = Some(agent_id.into());
254        self
255    }
256    /// Set the similarity metric (default: cosine).
257    pub fn metric(mut self, metric: Metric) -> Self {
258        self.metric = metric;
259        self
260    }
261    /// Replace the multi-signal score weights.
262    pub fn weights(mut self, weights: ScoreWeights) -> Self {
263        self.weights = weights;
264        self
265    }
266    /// Override the number of IVF partitions probed (accuracy/speed dial).
267    pub fn n_probe(mut self, n: usize) -> Self {
268        self.n_probe = Some(n);
269        self
270    }
271    /// Override the number of candidates reranked exactly (accuracy dial).
272    pub fn n_rerank(mut self, n: usize) -> Self {
273        self.n_rerank = Some(n);
274        self
275    }
276    /// Toggle whether recall updates `accessed_at` / `access_count` on
277    /// the returned memories' catalog entries (default `true`). Pass
278    /// `false` for a fully read-only recall — see the field doc.
279    pub fn track_access(mut self, track: bool) -> Self {
280        self.track_access = track;
281        self
282    }
283}
284
285/// One scored result from [`Mnemo::recall`].
286#[derive(Clone, Debug)]
287pub struct RecallResult {
288    /// The retrieved memory.
289    pub memory: Memory,
290    /// Combined multi-signal score.
291    pub score: f32,
292    /// The bare similarity component (before other signals).
293    pub similarity: f32,
294}
295
296/// Summary statistics for a database.
297#[derive(Clone, Debug)]
298pub struct Stats {
299    /// Live (non-deleted) memory count.
300    pub memories: usize,
301    /// Tombstoned entries awaiting compaction.
302    pub deleted: usize,
303    /// Embedding dimensionality.
304    pub dimensions: usize,
305    /// File size in bytes.
306    pub file_bytes: u64,
307    /// Distinct agent IDs present.
308    pub agents: Vec<String>,
309    /// Whether pages are encrypted (always true in v1).
310    pub encrypted: bool,
311    /// Creation time (unix seconds).
312    pub created_at: i64,
313    /// ANN index shape, if an index has been built.
314    pub index: Option<IndexInfo>,
315    /// Current size of the write-ahead log region, in 8 KiB pages.
316    pub wal_pages: u64,
317}
318
319/// Result of a [`Mnemo::compact_file`] run.
320#[derive(Clone, Copy, Debug)]
321pub struct CompactReport {
322    /// Live memories before compaction.
323    pub before: usize,
324    /// Live memories after compaction (expired ones dropped).
325    pub after: usize,
326}
327
328/// An encrypted, single-file agent memory database.
329pub struct Mnemo {
330    pager: Pager,
331    header: Header,
332    catalog: Vec<CatalogEntry>,
333    index: HashMap<u128, usize>,
334    #[allow(dead_code)]
335    path: PathBuf,
336    dimensions: usize,
337    kdf: KdfParams,
338    /// Set whenever the catalog changes; drives whether `flush` rewrites it.
339    dirty_catalog: bool,
340    /// Optional IVF+PQ approximate-nearest-neighbour index.
341    ann: Option<IvfPqIndex>,
342    /// Set whenever the ANN index changes; drives whether `flush` rewrites it.
343    dirty_index: bool,
344    /// Append-only manifest of committed snapshots, oldest first. Capped
345    /// at [`Mnemo::max_snapshots`] entries on every flush — the oldest
346    /// entries are pruned first.
347    manifest: Vec<Snapshot>,
348    /// Maximum manifest entries to retain across flushes. `0` disables
349    /// the cap (retain forever). Defaults to [`DEFAULT_MAX_SNAPSHOTS`].
350    max_snapshots: usize,
351    /// If `true`, this handle was opened via [`Mnemo::open_read_only`] and
352    /// every mutating API returns [`MnemoError::ReadOnly`]. The underlying
353    /// file was opened without write permission and holds an OS shared
354    /// lock; the OS would reject a write() anyway, but gating in Rust
355    /// gives callers a clean, typed error.
356    read_only: bool,
357    /// Result-cache directory (Phase 10.1). See [`crate::result_cache`].
358    /// Loaded from `header.cache_start` on open; persisted on `flush`
359    /// via the same seal_run pipeline as catalog/index/manifest.
360    cache_dir: Vec<CacheDirectoryEntry>,
361    /// In-memory index over `cache_dir` for O(1) `(namespace, key_hash)`
362    /// lookups. Rebuilt on open + kept in sync on `cache_put`/`delete`.
363    cache_idx: DirectoryIndex,
364    /// Set whenever `cache_dir` changes; drives whether `flush` rewrites
365    /// the cache directory run.
366    dirty_cache: bool,
367    /// Per-namespace eviction budgets. Missing entries fall back to
368    /// [`CacheBudget::default`]. Not persisted (per-handle policy).
369    cache_budgets: std::collections::HashMap<String, CacheBudget>,
370    /// Flush policy for cache mutations. Memory writes ignore this and
371    /// always behave as `Strict`.
372    cache_flush_policy: CacheFlushPolicy,
373    /// Bookkeeping for `CacheFlushPolicy::Batched` — pending mutation
374    /// count and first-dirty timestamp.
375    cache_batch: BatchState,
376    /// In-memory hit/miss/eviction counters (reset on reopen).
377    cache_counters: CounterState,
378}
379
380/// Open a file and take an OS advisory lock, returning `Locked { path }`
381/// if the requested lock cannot be acquired. On drop of the returned
382/// [`File`], the OS releases the lock. `flock`/`LockFileEx` semantics
383/// (per file-descriptor on Unix, per-open-file-handle on Windows) mean
384/// the lock is scoped to this handle: cloning the file (or the returned
385/// `File`) shares the lock, but a fresh `open()` from any process gets a
386/// distinct handle and will collide.
387fn open_with_lock(path: &Path, write: bool, create: bool) -> Result<File> {
388    let mut opts = OpenOptions::new();
389    opts.read(true);
390    if write {
391        opts.write(true);
392    }
393    if create {
394        opts.create(true).truncate(true);
395    }
396    let file = opts.open(path)?;
397    // On success the returned `Result<()>` is `Ok(())`. On conflict:
398    //   * Unix `flock(2)` → `ErrorKind::WouldBlock` (EWOULDBLOCK / EAGAIN)
399    //   * Windows `LockFileEx` → raw OS code 33 (ERROR_LOCK_VIOLATION)
400    //     or 32 (ERROR_SHARING_VIOLATION), both surfaced by Rust as
401    //     `ErrorKind::Uncategorized` because `std::io::ErrorKind`
402    //     doesn't carry a portable "locked" variant.
403    // Anything else is a real I/O failure (permissions, invalid handle,
404    // ...) and gets bubbled up as `MnemoError::Io`.
405    let result = if write {
406        FileExt::try_lock_exclusive(&file)
407    } else {
408        FileExt::try_lock_shared(&file)
409    };
410    match result {
411        Ok(()) => Ok(file),
412        Err(e) if is_lock_conflict(&e) => Err(MnemoError::Locked {
413            path: path.to_path_buf(),
414        }),
415        Err(e) => Err(MnemoError::Io(e)),
416    }
417}
418
419/// Was this I/O error the OS reporting a conflicting file lock, or a
420/// genuine I/O failure? Platform-conditional because
421/// [`std::io::ErrorKind`] carries no portable "resource-locked" variant
422/// as of Rust 1.75 — Windows surfaces `ERROR_LOCK_VIOLATION` /
423/// `ERROR_SHARING_VIOLATION` as `Uncategorized`.
424fn is_lock_conflict(e: &std::io::Error) -> bool {
425    if e.kind() == std::io::ErrorKind::WouldBlock {
426        return true;
427    }
428    #[cfg(windows)]
429    if let Some(code) = e.raw_os_error() {
430        // 33 = ERROR_LOCK_VIOLATION, 32 = ERROR_SHARING_VIOLATION —
431        // both mean "someone else holds an incompatible lock on this
432        // region of the file". See
433        // https://learn.microsoft.com/en-us/windows/win32/debug/system-error-codes--0-499-
434        if code == 33 || code == 32 {
435            return true;
436        }
437    }
438    false
439}
440
441/// Read a run of consecutive encrypted pages and concatenate their plaintext.
442fn read_run_bytes(pager: &mut Pager, start: u64, pages: u64, len: u64) -> Result<Vec<u8>> {
443    let mut buf = Vec::with_capacity(len as usize);
444    for i in 0..pages {
445        buf.extend_from_slice(&pager.read_page(start + i)?);
446    }
447    buf.truncate(len as usize);
448    Ok(buf)
449}
450
451/// Load a v5+ catalog run (an empty `pages` yields an empty catalog).
452fn load_catalog(pager: &mut Pager, start: u64, pages: u64, len: u64) -> Result<Vec<CatalogEntry>> {
453    if pages == 0 {
454        return Ok(Vec::new());
455    }
456    let buf = read_run_bytes(pager, start, pages, len)?;
457    rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
458}
459
460/// Load a **v4** catalog run using the frozen [`CatalogEntryV4`] shape.
461/// Used only by the v4→v5 migration path in [`Mnemo::open`].
462fn load_catalog_v4(
463    pager: &mut Pager,
464    start: u64,
465    pages: u64,
466    len: u64,
467) -> Result<Vec<CatalogEntryV4>> {
468    if pages == 0 {
469        return Ok(Vec::new());
470    }
471    let buf = read_run_bytes(pager, start, pages, len)?;
472    rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
473}
474
475/// Walk a v4 catalog and synthesize the v5 shape. For each live entry,
476/// reads the memory's serialized body and copies its `accessed_at` and
477/// `access_count` into the catalog row (the v4 record body still carries
478/// them — they're moving *from* the body *to* the catalog). Deleted
479/// entries skip the read and zero the fields.
480fn migrate_v4_catalog(pager: &mut Pager, v4: Vec<CatalogEntryV4>) -> Result<Vec<CatalogEntry>> {
481    let mut out = Vec::with_capacity(v4.len());
482    for e in &v4 {
483        let (accessed_at, access_count) = if e.deleted {
484            (0i64, 0u32)
485        } else {
486            let mut buf = Vec::with_capacity(e.len as usize);
487            for i in 0..e.page_count as u64 {
488                buf.extend_from_slice(&pager.read_page(e.start_page + i)?);
489            }
490            buf.truncate(e.len as usize);
491            let mem: Memory = rmp_serde::from_slice(&buf)
492                .map_err(|err| MnemoError::Serialize(err.to_string()))?;
493            (mem.accessed_at, mem.access_count)
494        };
495        out.push(CatalogEntry {
496            id: e.id,
497            start_page: e.start_page,
498            page_count: e.page_count,
499            len: e.len,
500            deleted: e.deleted,
501            accessed_at,
502            access_count,
503        });
504    }
505    Ok(out)
506}
507
508/// Load the snapshot manifest (an empty `pages` yields an empty manifest).
509fn load_manifest(pager: &mut Pager, start: u64, pages: u64, len: u64) -> Result<Vec<Snapshot>> {
510    if pages == 0 {
511        return Ok(Vec::new());
512    }
513    let buf = read_run_bytes(pager, start, pages, len)?;
514    rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
515}
516
517/// Load the result-cache directory (v8+). An empty `pages` yields an
518/// empty directory, which is the natural state for v7 files migrated
519/// to v8 (no cache entries yet) and for freshly created v8 files.
520fn load_cache_directory(
521    pager: &mut Pager,
522    start: u64,
523    pages: u64,
524    len: u64,
525) -> Result<Vec<CacheDirectoryEntry>> {
526    if pages == 0 {
527        return Ok(Vec::new());
528    }
529    let buf = read_run_bytes(pager, start, pages, len)?;
530    rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
531}
532
533/// Load an ANN index run, validating its dimensionality.
534fn load_index(
535    pager: &mut Pager,
536    start: u64,
537    pages: u64,
538    len: u64,
539    dims: usize,
540) -> Result<Option<IvfPqIndex>> {
541    if pages == 0 {
542        return Ok(None);
543    }
544    let buf = read_run_bytes(pager, start, pages, len)?;
545    let mut idx: IvfPqIndex =
546        rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
547    if idx.dims() != dims {
548        return Err(MnemoError::Invalid(
549            "ANN index dimensionality does not match the database".into(),
550        ));
551    }
552    idx.rebuild_assignment();
553    Ok(Some(idx))
554}
555
556/// True if a memory's metadata marks it as the canonical onboarding manifest
557/// (`metadata.topic == "manifest"`). Used by [`Mnemo::about`] to hoist the
558/// manifest to the top of the briefing regardless of importance ordering.
559fn is_manifest(m: &Memory) -> bool {
560    m.metadata
561        .get("topic")
562        .and_then(|v| v.as_str())
563        .map(|s| s.eq_ignore_ascii_case("manifest"))
564        .unwrap_or(false)
565}
566
567/// Build the ULID → catalog-slot lookup map.
568fn build_id_index(catalog: &[CatalogEntry]) -> HashMap<u128, usize> {
569    let mut m = HashMap::with_capacity(catalog.len());
570    for (i, e) in catalog.iter().enumerate() {
571        m.insert(e.id, i);
572    }
573    m
574}
575
576impl Mnemo {
577    /// Create a brand-new encrypted database at `path`.
578    pub fn create(
579        path: impl Into<PathBuf>,
580        passphrase: &str,
581        config: MnemoConfig,
582    ) -> Result<Mnemo> {
583        let path: PathBuf = path.into();
584        if config.dimensions == 0 {
585            return Err(MnemoError::Invalid("dimensions must be > 0".into()));
586        }
587        // Exclusive lock covers the write path; second writer sees `Locked`.
588        let file = open_with_lock(&path, /*write=*/ true, /*create=*/ true)?;
589
590        let salt = crypto::random_salt();
591        let dek = crypto::random_dek();
592        let kek = crypto::derive_kek(passphrase.as_bytes(), &salt, config.kdf)?;
593        let dek_nonce = crypto::random_nonce();
594        let wrapped = crypto::wrap_dek(&kek, &dek_nonce, &dek)?;
595        if wrapped.len() != WRAPPED_DEK_LEN {
596            return Err(MnemoError::Crypto("unexpected wrapped DEK length".into()));
597        }
598        let mut wrapped_dek = [0u8; WRAPPED_DEK_LEN];
599        wrapped_dek.copy_from_slice(&wrapped);
600
601        let mut header = Header {
602            version: VERSION,
603            page_size: PAGE_SIZE as u32,
604            flags: FLAG_ENCRYPTED,
605            dimensions: config.dimensions as u32,
606            created_at: memory::now_secs(),
607            write_counter: 0,
608            // Page 0 is the header; pages 1..=wal_pages are the WAL;
609            // record/catalog/index pages start after it. Initial WAL size is
610            // chosen per-config (default 8 pages = 64 KiB) and clamped to a
611            // sane floor so the first transaction can always commit.
612            next_page: 1 + config.wal_pages_initial.max(MIN_WAL_PAGES),
613            catalog_start: 0,
614            catalog_pages: 0,
615            catalog_len: 0,
616            vector_count: 0,
617            m_cost: config.kdf.m_cost,
618            t_cost: config.kdf.t_cost,
619            p_cost: config.kdf.p_cost,
620            salt,
621            dek_nonce,
622            wrapped_dek,
623            index_start: 0,
624            index_pages: 0,
625            index_len: 0,
626            wal_start: 1,
627            wal_pages: config.wal_pages_initial.max(MIN_WAL_PAGES),
628            wal_seq: 0,
629            manifest_start: 0,
630            manifest_pages: 0,
631            manifest_len: 0,
632            // Regenerated per write by `apply_seal`; `None` until we seal.
633            seal_nonce: None,
634            seal_tag: None,
635            // v8+ result-cache pointer — zero-initialized on create,
636            // populated only when `cache_put` runs at least once and
637            // the next `flush` seals a directory run.
638            cache_start: 0,
639            cache_pages: 0,
640            cache_len: 0,
641        };
642
643        let mut pager = Pager::new(file, dek, 0, VERSION);
644        let mut page = header.to_page();
645        header.apply_seal(&mut page, pager.dek())?;
646        pager.write_raw(0, &page)?;
647        pager.sync()?;
648
649        Ok(Mnemo {
650            pager,
651            header,
652            catalog: Vec::new(),
653            index: HashMap::new(),
654            path,
655            dimensions: config.dimensions,
656            kdf: config.kdf,
657            dirty_catalog: false,
658            ann: None,
659            dirty_index: false,
660            manifest: Vec::new(),
661            max_snapshots: config.max_snapshots,
662            read_only: false,
663            cache_dir: Vec::new(),
664            cache_idx: DirectoryIndex::new(),
665            dirty_cache: false,
666            cache_budgets: std::collections::HashMap::new(),
667            cache_flush_policy: CacheFlushPolicy::default(),
668            cache_batch: BatchState::default(),
669            cache_counters: CounterState::default(),
670        })
671    }
672
673    /// Open an existing database. A wrong passphrase fails cleanly with
674    /// [`MnemoError::WrongPassphrase`]. Takes an OS exclusive lock for
675    /// the lifetime of the returned handle; a second concurrent open
676    /// (from any process) returns [`MnemoError::Locked`].
677    ///
678    /// For a caller that only needs to read (and can tolerate the
679    /// pending-recovery / pending-migration cases documented on
680    /// [`Mnemo::open_read_only`]), prefer that entry point — it takes a
681    /// shared lock and coexists with other readers.
682    pub fn open(path: impl Into<PathBuf>, passphrase: &str) -> Result<Mnemo> {
683        let path: PathBuf = path.into();
684        let mut file = open_with_lock(&path, /*write=*/ true, /*create=*/ false)?;
685
686        let mut hbuf = [0u8; PAGE_SIZE];
687        file.seek(SeekFrom::Start(0))?;
688        file.read_exact(&mut hbuf)?;
689        let mut header = match Header::from_page(&hbuf) {
690            Ok(h) => h,
691            Err(MnemoError::HeaderChecksum) => {
692                // Page 0 is torn — most likely a crash during a checkpoint's
693                // header write. Try to heal from the WAL at its default site
694                // (a never-grown WAL never moves from page 1). We don't know
695                // the WAL size (the header that holds it is what's torn), so
696                // scan a generous fixed window — `wal::recover` stops at the
697                // first non-magic frame, so reading extra is safe.
698                let healed = wal::recover(&mut file, 1, TORN_HEADER_SCAN_PAGES, 0)?;
699                let frames = healed.ok_or(MnemoError::HeaderChecksum)?;
700                for (page_no, bytes) in &frames {
701                    if bytes.len() != PAGE_SIZE {
702                        return Err(MnemoError::Invalid("WAL frame is not page-sized".into()));
703                    }
704                    file.seek(SeekFrom::Start(page_no * PAGE_SIZE as u64))?;
705                    file.write_all(bytes)?;
706                }
707                file.sync_all()?;
708                file.seek(SeekFrom::Start(0))?;
709                file.read_exact(&mut hbuf)?;
710                Header::from_page(&hbuf)?
711            }
712            Err(e) => return Err(e),
713        };
714
715        // Crash recovery: replay a committed-but-uncheckpointed transaction.
716        // Each frame is a finished page image; the header frame, replayed to
717        // page 0, supersedes the header just read.
718        if let Some(frames) = wal::recover(
719            &mut file,
720            header.wal_start,
721            header.wal_pages,
722            header.wal_seq,
723        )? {
724            for (page_no, bytes) in &frames {
725                if bytes.len() != PAGE_SIZE {
726                    return Err(MnemoError::Invalid("WAL frame is not page-sized".into()));
727                }
728                file.seek(SeekFrom::Start(page_no * PAGE_SIZE as u64))?;
729                file.write_all(bytes)?;
730            }
731            file.sync_all()?;
732            // Re-read the now-current header from the replayed page 0.
733            file.seek(SeekFrom::Start(0))?;
734            file.read_exact(&mut hbuf)?;
735            header = Header::from_page(&hbuf)?;
736        }
737
738        let kdf = KdfParams {
739            m_cost: header.m_cost,
740            t_cost: header.t_cost,
741            p_cost: header.p_cost,
742        };
743        let kek = crypto::derive_kek(passphrase.as_bytes(), &header.salt, kdf)?;
744        let dek = crypto::unwrap_dek(&kek, &header.dek_nonce, &header.wrapped_dek)?;
745
746        // Validate the v7 header seal now that we have the DEK. Returns
747        // `MnemoError::HeaderTampered` if any of the sealed mutable fields
748        // were rewritten after the last legitimate flush. No-op for v6 and
749        // older — pre-v7 files don't carry a seal.
750        header.validate_seal(&dek)?;
751
752        // Build the pager at the file's *current* version so initial reads
753        // use the matching AAD scheme — `Pager::page_aad` returns empty AAD
754        // for v4/v5 (no page binding) and `page_no.to_le_bytes()` for v6+.
755        let on_disk_version = header.version;
756        let mut pager = Pager::new(file, dek, header.write_counter, on_disk_version);
757        let dimensions = header.dimensions as usize;
758
759        // === v4→v5 catalog migration =====================================
760        // v6 reads directly; v5 reads directly; v4 replays the catalog into
761        // the v5 shape by reading each memory's body to populate the new
762        // `accessed_at` / `access_count` fields. All page reads here still
763        // use the v4/v5 AAD scheme (none).
764        let (catalog, migrated_from_v4) = if on_disk_version >= 5 {
765            (
766                load_catalog(
767                    &mut pager,
768                    header.catalog_start,
769                    header.catalog_pages,
770                    header.catalog_len,
771                )?,
772                false,
773            )
774        } else if on_disk_version == 4 {
775            let v4 = load_catalog_v4(
776                &mut pager,
777                header.catalog_start,
778                header.catalog_pages,
779                header.catalog_len,
780            )?;
781            (migrate_v4_catalog(&mut pager, v4)?, true)
782        } else {
783            return Err(MnemoError::UnsupportedVersion(on_disk_version));
784        };
785        let index = build_id_index(&catalog);
786        let ann = load_index(
787            &mut pager,
788            header.index_start,
789            header.index_pages,
790            header.index_len,
791            dimensions,
792        )?;
793        let mut manifest = load_manifest(
794            &mut pager,
795            header.manifest_start,
796            header.manifest_pages,
797            header.manifest_len,
798        )?;
799
800        // === v5→v6 page-encryption migration =============================
801        // Pre-v6 data pages were encrypted under no AAD; v6 binds the home
802        // `page_no` as AAD so a page-transplant attack fails authentication
803        // on read. Re-stage every live record page through `write_page` so
804        // the next flush re-encrypts it under v6 AAD at the same home slot.
805        // Tombstoned entries skip — they're reclaimed by `compact_file`.
806        // Old catalog / ANN / manifest pages stay as orphan ciphertext;
807        // the next flush writes the new runs at fresh slots.
808        let migrated_pages_to_v6 = on_disk_version < 6;
809        if migrated_pages_to_v6 {
810            for e in &catalog {
811                if e.deleted {
812                    continue;
813                }
814                for i in 0..e.page_count as u64 {
815                    let payload = pager.read_page(e.start_page + i)?;
816                    pager.write_page(e.start_page + i, &payload)?;
817                }
818            }
819            pager.set_version(6);
820        }
821
822        // === v6→v7 header-seal migration =================================
823        // v7 adds an AES-GCM seal at the tail of the header page that
824        // authenticates every mutable header field. Pre-v7 files have no
825        // seal — we just bump the version in memory and mark `dirty_catalog`
826        // so the next flush writes a sealed v7 header. Pages are untouched
827        // (page crypto is unchanged from v6).
828        let migrated_header_to_v7 = on_disk_version < 7;
829        if migrated_header_to_v7 {
830            pager.set_version(7);
831        }
832
833        // === v7→v8 cache-directory migration =============================
834        // v8 adds three u64 header fields (cache_start/pages/len) for the
835        // result-cache directory (Phase 10.1). Pre-v8 files have those
836        // three bytes as zero (the on-disk region beyond the v7 seal is
837        // zero-initialized), which is the legal "empty cache directory"
838        // state — no data migration required. We bump the version in
839        // memory and mark dirty_catalog so the next flush writes a
840        // v8-sealed header with the new AAD shape covering the cache
841        // pointer. Pages are untouched.
842        let migrated_header_to_v8 = on_disk_version < 8;
843        if migrated_header_to_v8 {
844            pager.set_version(VERSION);
845        }
846
847        let migrated = migrated_from_v4
848            || migrated_pages_to_v6
849            || migrated_header_to_v7
850            || migrated_header_to_v8;
851        if migrated {
852            // Stamp the version forward in memory; the next flush persists
853            // it via the WAL-committed header frame (and the v7 seal).
854            header.version = VERSION;
855            // Snapshots written under an older format reference page runs
856            // encrypted under the old AAD scheme; this build can't decrypt
857            // them under the new pager. Drop the manifest so the next
858            // flush records a fresh snapshot of the migrated state.
859            // Point-in-time recovery into the pre-migration past is
860            // sacrificed for migration simplicity. Old pages remain on
861            // disk until `compact_file`. (For v6→v7 alone — which doesn't
862            // change page crypto — the old snapshots would technically
863            // still be readable, but dropping them uniformly keeps the
864            // migration policy simple.)
865            manifest.clear();
866        }
867
868        // The ANN index pages on disk were sealed under the *old* AAD scheme
869        // if we migrated pages to v6; force a rewrite at fresh v6-encrypted
870        // slots on the next flush so a future reopen doesn't try to read
871        // them under the new AAD and fail. (Catalog and manifest already
872        // get rewritten via dirty_catalog and the manifest.clear() above.)
873        let dirty_index = migrated_pages_to_v6 && ann.is_some();
874
875        // Load the cache directory (v8+ only). Pre-v8 files have
876        // cache_pages == 0 by header layout, so this trivially yields
877        // an empty directory without touching the disk.
878        let cache_dir = load_cache_directory(
879            &mut pager,
880            header.cache_start,
881            header.cache_pages,
882            header.cache_len,
883        )?;
884        let cache_idx = DirectoryIndex::build(&cache_dir);
885
886        Ok(Mnemo {
887            pager,
888            header,
889            catalog,
890            index,
891            path,
892            dimensions,
893            kdf,
894            dirty_catalog: migrated,
895            ann,
896            dirty_index,
897            manifest,
898            // `Mnemo::open` has no [`MnemoConfig`] to read; default the cap
899            // to [`DEFAULT_MAX_SNAPSHOTS`]. Override with
900            // [`Mnemo::set_max_snapshots`] right after open if you need
901            // unlimited retention or a different cap for this handle.
902            max_snapshots: DEFAULT_MAX_SNAPSHOTS,
903            read_only: false,
904            cache_dir,
905            cache_idx,
906            dirty_cache: false,
907            cache_budgets: std::collections::HashMap::new(),
908            cache_flush_policy: CacheFlushPolicy::default(),
909            cache_batch: BatchState::default(),
910            cache_counters: CounterState::default(),
911        })
912    }
913
914    /// Open an existing database in **read-only** mode. Takes an OS shared
915    /// lock; multiple read-only handles can coexist, but any attempt to
916    /// open the same file read-write elsewhere returns
917    /// [`MnemoError::Locked`]. All mutating methods on the returned handle
918    /// return [`MnemoError::ReadOnly`], including `recall` with
919    /// `track_access(true)`.
920    ///
921    /// Two conditions block a read-only open and return
922    /// [`MnemoError::NeedsWriteOpen`]:
923    ///
924    /// 1. The file has a committed-but-uncheckpointed WAL transaction that
925    ///    would ordinarily be replayed on open. A read-only handle
926    ///    cannot write the recovered pages back.
927    /// 2. The on-disk format version is older than the current
928    ///    `format::VERSION`, requiring an on-disk migration.
929    ///
930    /// In either case, open the file read-write once (which recovers /
931    /// migrates in place), close the handle cleanly, and reopen
932    /// read-only. This is documented in the README's Durability section.
933    pub fn open_read_only(path: impl Into<PathBuf>, passphrase: &str) -> Result<Mnemo> {
934        let path: PathBuf = path.into();
935        let mut file = open_with_lock(&path, /*write=*/ false, /*create=*/ false)?;
936
937        let mut hbuf = [0u8; PAGE_SIZE];
938        file.seek(SeekFrom::Start(0))?;
939        file.read_exact(&mut hbuf)?;
940        // A torn header on a read-only open cannot be healed (that requires
941        // writing recovered frames back). Surface as `NeedsWriteOpen` so the
942        // caller knows to reopen read-write once.
943        let header = Header::from_page(&hbuf).map_err(|e| match e {
944            MnemoError::HeaderChecksum => MnemoError::NeedsWriteOpen {
945                reason: "torn header page needs recovery from WAL",
946            },
947            other => other,
948        })?;
949
950        // Refuse to migrate on read-only opens. v6→v7 is header-only
951        // (no page writes required), but even that bumps `version` on
952        // disk and re-seals the header — which needs write access. Keep
953        // the policy uniform across all sub-version bumps.
954        if header.version < VERSION {
955            return Err(MnemoError::NeedsWriteOpen {
956                reason: "pending format-version migration",
957            });
958        }
959
960        // Refuse if there is a committed-but-uncheckpointed transaction
961        // waiting for `wal::recover` to fold it into home pages.
962        if wal::recover(
963            &mut file,
964            header.wal_start,
965            header.wal_pages,
966            header.wal_seq,
967        )?
968        .is_some()
969        {
970            return Err(MnemoError::NeedsWriteOpen {
971                reason: "pending WAL recovery",
972            });
973        }
974
975        let kdf = KdfParams {
976            m_cost: header.m_cost,
977            t_cost: header.t_cost,
978            p_cost: header.p_cost,
979        };
980        let kek = crypto::derive_kek(passphrase.as_bytes(), &header.salt, kdf)?;
981        let dek = crypto::unwrap_dek(&kek, &header.dek_nonce, &header.wrapped_dek)?;
982        header.validate_seal(&dek)?;
983
984        let mut pager = Pager::new(file, dek, header.write_counter, header.version);
985        let dimensions = header.dimensions as usize;
986
987        let catalog = load_catalog(
988            &mut pager,
989            header.catalog_start,
990            header.catalog_pages,
991            header.catalog_len,
992        )?;
993        let index = build_id_index(&catalog);
994        let ann = load_index(
995            &mut pager,
996            header.index_start,
997            header.index_pages,
998            header.index_len,
999            dimensions,
1000        )?;
1001        let manifest = load_manifest(
1002            &mut pager,
1003            header.manifest_start,
1004            header.manifest_pages,
1005            header.manifest_len,
1006        )?;
1007        let cache_dir = load_cache_directory(
1008            &mut pager,
1009            header.cache_start,
1010            header.cache_pages,
1011            header.cache_len,
1012        )?;
1013        let cache_idx = DirectoryIndex::build(&cache_dir);
1014
1015        Ok(Mnemo {
1016            pager,
1017            header,
1018            catalog,
1019            index,
1020            path,
1021            dimensions,
1022            kdf,
1023            // No mutation is possible; leave both dirty flags false so
1024            // `close()` is a genuine no-op even if we later add an
1025            // auto-flush on drop.
1026            dirty_catalog: false,
1027            ann,
1028            dirty_index: false,
1029            manifest,
1030            max_snapshots: DEFAULT_MAX_SNAPSHOTS,
1031            read_only: true,
1032            cache_dir,
1033            cache_idx,
1034            dirty_cache: false,
1035            cache_budgets: std::collections::HashMap::new(),
1036            cache_flush_policy: CacheFlushPolicy::default(),
1037            cache_batch: BatchState::default(),
1038            cache_counters: CounterState::default(),
1039        })
1040    }
1041
1042    /// Override the manifest snapshot cap on this open handle. `0` disables
1043    /// the cap; any positive value keeps the most-recent `max` snapshots and
1044    /// drops the rest on the next flush. See [`MnemoConfig::max_snapshots`].
1045    pub fn set_max_snapshots(&mut self, max: usize) {
1046        self.max_snapshots = max;
1047    }
1048
1049    /// Embedding dimensionality this database expects.
1050    pub fn dimensions(&self) -> usize {
1051        self.dimensions
1052    }
1053
1054    /// Number of live (non-deleted) memories.
1055    pub fn len(&self) -> usize {
1056        self.catalog.iter().filter(|e| !e.deleted).count()
1057    }
1058
1059    /// True if the database holds no live memories.
1060    pub fn is_empty(&self) -> bool {
1061        self.len() == 0
1062    }
1063
1064    // --- internal page/record helpers -------------------------------------
1065
1066    fn write_record(&mut self, bytes: &[u8]) -> Result<(u64, u32)> {
1067        let pc = bytes.len().div_ceil(PAYLOAD).max(1);
1068        let start = self.header.next_page;
1069        self.header.next_page += pc as u64;
1070        for i in 0..pc {
1071            let lo = i * PAYLOAD;
1072            let hi = ((i + 1) * PAYLOAD).min(bytes.len());
1073            self.pager.write_page(start + i as u64, &bytes[lo..hi])?;
1074        }
1075        Ok((start, pc as u32))
1076    }
1077
1078    fn read_record(&mut self, e: &CatalogEntry) -> Result<Vec<u8>> {
1079        let mut buf = Vec::with_capacity(e.len as usize);
1080        for i in 0..e.page_count as u64 {
1081            buf.extend_from_slice(&self.pager.read_page(e.start_page + i)?);
1082        }
1083        buf.truncate(e.len as usize);
1084        Ok(buf)
1085    }
1086
1087    fn read_memory(&mut self, e: &CatalogEntry) -> Result<Memory> {
1088        let bytes = self.read_record(e)?;
1089        let mut m: Memory =
1090            rmp_serde::from_slice(&bytes).map_err(|err| MnemoError::Serialize(err.to_string()))?;
1091        // The catalog is the source of truth for access stats — the values
1092        // in the record body are a stale snapshot from when the record was
1093        // last written. Overwrite them here so consumers always see live
1094        // values regardless of when the record was last serialized.
1095        m.accessed_at = e.accessed_at;
1096        m.access_count = e.access_count;
1097        Ok(m)
1098    }
1099
1100    /// Serialize and store a memory (insert or overwrite by ID).
1101    fn put(&mut self, mut m: Memory) -> Result<Ulid> {
1102        if m.vector.len() != self.dimensions {
1103            return Err(MnemoError::DimensionMismatch {
1104                expected: self.dimensions,
1105                got: m.vector.len(),
1106            });
1107        }
1108        if m.id == Ulid::nil() {
1109            m.id = Ulid::new();
1110        }
1111        let id_u: u128 = m.id.0;
1112        let bytes = rmp_serde::to_vec(&m).map_err(|e| MnemoError::Serialize(e.to_string()))?;
1113        let (start, pc) = self.write_record(&bytes)?;
1114        // For overwrites preserve the existing access stats; for inserts
1115        // seed them from the memory (which carries them in its body).
1116        let (prev_accessed, prev_count) = match self.index.get(&id_u).copied() {
1117            Some(idx) => (
1118                self.catalog[idx].accessed_at,
1119                self.catalog[idx].access_count,
1120            ),
1121            None => (m.accessed_at, m.access_count),
1122        };
1123        let entry = CatalogEntry {
1124            id: id_u,
1125            start_page: start,
1126            page_count: pc,
1127            len: bytes.len() as u32,
1128            deleted: false,
1129            accessed_at: prev_accessed,
1130            access_count: prev_count,
1131        };
1132        match self.index.get(&id_u).copied() {
1133            Some(idx) => self.catalog[idx] = entry,
1134            None => {
1135                self.index.insert(id_u, self.catalog.len());
1136                self.catalog.push(entry);
1137            }
1138        }
1139        self.dirty_catalog = true;
1140
1141        // Keep the ANN index complete: assign the (new or changed) vector to
1142        // its nearest partition. Centroids/codebook stay fixed until rebuild.
1143        if let Some(ann) = &mut self.ann {
1144            ann.add(id_u, &m.vector);
1145            self.dirty_index = true;
1146        }
1147        Ok(m.id)
1148    }
1149
1150    // --- public CRUD ------------------------------------------------------
1151
1152    /// Store a memory. Returns its ULID. If the memory's ID is nil a fresh
1153    /// one is assigned; an existing ID overwrites in place.
1154    pub fn remember(&mut self, memory: Memory) -> Result<Ulid> {
1155        if self.read_only {
1156            return Err(MnemoError::ReadOnly);
1157        }
1158        self.put(memory)
1159    }
1160
1161    /// Fetch a memory by ID.
1162    pub fn get(&mut self, id: &Ulid) -> Result<Memory> {
1163        let idx = *self
1164            .index
1165            .get(&id.0)
1166            .ok_or_else(|| MnemoError::NotFound(id.to_string()))?;
1167        let entry = self.catalog[idx].clone();
1168        if entry.deleted {
1169            return Err(MnemoError::NotFound(id.to_string()));
1170        }
1171        self.read_memory(&entry)
1172    }
1173
1174    /// Soft-delete a memory (tombstoned; space reclaimed by `compact`).
1175    pub fn delete(&mut self, id: &Ulid) -> Result<()> {
1176        if self.read_only {
1177            return Err(MnemoError::ReadOnly);
1178        }
1179        let idx = *self
1180            .index
1181            .get(&id.0)
1182            .ok_or_else(|| MnemoError::NotFound(id.to_string()))?;
1183        if !self.catalog[idx].deleted {
1184            self.catalog[idx].deleted = true;
1185            self.dirty_catalog = true;
1186        }
1187        if let Some(ann) = &mut self.ann {
1188            ann.remove(id.0);
1189            self.dirty_index = true;
1190        }
1191        Ok(())
1192    }
1193
1194    /// Return every live memory (used by tooling and compaction).
1195    pub fn memories(&mut self) -> Result<Vec<Memory>> {
1196        let entries: Vec<CatalogEntry> = self
1197            .catalog
1198            .iter()
1199            .filter(|e| !e.deleted)
1200            .cloned()
1201            .collect();
1202        let mut out = Vec::with_capacity(entries.len());
1203        for e in &entries {
1204            out.push(self.read_memory(e)?);
1205        }
1206        Ok(out)
1207    }
1208
1209    /// Return the database's **self-describing onboarding memories** — the
1210    /// ones tagged `metadata.area = "onboarding"`. This is the engine-level
1211    /// surface for the *single-file philosophy*: an agent who receives a
1212    /// `.mnemo` file (and its passphrase) can call this to learn what the
1213    /// file is, which embedder it expects, the recommended agent id, and any
1214    /// other conventions the file's author chose to record — all without
1215    /// needing any external documentation.
1216    ///
1217    /// Ordering: the canonical manifest (tag `metadata.topic = "manifest"`)
1218    /// always comes first; everything else follows in `importance` descending,
1219    /// then `created_at` ascending for deterministic results.
1220    pub fn about(&mut self) -> Result<Vec<Memory>> {
1221        let mut out: Vec<Memory> = self
1222            .memories()?
1223            .into_iter()
1224            .filter(|m| {
1225                m.metadata
1226                    .get("area")
1227                    .and_then(|v| v.as_str())
1228                    .map(|s| s.eq_ignore_ascii_case("onboarding"))
1229                    .unwrap_or(false)
1230            })
1231            .collect();
1232        out.sort_by(|a, b| {
1233            let a_manifest = is_manifest(a);
1234            let b_manifest = is_manifest(b);
1235            // Manifest topic wins first; then importance desc; then created_at asc.
1236            b_manifest
1237                .cmp(&a_manifest)
1238                .then_with(|| b.importance.total_cmp(&a.importance))
1239                .then_with(|| a.created_at.cmp(&b.created_at))
1240        });
1241        Ok(out)
1242    }
1243
1244    // --- retrieval --------------------------------------------------------
1245
1246    /// Phase-1 brute-force search: rank live memories by raw similarity only.
1247    /// Read-only — does not update access statistics.
1248    pub fn search(
1249        &mut self,
1250        query: &[f32],
1251        top_k: usize,
1252        metric: Metric,
1253    ) -> Result<Vec<(Memory, f32)>> {
1254        if query.len() != self.dimensions {
1255            return Err(MnemoError::DimensionMismatch {
1256                expected: self.dimensions,
1257                got: query.len(),
1258            });
1259        }
1260        let now = memory::now_secs();
1261        let entries: Vec<CatalogEntry> = self
1262            .catalog
1263            .iter()
1264            .filter(|e| !e.deleted)
1265            .cloned()
1266            .collect();
1267
1268        let mut scored: Vec<(Memory, f32)> = Vec::new();
1269        for e in &entries {
1270            let m = self.read_memory(e)?;
1271            if m.is_expired(now) {
1272                continue;
1273            }
1274            let sim = memory::similarity(metric, query, &m.vector);
1275            scored.push((m, sim));
1276        }
1277        scored.sort_by(|a, b| b.1.total_cmp(&a.1));
1278        scored.truncate(top_k);
1279        Ok(scored)
1280    }
1281
1282    /// Phase-5 multi-signal recall: rank by
1283    /// `α·similarity + β·recency + γ·importance + δ·ln(freq)`, with type and
1284    /// agent filtering. Updates `accessed_at` / `access_count` on the
1285    /// returned memories (persisted on the next `flush`).
1286    ///
1287    /// When an ANN index has been built ([`Mnemo::build_index`]), recall runs
1288    /// the tiered IVF→PQ→rerank pipeline: it scores only the similarity-nearest
1289    /// `n_rerank` candidates rather than every memory. Without an index it
1290    /// scores every live memory exactly.
1291    pub fn recall(&mut self, req: &RecallRequest) -> Result<Vec<RecallResult>> {
1292        if req.query.len() != self.dimensions {
1293            return Err(MnemoError::DimensionMismatch {
1294                expected: self.dimensions,
1295                got: req.query.len(),
1296            });
1297        }
1298        // Access-tracking bumps `accessed_at` / `access_count` in the
1299        // catalog, which is a mutation. On a read-only handle the caller
1300        // must pass `track_access(false)` (via CLI `--no-track`).
1301        if self.read_only && req.track_access {
1302            return Err(MnemoError::ReadOnly);
1303        }
1304        let now = memory::now_secs();
1305
1306        // Candidate set: ANN-narrowed when an index exists, else everything.
1307        let entries: Vec<CatalogEntry> = if let Some(ann) = &self.ann {
1308            let ids = ann.query(&req.query, req.n_probe, req.n_rerank);
1309            ids.iter()
1310                .filter_map(|id| self.index.get(id).copied())
1311                .map(|i| self.catalog[i].clone())
1312                .filter(|e| !e.deleted)
1313                .collect()
1314        } else {
1315            self.catalog
1316                .iter()
1317                .filter(|e| !e.deleted)
1318                .cloned()
1319                .collect()
1320        };
1321
1322        let mut scored: Vec<RecallResult> = Vec::new();
1323        for e in &entries {
1324            let m = self.read_memory(e)?;
1325            if m.is_expired(now) {
1326                continue;
1327            }
1328            // Type filter.
1329            if let Some(types) = &req.memory_types {
1330                if !types.contains(&m.memory_type) {
1331                    continue;
1332                }
1333            }
1334            // Agent-scoping: an agent sees its own memories plus shared ones.
1335            if let Some(agent) = &req.agent_id {
1336                let visible = m.agent_id == *agent || m.scope == Scope::Shared;
1337                if !visible {
1338                    continue;
1339                }
1340            }
1341            let sim = memory::similarity(req.metric, &req.query, &m.vector);
1342            let age = (now - m.accessed_at) as f32;
1343            let score = req.weights.score(sim, age, m.importance, m.access_count);
1344            scored.push(RecallResult {
1345                memory: m,
1346                score,
1347                similarity: sim,
1348            });
1349        }
1350        scored.sort_by(|a, b| b.score.total_cmp(&a.score));
1351        scored.truncate(req.top_k);
1352
1353        // Update access statistics for everything we surfaced — but only
1354        // mutate the catalog entry, not the full record on disk. Pre-v5
1355        // this loop called `self.put(r.memory.clone())`, which serialized
1356        // and rewrote every result (vector included) to fresh pages,
1357        // turning a top-K recall into K full record rewrites. Now the
1358        // catalog is the source of truth for these two fields and the
1359        // record body stays untouched until the next real edit.
1360        if req.track_access {
1361            for r in &mut scored {
1362                if let Some(&idx) = self.index.get(&r.memory.id.0) {
1363                    let entry = &mut self.catalog[idx];
1364                    entry.accessed_at = now;
1365                    entry.access_count = entry.access_count.saturating_add(1);
1366                    // Propagate the just-updated values onto the returned
1367                    // Memory so the caller sees the post-recall state.
1368                    r.memory.accessed_at = entry.accessed_at;
1369                    r.memory.access_count = entry.access_count;
1370                }
1371            }
1372            // Mark the catalog dirty so the next flush persists these
1373            // bumps. The records themselves are not dirty.
1374            if !scored.is_empty() {
1375                self.dirty_catalog = true;
1376            }
1377        }
1378        Ok(scored)
1379    }
1380
1381    // --- approximate index ------------------------------------------------
1382
1383    /// Build an IVF+PQ approximate-nearest-neighbour index over every live
1384    /// memory, using default tuning. After this, [`Mnemo::recall`] runs the
1385    /// tiered pipeline instead of an exact scan. Persisted on the next
1386    /// `flush`. Returns a snapshot of the index shape.
1387    pub fn build_index(&mut self) -> Result<IndexInfo> {
1388        self.build_index_with(IndexConfig::default())
1389    }
1390
1391    /// Build the ANN index with explicit tuning.
1392    pub fn build_index_with(&mut self, cfg: IndexConfig) -> Result<IndexInfo> {
1393        if self.read_only {
1394            return Err(MnemoError::ReadOnly);
1395        }
1396        let mems = self.memories()?;
1397        if mems.is_empty() {
1398            return Err(MnemoError::Invalid(
1399                "cannot build an index over an empty database".into(),
1400            ));
1401        }
1402        let items: Vec<(u128, &[f32])> =
1403            mems.iter().map(|m| (m.id.0, m.vector.as_slice())).collect();
1404        let idx = IvfPqIndex::build(self.dimensions, &items, cfg)?;
1405        let info = idx.info();
1406        self.ann = Some(idx);
1407        self.dirty_index = true;
1408        Ok(info)
1409    }
1410
1411    /// Rebuild the ANN index from scratch — re-clusters centroids and retrains
1412    /// the PQ codebook, undoing the cluster drift that accumulates as memories
1413    /// are inserted against fixed centroids.
1414    pub fn rebuild_index(&mut self) -> Result<IndexInfo> {
1415        let cfg = match &self.ann {
1416            Some(a) => IndexConfig {
1417                n_probe: a.n_probe(),
1418                n_rerank: a.n_rerank(),
1419                ..Default::default()
1420            },
1421            None => IndexConfig::default(),
1422        };
1423        self.build_index_with(cfg)
1424    }
1425
1426    /// Drop the ANN index; recall reverts to exact scans. Persisted on flush.
1427    /// No-op on a read-only handle (signature is infallible — callers of
1428    /// `build_index` / `flush` on a read-only handle already get a typed
1429    /// `ReadOnly` error, which is the enforceable boundary).
1430    pub fn drop_index(&mut self) {
1431        if self.read_only {
1432            return;
1433        }
1434        if self.ann.is_some() {
1435            self.ann = None;
1436            self.dirty_index = true;
1437        }
1438    }
1439
1440    /// Whether an ANN index is currently loaded.
1441    pub fn has_index(&self) -> bool {
1442        self.ann.is_some()
1443    }
1444
1445    // --- durability & maintenance ----------------------------------------
1446
1447    /// Seal a serialized buffer into a fresh run of encrypted home pages and
1448    /// append a WAL frame for each. Returns the run's `(start_page, pages)`.
1449    /// The pages are *not* written to their home locations here — they go to
1450    /// the WAL and are folded in later by [`Mnemo::checkpoint`].
1451    fn seal_run(&mut self, bytes: &[u8], frames: &mut Vec<wal::Frame>) -> Result<(u64, u32)> {
1452        let pc = bytes.len().div_ceil(PAYLOAD).max(1);
1453        let start = self.header.next_page;
1454        self.header.next_page += pc as u64;
1455        for i in 0..pc {
1456            let lo = i * PAYLOAD;
1457            let hi = ((i + 1) * PAYLOAD).min(bytes.len());
1458            let page_no = start + i as u64;
1459            let sealed = self.pager.seal_page(page_no, &bytes[lo..hi])?;
1460            frames.push((page_no, sealed.to_vec()));
1461        }
1462        Ok((start, pc as u32))
1463    }
1464
1465    /// Grow the WAL region if `frame_pages` worth of frames would not fit.
1466    /// Called at the top of `flush`, where the WAL is always spent (every
1467    /// flush ends with a checkpoint), so relocating it cannot strand a
1468    /// committed transaction.
1469    fn ensure_wal_capacity(&mut self, frame_pages: usize) -> Result<()> {
1470        let need = wal::txn_byte_len(frame_pages, PAGE_SIZE);
1471        if need <= self.header.wal_pages * PAGE_SIZE as u64 {
1472            return Ok(());
1473        }
1474        // Allocate a new, larger region at the file tail (~1.5x headroom).
1475        let req = need.div_ceil(PAGE_SIZE as u64);
1476        let new_pages = (req + req / 2 + 4).max(DEFAULT_WAL_PAGES);
1477        let new_start = self.header.next_page;
1478        self.header.next_page += new_pages;
1479        self.header.wal_start = new_start;
1480        self.header.wal_pages = new_pages;
1481        // Persist the relocation now: an isolated header write over an empty
1482        // WAL. A crash here leaves a consistent state with a bigger WAL.
1483        self.header.write_counter = self.pager.write_counter;
1484        let mut page = self.header.to_page();
1485        self.header.apply_seal(&mut page, self.pager.dek())?;
1486        self.pager.write_raw(0, &page)?;
1487        self.pager.sync()?;
1488        Ok(())
1489    }
1490
1491    /// Fold a committed transaction's WAL frames into their home pages.
1492    fn checkpoint(&mut self, frames: &[wal::Frame]) -> Result<()> {
1493        for (page_no, bytes) in frames {
1494            let mut img = [0u8; PAGE_SIZE];
1495            img.copy_from_slice(bytes);
1496            self.pager.write_sealed(*page_no, &img)?;
1497        }
1498        self.pager.sync()?;
1499        Ok(())
1500    }
1501
1502    /// Persist all pending changes as one **write-ahead-logged transaction**.
1503    ///
1504    /// Sequence:
1505    ///
1506    /// 1. **Prepare.** Serialize the control plane (catalog, ANN index),
1507    ///    grow the WAL if needed, and **lease** counter and page slots —
1508    ///    bump `header.write_counter` and `header.next_page` by upper
1509    ///    bounds on this transaction's consumption and persist that
1510    ///    *leased* header to disk before any encrypted page is written.
1511    /// 2. **Data pages.** [`Pager::flush`] writes dirty record pages
1512    ///    copy-on-write under fresh nonces and fsyncs them. The orphan-page
1513    ///    nonce-reuse window (see [Phase 1.1 of the improvement plan])
1514    ///    is closed because the leased header on disk already records a
1515    ///    `write_counter` past anything this step can produce.
1516    /// 3. **Control plane.** Seal the new catalog / ANN index / snapshot
1517    ///    manifest into fresh home page runs.
1518    /// 4. **Commit.** Log all of step 3's frames plus the final header
1519    ///    frame into the WAL and fsync — the durability point.
1520    /// 5. **Checkpoint.** Fold the WAL into the home pages.
1521    ///
1522    /// A crash before step 4 leaves the previous state intact. A crash
1523    /// after step 4 is repaired by [`Mnemo::open`] replaying the WAL.
1524    /// Safe to call repeatedly.
1525    pub fn flush(&mut self) -> Result<()> {
1526        if self.read_only {
1527            return Err(MnemoError::ReadOnly);
1528        }
1529        if !self.dirty_catalog && !self.dirty_index && !self.dirty_cache {
1530            // No control-plane changes pending. Because every `put` that
1531            // dirties a data page also dirties the catalog, and every
1532            // `cache_put` that dirties a data page also dirties
1533            // `dirty_cache`, there can't be dirty data pages either —
1534            // nothing to do.
1535            return Ok(());
1536        }
1537
1538        // 1. Serialize control plane, ensure WAL capacity, lease counter +
1539        //    page slots, and persist the leased header *before* any data
1540        //    page hits the disk.
1541        let ctx = self.prepare_for_flush()?;
1542
1543        // 2. Record (vector) data pages: copy-on-write to fresh pages, fsynced.
1544        //    Safe: the lease guarantees that even if we crash here, reopen
1545        //    will see a `write_counter` past every nonce this step uses.
1546        self.pager.flush()?;
1547
1548        // 3. Seal the catalog / index control pages into fresh home runs.
1549        let mut frames: Vec<wal::Frame> = Vec::new();
1550        if let Some(bytes) = &ctx.cat_bytes {
1551            let (start, pages) = self.seal_run(bytes, &mut frames)?;
1552            self.header.catalog_start = start;
1553            self.header.catalog_pages = pages as u64;
1554            self.header.catalog_len = bytes.len() as u64;
1555        }
1556        if self.dirty_index {
1557            match &ctx.idx_bytes {
1558                Some(bytes) => {
1559                    let (start, pages) = self.seal_run(bytes, &mut frames)?;
1560                    self.header.index_start = start;
1561                    self.header.index_pages = pages as u64;
1562                    self.header.index_len = bytes.len() as u64;
1563                }
1564                None => {
1565                    self.header.index_start = 0;
1566                    self.header.index_pages = 0;
1567                    self.header.index_len = 0;
1568                }
1569            }
1570        }
1571        // v8+ result-cache directory (Phase 10.1). Seal the serialized
1572        // directory into a fresh page run and update the pointer in the
1573        // header. The seal AAD already covers cache_start/pages/len for
1574        // v8+ (see `header_seal_aad`), so tampering with the pointer
1575        // trips the header seal on the next open.
1576        if self.dirty_cache {
1577            match &ctx.cache_bytes {
1578                Some(bytes) if !self.cache_dir.is_empty() => {
1579                    let (start, pages) = self.seal_run(bytes, &mut frames)?;
1580                    self.header.cache_start = start;
1581                    self.header.cache_pages = pages as u64;
1582                    self.header.cache_len = bytes.len() as u64;
1583                }
1584                _ => {
1585                    // Either the directory serialized to nothing (all
1586                    // entries deleted after tombstone-only migration) or
1587                    // it's empty — collapse the pointer to zero.
1588                    self.header.cache_start = 0;
1589                    self.header.cache_pages = 0;
1590                    self.header.cache_len = 0;
1591                }
1592            }
1593        }
1594
1595        // 3b. Record this transaction as a restorable snapshot. The manifest
1596        //     update is staged in a local and adopted only once the commit
1597        //     below succeeds, so a failed flush leaves no phantom entry.
1598        //
1599        //     The cap (`self.max_snapshots`) is applied to the staged local
1600        //     before the new entry lands, so after appending the manifest
1601        //     holds at most `max_snapshots` entries — the most-recent N.
1602        //     `max_snapshots == 0` disables the cap. Pages referenced only
1603        //     by pruned snapshots stay on disk until `compact_file`.
1604        let live = self.len() as u64;
1605        let txn_id = self.header.wal_seq + 1;
1606        let mut manifest = self.manifest.clone();
1607        if self.max_snapshots > 0 {
1608            // Drop the oldest entries so there's room for the new one.
1609            while manifest.len() >= self.max_snapshots {
1610                manifest.remove(0);
1611            }
1612        }
1613        manifest.push(Snapshot {
1614            txn_id,
1615            created_at: memory::now_secs(),
1616            catalog_start: self.header.catalog_start,
1617            catalog_pages: self.header.catalog_pages,
1618            catalog_len: self.header.catalog_len,
1619            index_start: self.header.index_start,
1620            index_pages: self.header.index_pages,
1621            index_len: self.header.index_len,
1622            memory_count: live,
1623        });
1624        let man_bytes =
1625            rmp_serde::to_vec(&manifest).map_err(|e| MnemoError::Serialize(e.to_string()))?;
1626        let (m_start, m_pages) = self.seal_run(&man_bytes, &mut frames)?;
1627        self.header.manifest_start = m_start;
1628        self.header.manifest_pages = m_pages as u64;
1629        self.header.manifest_len = man_bytes.len() as u64;
1630
1631        // 3c. The header is the transaction's final frame; stamp the new id
1632        //     and the *actual* (post-flush) write counter, which is <= the
1633        //     leased value persisted in `prepare_for_flush`. This is the
1634        //     value that ultimately replaces the leased header on checkpoint.
1635        self.header.vector_count = live;
1636        self.header.write_counter = self.pager.write_counter;
1637        self.header.wal_seq = txn_id;
1638        let mut hpage = self.header.to_page();
1639        self.header.apply_seal(&mut hpage, self.pager.dek())?;
1640        frames.push((0, hpage.to_vec()));
1641
1642        // 4. COMMIT — log the transaction and fsync. Nothing at a home page
1643        //    other than the leased header has changed yet; this single
1644        //    fsync is what makes the rest durable.
1645        let (wal_start, wal_pages) = (self.header.wal_start, self.header.wal_pages);
1646        wal::commit(self.pager.file_mut(), wal_start, wal_pages, txn_id, &frames)?;
1647
1648        // 5. Checkpoint — fold the WAL into the home pages.
1649        self.checkpoint(&frames)?;
1650
1651        self.manifest = manifest;
1652        self.dirty_catalog = false;
1653        self.dirty_index = false;
1654        self.dirty_cache = false;
1655        // Batched policy: successful flush consumes the pending batch.
1656        self.cache_batch.reset();
1657        Ok(())
1658    }
1659
1660    /// Pre-flush prelude shared by [`Mnemo::flush`] and the
1661    /// `__crash_partial_flush_for_testing` hook.
1662    ///
1663    /// Serializes the control plane (catalog, ANN index), grows the WAL if
1664    /// needed, then **leases** counter and page slots — bumps a *clone* of
1665    /// the header by upper-bound amounts and persists that leased clone
1666    /// with `pager.write_raw + sync`. The in-memory `self.header` keeps
1667    /// its pre-lease `next_page` so [`Mnemo::seal_run`] continues to
1668    /// allocate from the right slot; the final committed header carries
1669    /// the actual post-flush values and overwrites the leased one on
1670    /// checkpoint.
1671    ///
1672    /// This pre-stamping is what closes the nonce-reuse window flagged in
1673    /// Phase 1.1 of the improvement plan: a crash anywhere between here
1674    /// and the WAL commit leaves a file whose on-disk header records a
1675    /// `write_counter` *past* every nonce the rest of this transaction
1676    /// could produce, so the next reopen starts from a counter that
1677    /// guarantees uniqueness against any orphan data pages still on disk.
1678    fn prepare_for_flush(&mut self) -> Result<FlushPrelude> {
1679        // Serialize.
1680        let cat_bytes = if self.dirty_catalog {
1681            Some(
1682                rmp_serde::to_vec(&self.catalog)
1683                    .map_err(|e| MnemoError::Serialize(e.to_string()))?,
1684            )
1685        } else {
1686            None
1687        };
1688        let idx_bytes: Option<Vec<u8>> = if self.dirty_index {
1689            match &self.ann {
1690                Some(ann) => {
1691                    Some(rmp_serde::to_vec(ann).map_err(|e| MnemoError::Serialize(e.to_string()))?)
1692                }
1693                None => None,
1694            }
1695        } else {
1696            None
1697        };
1698        // Serialize the result-cache directory when it's dirty. An
1699        // empty directory serializes to a small header + 0-element
1700        // array; still cheap enough that we don't special-case it.
1701        let cache_bytes: Option<Vec<u8>> = if self.dirty_cache {
1702            Some(
1703                rmp_serde::to_vec(&self.cache_dir)
1704                    .map_err(|e| MnemoError::Serialize(e.to_string()))?,
1705            )
1706        } else {
1707            None
1708        };
1709
1710        let cat_pc = cat_bytes
1711            .as_ref()
1712            .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
1713        let idx_pc = idx_bytes
1714            .as_ref()
1715            .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
1716        let cache_pc = cache_bytes
1717            .as_ref()
1718            .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
1719        // Upper bound on the manifest run: one extra entry, <=82 bytes each
1720        // (rmp_serde encodes a Snapshot as a 9-element fixarray of i64/u64,
1721        // worst case 81 bytes plus a 1-byte array header).
1722        let man_upper = 9 + (self.manifest.len() + 1) * 82;
1723        let man_pc_est = man_upper.div_ceil(PAYLOAD).max(1);
1724
1725        // Grow WAL if needed (does its own header write+sync). The
1726        // cache directory rides in the same transaction so its page
1727        // count contributes to the reservation.
1728        self.ensure_wal_capacity(cat_pc + idx_pc + cache_pc + man_pc_est + 1)?;
1729
1730        // Lease counter and page slots. counter_lease is an upper bound on
1731        // how many `pager.write_counter` advances this transaction will
1732        // make (one per dirty data page + one per sealed control-plane
1733        // page). page_lease covers only the control-plane allocations,
1734        // because dirty data pages were already allocated past
1735        // `header.next_page` by `write_record` when `remember` ran or
1736        // `cache_put` ran.
1737        let data_dirty = self.pager.dirty_page_count() as u64;
1738        let counter_lease = data_dirty + (cat_pc + idx_pc + cache_pc + man_pc_est) as u64;
1739        let page_lease = (cat_pc + idx_pc + cache_pc + man_pc_est) as u64;
1740
1741        // Persist a leased *clone* of the header — in-memory `self.header`
1742        // keeps its pre-lease `next_page` for `seal_run` to consume from
1743        // the correct slot.
1744        let mut leased = self.header.clone();
1745        leased.write_counter = self.pager.write_counter + counter_lease;
1746        leased.next_page += page_lease;
1747        let mut page = leased.to_page();
1748        leased.apply_seal(&mut page, self.pager.dek())?;
1749        self.pager.write_raw(0, &page)?;
1750        self.pager.sync()?;
1751
1752        Ok(FlushPrelude {
1753            cat_bytes,
1754            idx_bytes,
1755            cache_bytes,
1756        })
1757    }
1758
1759    /// Flush and close. Equivalent to `flush()` on a read-write handle; on
1760    /// a read-only handle it's a no-op (no dirty state to persist). The
1761    /// file is released on drop, which also drops the OS advisory lock.
1762    pub fn close(&mut self) -> Result<()> {
1763        if self.read_only {
1764            return Ok(());
1765        }
1766        self.flush()
1767    }
1768
1769    /// **TEST ONLY — do not use.** Reproduces a crash inside
1770    /// [`Mnemo::flush`] *after* the prelude has leased counter+page slots
1771    /// and persisted the leased header, but *before* the WAL is committed.
1772    /// Runs `prepare_for_flush` + `pager.flush` and returns — leaving
1773    /// data pages and a leased header on disk, with no commit frame.
1774    ///
1775    /// Used by `tests/integration.rs::nonce_unique_after_crashed_data_flush`
1776    /// to verify that this window (Phase 1.1 of the improvement plan) does
1777    /// not enable AES-GCM nonce reuse. Calling this in production code
1778    /// strands data pages; never expose it from a binding.
1779    #[doc(hidden)]
1780    pub fn __crash_partial_flush_for_testing(&mut self) -> Result<()> {
1781        if self.read_only {
1782            return Err(MnemoError::ReadOnly);
1783        }
1784        if !self.dirty_catalog && !self.dirty_index {
1785            return Ok(());
1786        }
1787        let _prelude = self.prepare_for_flush()?;
1788        self.pager.flush()
1789    }
1790
1791    /// Bound the in-memory page cache to `pages` decrypted pages.
1792    ///
1793    /// The cache holds decrypted page payloads to speed repeated reads. By
1794    /// default it is capped at 8192 pages (~64 MiB); lower the cap to trade
1795    /// hit rate for a smaller footprint, or raise it for a hotter cache. The
1796    /// cap governs *clean* pages — pages with un-flushed writes are always
1797    /// retained until [`Mnemo::flush`], regardless of the cap.
1798    pub fn set_cache_capacity(&mut self, pages: usize) {
1799        self.pager.set_cache_capacity(pages);
1800    }
1801
1802    /// Page-cache occupancy: `(pages_cached, capacity)`. Renamed from
1803    /// `cache_stats` in v0.4.0: the result-cache API (Phase 10.1) took
1804    /// that name for its per-namespace [`CacheStats`] summary, and the
1805    /// page-cache pair is the older/less-user-facing of the two, so it
1806    /// gets the more-specific name.
1807    pub fn page_cache_stats(&self) -> (usize, usize) {
1808        self.pager.cache_stats()
1809    }
1810
1811    // === Result cache (Phase 10.1 + 10.3) ============================
1812
1813    /// Insert or overwrite a cache entry at `(namespace, key)`.
1814    ///
1815    /// The engine SHA-256-hashes `key` — callers pass the raw input
1816    /// string. Persists to disk on the next `flush()` under the current
1817    /// flush policy (see [`Mnemo::set_cache_flush_policy`]); in
1818    /// [`CacheFlushPolicy::Batched`] the engine may call `flush()`
1819    /// internally when a threshold trips.
1820    ///
1821    /// On overflow of the namespace's [`CacheBudget`] (default 10,000
1822    /// entries / 64 MiB) the LRU entry is tombstoned before the new
1823    /// insert. Tombstones are reclaimed by
1824    /// [`Mnemo::compact_file`], not the runtime flush.
1825    ///
1826    /// Returns [`MnemoError::ReadOnly`] on a read-only handle.
1827    pub fn cache_put(
1828        &mut self,
1829        namespace: &str,
1830        key: &str,
1831        value: &[u8],
1832        opts: CachePutOpts,
1833    ) -> Result<()> {
1834        if self.read_only {
1835            return Err(MnemoError::ReadOnly);
1836        }
1837        let key_hash = result_cache::hash_key(key);
1838        let key_preview = result_cache::key_preview(key);
1839        let now = memory::now_secs();
1840        let size_bytes = value.len() as u64;
1841
1842        let entry = CacheEntry {
1843            namespace: namespace.to_string(),
1844            key_hash,
1845            key_preview: key_preview.clone(),
1846            value: value.to_vec(),
1847            content_type: opts.content_type.clone(),
1848            created_at: now,
1849            accessed_at: now,
1850            access_count: 0,
1851            ttl_secs: opts.ttl_secs,
1852            cost_hint_ms: opts.cost_hint_ms,
1853            size_bytes,
1854            vector: None,
1855            model: None,
1856        };
1857        let bytes = rmp_serde::to_vec(&entry).map_err(|e| MnemoError::Serialize(e.to_string()))?;
1858        let (start_page, page_count) = self.write_record(&bytes)?;
1859
1860        // Tombstone any prior entry for this key. Its pages remain on
1861        // disk until `compact_file`, exactly like memory-record
1862        // overwrites — the tombstone just hides it from lookups.
1863        if let Some(prev_pos) = self.cache_idx.get(namespace, &key_hash) {
1864            self.cache_dir[prev_pos].deleted = true;
1865            self.cache_idx.remove(namespace, &key_hash);
1866        }
1867
1868        let dir_entry = CacheDirectoryEntry {
1869            namespace: namespace.to_string(),
1870            key_hash,
1871            start_page,
1872            page_count,
1873            len: bytes.len() as u32,
1874            deleted: false,
1875            accessed_at: now,
1876            access_count: 0,
1877            size_bytes,
1878            ttl_secs: opts.ttl_secs,
1879            created_at: now,
1880            vector: None,
1881            model: None,
1882        };
1883        let pos = self.cache_dir.len();
1884        self.cache_dir.push(dir_entry);
1885        self.cache_idx.insert(namespace, key_hash, pos);
1886
1887        // Enforce budget — evict LRU tombstones until the namespace
1888        // fits under both caps.
1889        self.evict_to_budget(namespace)?;
1890
1891        self.dirty_cache = true;
1892        self.cache_batch.record_dirty();
1893
1894        // Batched policy: auto-flush if a threshold tripped. Successful
1895        // `flush()` calls `cache_batch.reset()` for us.
1896        if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
1897            self.flush()?;
1898        }
1899        Ok(())
1900    }
1901
1902    /// Look up `(namespace, key)`. Returns `None` on miss (either the
1903    /// key was never inserted or the entry is TTL-expired). On hit
1904    /// updates catalog-side access stats (mirrors the recall v5-trick);
1905    /// on a read-only handle the stats update is silently skipped.
1906    ///
1907    /// A miss is recorded as a miss in this handle's counters even if
1908    /// the key was TTL-expired — from the caller's perspective the two
1909    /// are indistinguishable.
1910    pub fn cache_get(&mut self, namespace: &str, key: &str) -> Result<Option<CachedValue>> {
1911        let key_hash = result_cache::hash_key(key);
1912        let pos = match self.cache_idx.get(namespace, &key_hash) {
1913            Some(p) => p,
1914            None => {
1915                self.cache_counters.misses += 1;
1916                return Ok(None);
1917            }
1918        };
1919        let now = memory::now_secs();
1920        // Snapshot the directory entry fields we need before any
1921        // mutation — the borrow checker gets grumpy otherwise.
1922        let (start_page, page_count, len, ttl_secs, expired, created_at) = {
1923            let e = &self.cache_dir[pos];
1924            (
1925                e.start_page,
1926                e.page_count,
1927                e.len,
1928                e.ttl_secs,
1929                e.is_expired(now),
1930                e.created_at,
1931            )
1932        };
1933        if expired {
1934            // Treat expired as a miss for the caller. Don't tombstone
1935            // here — cache_purge does that. Just skip.
1936            self.cache_counters.misses += 1;
1937            return Ok(None);
1938        }
1939        // Decrypt and decode the record body.
1940        let mut buf = Vec::with_capacity(len as usize);
1941        for i in 0..page_count as u64 {
1942            buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
1943        }
1944        buf.truncate(len as usize);
1945        let entry: CacheEntry =
1946            rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
1947
1948        // Bump access stats catalog-only (v5 trick — no full record
1949        // rewrite on a hit). Skipped on read-only handles so a shared-lock
1950        // reader can't dirty the catalog.
1951        if !self.read_only {
1952            self.cache_dir[pos].accessed_at = now;
1953            self.cache_dir[pos].access_count = self.cache_dir[pos].access_count.saturating_add(1);
1954            self.dirty_cache = true;
1955        }
1956        self.cache_counters.hits += 1;
1957        // On a mutating handle: access_count reflects the just-bumped
1958        // value. On a read-only handle: the on-disk value is unchanged,
1959        // so returning it verbatim is honest (the caller's reads did
1960        // not, and cannot, alter the persisted counter).
1961        let access_count = self.cache_dir[pos].access_count;
1962        Ok(Some(CachedValue {
1963            value: entry.value,
1964            content_type: entry.content_type,
1965            created_at,
1966            accessed_at: if self.read_only {
1967                entry.accessed_at
1968            } else {
1969                now
1970            },
1971            access_count,
1972            ttl_secs,
1973        }))
1974    }
1975
1976    /// Tombstone the entry at `(namespace, key)`. Silently succeeds if
1977    /// no such entry exists. Space is reclaimed by `compact_file`.
1978    pub fn cache_delete(&mut self, namespace: &str, key: &str) -> Result<()> {
1979        if self.read_only {
1980            return Err(MnemoError::ReadOnly);
1981        }
1982        let key_hash = result_cache::hash_key(key);
1983        if let Some(pos) = self.cache_idx.get(namespace, &key_hash) {
1984            self.cache_dir[pos].deleted = true;
1985            self.cache_idx.remove(namespace, &key_hash);
1986            self.dirty_cache = true;
1987            self.cache_batch.record_dirty();
1988            if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
1989                self.flush()?;
1990            }
1991        }
1992        Ok(())
1993    }
1994
1995    /// Purge cache entries. If `namespace` is `Some`, restrict to that
1996    /// namespace; otherwise purge across the whole cache. If
1997    /// `expired_only` is `true`, only TTL-expired entries are removed;
1998    /// otherwise every live entry in the target scope is tombstoned.
1999    /// Returns the number of entries newly tombstoned.
2000    pub fn cache_purge(&mut self, namespace: Option<&str>, expired_only: bool) -> Result<usize> {
2001        if self.read_only {
2002            return Err(MnemoError::ReadOnly);
2003        }
2004        let now = memory::now_secs();
2005        let mut purged = 0usize;
2006        // Iterate the directory by index; mark matching entries as
2007        // deleted and pull them out of the in-memory index.
2008        for pos in 0..self.cache_dir.len() {
2009            if self.cache_dir[pos].deleted {
2010                continue;
2011            }
2012            if let Some(ns) = namespace {
2013                if self.cache_dir[pos].namespace != ns {
2014                    continue;
2015                }
2016            }
2017            if expired_only && !self.cache_dir[pos].is_expired(now) {
2018                continue;
2019            }
2020            let key_hash = self.cache_dir[pos].key_hash;
2021            let ns = self.cache_dir[pos].namespace.clone();
2022            self.cache_dir[pos].deleted = true;
2023            self.cache_idx.remove(&ns, &key_hash);
2024            purged += 1;
2025        }
2026        if purged > 0 {
2027            self.dirty_cache = true;
2028            self.cache_batch.record_dirty();
2029            if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
2030                self.flush()?;
2031            }
2032        }
2033        Ok(purged)
2034    }
2035
2036    /// Statistics for `namespace` (or the whole cache if `None`).
2037    /// `hits`/`misses`/`evictions` are per-handle in-memory counters
2038    /// that reset on reopen; `entries`/`bytes` reflect the durable
2039    /// on-disk state (live entries only, tombstones excluded).
2040    pub fn cache_stats(&self, namespace: Option<&str>) -> CacheStats {
2041        let mut entries = 0usize;
2042        let mut bytes = 0u64;
2043        for e in &self.cache_dir {
2044            if e.deleted {
2045                continue;
2046            }
2047            if let Some(ns) = namespace {
2048                if e.namespace != ns {
2049                    continue;
2050                }
2051            }
2052            entries += 1;
2053            bytes += e.size_bytes;
2054        }
2055        let hits = self.cache_counters.hits;
2056        let misses = self.cache_counters.misses;
2057        let total = hits + misses;
2058        let hit_rate = if total > 0 {
2059            hits as f64 / total as f64
2060        } else {
2061            0.0
2062        };
2063        CacheStats {
2064            entries,
2065            bytes,
2066            hits,
2067            misses,
2068            hit_rate,
2069            evictions: self.cache_counters.evictions,
2070        }
2071    }
2072
2073    /// Set (or update) the eviction budget for `namespace`. Applies to
2074    /// subsequent `cache_put` calls; does not retroactively evict
2075    /// entries that already sit under a looser cap.
2076    pub fn set_cache_budget(&mut self, namespace: &str, budget: CacheBudget) {
2077        self.cache_budgets.insert(namespace.to_string(), budget);
2078    }
2079
2080    /// Set the per-handle cache flush policy. Defaults to
2081    /// [`CacheFlushPolicy::Strict`]. Memory writes are never batched;
2082    /// this affects only cache mutations.
2083    pub fn set_cache_flush_policy(&mut self, policy: CacheFlushPolicy) {
2084        self.cache_flush_policy = policy;
2085        // Switching mid-run: if we're moving to Strict and the batch is
2086        // non-empty, the pending count still counts toward dirty_cache
2087        // which the next flush() will consume.
2088    }
2089
2090    /// Evict LRU live entries in `namespace` until it fits under both
2091    /// budget caps. Called from `cache_put` after the insert; the new
2092    /// entry is included in the count so we always leave room.
2093    fn evict_to_budget(&mut self, namespace: &str) -> Result<()> {
2094        let budget = self
2095            .cache_budgets
2096            .get(namespace)
2097            .copied()
2098            .unwrap_or_default();
2099        loop {
2100            // Sum current live count + bytes in the namespace.
2101            let mut live_count = 0usize;
2102            let mut live_bytes = 0u64;
2103            for e in &self.cache_dir {
2104                if !e.deleted && e.namespace == namespace {
2105                    live_count += 1;
2106                    live_bytes += e.size_bytes;
2107                }
2108            }
2109            if live_count <= budget.max_entries && live_bytes <= budget.max_bytes {
2110                return Ok(());
2111            }
2112            // Find the LRU (oldest accessed_at) live entry.
2113            let victim_pos = self
2114                .cache_dir
2115                .iter()
2116                .enumerate()
2117                .filter(|(_, e)| !e.deleted && e.namespace == namespace)
2118                .min_by_key(|(_, e)| e.accessed_at)
2119                .map(|(i, _)| i);
2120            let Some(pos) = victim_pos else {
2121                // Nothing to evict but still over budget — the single
2122                // just-inserted entry itself is bigger than
2123                // max_bytes. Nothing we can safely do here without
2124                // rejecting the put; leave it and return.
2125                return Ok(());
2126            };
2127            let key_hash = self.cache_dir[pos].key_hash;
2128            let ns = self.cache_dir[pos].namespace.clone();
2129            self.cache_dir[pos].deleted = true;
2130            self.cache_idx.remove(&ns, &key_hash);
2131            self.cache_counters.evictions += 1;
2132        }
2133    }
2134
2135    // === Semantic cache (Phase 10.2) ==================================
2136
2137    /// Insert or overwrite a **semantic** cache entry — same as
2138    /// [`Mnemo::cache_put`] but with a required embedding vector and
2139    /// model identifier. The vector is stored on the directory entry
2140    /// so [`Mnemo::cache_get_semantic`] can score without decrypting
2141    /// record bodies.
2142    ///
2143    /// Vector dimensionality is validated against the database's
2144    /// configured `dimensions` — a mismatch returns
2145    /// [`MnemoError::DimensionMismatch`], the same signal `remember`
2146    /// gives for the memory catalog.
2147    ///
2148    /// The `model` string is matched **exactly** on lookup — a
2149    /// different model's cache is invisible to a given query. Pick
2150    /// canonical identifiers (`"text-embedding-3-small"`,
2151    /// `"bge-large-en-v1.5"`, ...) and stick to them.
2152    pub fn cache_put_semantic(
2153        &mut self,
2154        namespace: &str,
2155        key: &str,
2156        vector: Vec<f32>,
2157        value: &[u8],
2158        opts: SemanticCachePutOpts,
2159    ) -> Result<()> {
2160        if self.read_only {
2161            return Err(MnemoError::ReadOnly);
2162        }
2163        if vector.len() != self.dimensions {
2164            return Err(MnemoError::DimensionMismatch {
2165                expected: self.dimensions,
2166                got: vector.len(),
2167            });
2168        }
2169        let key_hash = result_cache::hash_key(key);
2170        let key_preview = result_cache::key_preview(key);
2171        let now = memory::now_secs();
2172        let size_bytes = value.len() as u64;
2173
2174        let entry = CacheEntry {
2175            namespace: namespace.to_string(),
2176            key_hash,
2177            key_preview: key_preview.clone(),
2178            value: value.to_vec(),
2179            content_type: opts.content_type.clone(),
2180            created_at: now,
2181            accessed_at: now,
2182            access_count: 0,
2183            ttl_secs: opts.ttl_secs,
2184            cost_hint_ms: opts.cost_hint_ms,
2185            size_bytes,
2186            vector: Some(vector.clone()),
2187            model: Some(opts.model.clone()),
2188        };
2189        let bytes = rmp_serde::to_vec(&entry).map_err(|e| MnemoError::Serialize(e.to_string()))?;
2190        let (start_page, page_count) = self.write_record(&bytes)?;
2191
2192        if let Some(prev_pos) = self.cache_idx.get(namespace, &key_hash) {
2193            self.cache_dir[prev_pos].deleted = true;
2194            self.cache_idx.remove(namespace, &key_hash);
2195        }
2196
2197        let dir_entry = CacheDirectoryEntry {
2198            namespace: namespace.to_string(),
2199            key_hash,
2200            start_page,
2201            page_count,
2202            len: bytes.len() as u32,
2203            deleted: false,
2204            accessed_at: now,
2205            access_count: 0,
2206            size_bytes,
2207            ttl_secs: opts.ttl_secs,
2208            created_at: now,
2209            vector: Some(vector),
2210            model: Some(opts.model),
2211        };
2212        let pos = self.cache_dir.len();
2213        self.cache_dir.push(dir_entry);
2214        self.cache_idx.insert(namespace, key_hash, pos);
2215
2216        self.evict_to_budget(namespace)?;
2217
2218        self.dirty_cache = true;
2219        self.cache_batch.record_dirty();
2220        if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
2221            self.flush()?;
2222        }
2223        Ok(())
2224    }
2225
2226    /// Semantic recall over the cache: score every live entry in
2227    /// `namespace` whose `model` matches against `query_vector` using
2228    /// cosine similarity; return the top-1 hit and its similarity iff
2229    /// `sim >= threshold`. `None` otherwise.
2230    ///
2231    /// Uses [`crate::result_cache::DEFAULT_SEMANTIC_THRESHOLD`] if you
2232    /// want a sensible default — pass it explicitly (`0.97` today).
2233    ///
2234    /// Exact-key entries (those without a vector) are transparently
2235    /// skipped, as are entries whose model doesn't match — a
2236    /// different-model hit would be a semantic bug, not a win.
2237    /// TTL-expired entries are also skipped.
2238    ///
2239    /// Access-stat updates on hit follow the same policy as
2240    /// [`Mnemo::cache_get`]: bumped on read-write handles, silently
2241    /// skipped on read-only. Hits and misses are recorded in the
2242    /// per-handle counters just like exact-key lookups.
2243    pub fn cache_get_semantic(
2244        &mut self,
2245        namespace: &str,
2246        query_vector: &[f32],
2247        threshold: f32,
2248        model: &str,
2249    ) -> Result<Option<(CachedValue, f32)>> {
2250        if query_vector.len() != self.dimensions {
2251            return Err(MnemoError::DimensionMismatch {
2252                expected: self.dimensions,
2253                got: query_vector.len(),
2254            });
2255        }
2256        let now = memory::now_secs();
2257
2258        // Scan namespace live entries, filter by model, compute
2259        // cosine, keep the best. Exact-key entries have `vector: None`
2260        // and are skipped by the filter.
2261        let mut best: Option<(usize, f32)> = None;
2262        for (i, e) in self.cache_dir.iter().enumerate() {
2263            if e.deleted || e.namespace != namespace || e.is_expired(now) {
2264                continue;
2265            }
2266            let Some(ref v) = e.vector else { continue };
2267            let Some(ref m) = e.model else { continue };
2268            if m != model {
2269                continue;
2270            }
2271            let sim = memory::cosine(query_vector, v);
2272            if best.map(|(_, s)| sim > s).unwrap_or(true) {
2273                best = Some((i, sim));
2274            }
2275        }
2276
2277        let Some((pos, sim)) = best else {
2278            self.cache_counters.misses += 1;
2279            return Ok(None);
2280        };
2281        if sim < threshold {
2282            self.cache_counters.misses += 1;
2283            return Ok(None);
2284        }
2285
2286        // Extract fields before any mutation to keep the borrow checker
2287        // happy.
2288        let (start_page, page_count, len, ttl_secs, created_at) = {
2289            let e = &self.cache_dir[pos];
2290            (e.start_page, e.page_count, e.len, e.ttl_secs, e.created_at)
2291        };
2292
2293        // Decrypt and decode the record body.
2294        let mut buf = Vec::with_capacity(len as usize);
2295        for i in 0..page_count as u64 {
2296            buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
2297        }
2298        buf.truncate(len as usize);
2299        let entry: CacheEntry =
2300            rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
2301
2302        if !self.read_only {
2303            self.cache_dir[pos].accessed_at = now;
2304            self.cache_dir[pos].access_count = self.cache_dir[pos].access_count.saturating_add(1);
2305            self.dirty_cache = true;
2306        }
2307        self.cache_counters.hits += 1;
2308        let access_count = self.cache_dir[pos].access_count;
2309        Ok(Some((
2310            CachedValue {
2311                value: entry.value,
2312                content_type: entry.content_type,
2313                created_at,
2314                accessed_at: if self.read_only {
2315                    entry.accessed_at
2316                } else {
2317                    now
2318                },
2319                access_count,
2320                ttl_secs,
2321            },
2322            sim,
2323        )))
2324    }
2325
2326    /// Begin a conversation [`Session`](crate::Session) for `agent_id`.
2327    ///
2328    /// The session borrows the database for its lifetime, records turns as
2329    /// working memory, and consolidates them into episodic memory when closed.
2330    pub fn session(&mut self, agent_id: impl Into<String>) -> crate::session::Session<'_> {
2331        crate::session::Session::new(self, agent_id.into())
2332    }
2333
2334    // --- snapshots & point-in-time recovery ------------------------------
2335
2336    /// Every committed transaction, oldest first — the restore points
2337    /// available to [`Mnemo::restore_to`] and [`Mnemo::restore_to_time`].
2338    ///
2339    /// Each `flush` appends one snapshot. Because the storage engine is
2340    /// append-only, the pages a past flush wrote are still on disk, so any
2341    /// listed snapshot can be reinstated exactly. The history reaches back to
2342    /// the last [`Mnemo::compact_file`], which reclaims space by collapsing
2343    /// it.
2344    pub fn snapshots(&self) -> Vec<SnapshotInfo> {
2345        let mut v: Vec<SnapshotInfo> = self
2346            .manifest
2347            .iter()
2348            .map(|s| SnapshotInfo {
2349                txn_id: s.txn_id,
2350                created_at: s.created_at,
2351                memory_count: s.memory_count,
2352            })
2353            .collect();
2354        v.sort_by_key(|s| s.txn_id);
2355        v
2356    }
2357
2358    /// Load a past snapshot's state and commit it as a new transaction.
2359    fn apply_snapshot(&mut self, snap: &Snapshot) -> Result<()> {
2360        let catalog = load_catalog(
2361            &mut self.pager,
2362            snap.catalog_start,
2363            snap.catalog_pages,
2364            snap.catalog_len,
2365        )?;
2366        let ann = load_index(
2367            &mut self.pager,
2368            snap.index_start,
2369            snap.index_pages,
2370            snap.index_len,
2371            self.dimensions,
2372        )?;
2373        self.index = build_id_index(&catalog);
2374        self.catalog = catalog;
2375        self.ann = ann;
2376        // Re-commit as a fresh transaction: crash-safe, and itself recorded
2377        // as a new snapshot so a restore can always be undone.
2378        self.dirty_catalog = true;
2379        self.dirty_index = true;
2380        self.flush()
2381    }
2382
2383    /// Restore the database to the snapshot produced by transaction `txn_id`.
2384    ///
2385    /// The restore is itself a new committed transaction (and a new
2386    /// snapshot), so it is crash-safe and reversible — restoring forward to a
2387    /// later snapshot afterwards works just as well.
2388    pub fn restore_to(&mut self, txn_id: u64) -> Result<SnapshotInfo> {
2389        if self.read_only {
2390            return Err(MnemoError::ReadOnly);
2391        }
2392        let snap = self
2393            .manifest
2394            .iter()
2395            .find(|s| s.txn_id == txn_id)
2396            .cloned()
2397            .ok_or_else(|| MnemoError::NotFound(format!("snapshot for transaction {txn_id}")))?;
2398        self.apply_snapshot(&snap)?;
2399        Ok(SnapshotInfo {
2400            txn_id: snap.txn_id,
2401            created_at: snap.created_at,
2402            memory_count: snap.memory_count,
2403        })
2404    }
2405
2406    /// Restore the database to the latest snapshot committed at or before
2407    /// `unix_secs`. Returns [`MnemoError::NotFound`] if no snapshot is that
2408    /// old. Like [`Mnemo::restore_to`], the restore is a new transaction.
2409    pub fn restore_to_time(&mut self, unix_secs: i64) -> Result<SnapshotInfo> {
2410        if self.read_only {
2411            return Err(MnemoError::ReadOnly);
2412        }
2413        let snap = self
2414            .manifest
2415            .iter()
2416            .filter(|s| s.created_at <= unix_secs)
2417            .max_by_key(|s| s.txn_id)
2418            .cloned()
2419            .ok_or_else(|| {
2420                MnemoError::NotFound(format!("no snapshot at or before time {unix_secs}"))
2421            })?;
2422        self.apply_snapshot(&snap)?;
2423        Ok(SnapshotInfo {
2424            txn_id: snap.txn_id,
2425            created_at: snap.created_at,
2426            memory_count: snap.memory_count,
2427        })
2428    }
2429
2430    /// Change the passphrase. Cheap: re-derives the KEK and re-wraps the DEK;
2431    /// the encrypted pages are never rewritten.
2432    pub fn rekey(&mut self, new_passphrase: &str, kdf: KdfParams) -> Result<()> {
2433        if self.read_only {
2434            return Err(MnemoError::ReadOnly);
2435        }
2436        self.flush()?;
2437        let new_salt = crypto::random_salt();
2438        let new_kek = crypto::derive_kek(new_passphrase.as_bytes(), &new_salt, kdf)?;
2439        let new_nonce = crypto::random_nonce();
2440        let wrapped = crypto::wrap_dek(&new_kek, &new_nonce, self.pager.dek())?;
2441        if wrapped.len() != WRAPPED_DEK_LEN {
2442            return Err(MnemoError::Crypto("unexpected wrapped DEK length".into()));
2443        }
2444        let mut wrapped_dek = [0u8; WRAPPED_DEK_LEN];
2445        wrapped_dek.copy_from_slice(&wrapped);
2446
2447        self.header.salt = new_salt;
2448        self.header.dek_nonce = new_nonce;
2449        self.header.wrapped_dek = wrapped_dek;
2450        self.header.m_cost = kdf.m_cost;
2451        self.header.t_cost = kdf.t_cost;
2452        self.header.p_cost = kdf.p_cost;
2453        self.header.write_counter = self.pager.write_counter;
2454        self.kdf = kdf;
2455
2456        let mut page = self.header.to_page();
2457        self.header.apply_seal(&mut page, self.pager.dek())?;
2458        self.pager.write_raw(0, &page)?;
2459        self.pager.sync()?;
2460        Ok(())
2461    }
2462
2463    /// Decrypt and validate every live record. Returns the count verified.
2464    pub fn verify(&mut self) -> Result<usize> {
2465        let entries: Vec<CatalogEntry> = self
2466            .catalog
2467            .iter()
2468            .filter(|e| !e.deleted)
2469            .cloned()
2470            .collect();
2471        let n = entries.len();
2472        for e in &entries {
2473            let bytes = self.read_record(e)?;
2474            let _m: Memory = rmp_serde::from_slice(&bytes)
2475                .map_err(|err| MnemoError::Serialize(err.to_string()))?;
2476        }
2477        Ok(n)
2478    }
2479
2480    /// Summary statistics for the open database.
2481    pub fn stats(&mut self) -> Result<Stats> {
2482        let deleted = self.catalog.iter().filter(|e| e.deleted).count();
2483        let mut agents: Vec<String> = self.memories()?.into_iter().map(|m| m.agent_id).collect();
2484        agents.sort();
2485        agents.dedup();
2486        let file_bytes = self.header.next_page * PAGE_SIZE as u64;
2487        Ok(Stats {
2488            memories: self.len(),
2489            deleted,
2490            dimensions: self.dimensions,
2491            file_bytes,
2492            agents,
2493            encrypted: self.header.flags & FLAG_ENCRYPTED != 0,
2494            created_at: self.header.created_at,
2495            index: self.ann.as_ref().map(|a| a.info()),
2496            wal_pages: self.header.wal_pages,
2497        })
2498    }
2499
2500    /// Rewrite the file, dropping tombstoned and expired memories and
2501    /// reclaiming stale pages left by updates. Done out-of-place via a temp
2502    /// file and an atomic rename.
2503    pub fn compact_file(path: &str, passphrase: &str) -> Result<CompactReport> {
2504        let mut old = Mnemo::open(path, passphrase)?;
2505        let dims = old.dimensions;
2506        let kdf = old.kdf;
2507        let tmp = format!("{path}.compact-tmp");
2508
2509        let mut new = Mnemo::create(
2510            &tmp,
2511            passphrase,
2512            MnemoConfig {
2513                dimensions: dims,
2514                kdf,
2515                ..Default::default()
2516            },
2517        )?;
2518        let want_index = old.ann.as_ref().map(|a| (a.n_probe(), a.n_rerank()));
2519        let now = memory::now_secs();
2520        let all = old.memories()?;
2521        let before = all.len();
2522        let mut after = 0;
2523        for m in all {
2524            if m.is_expired(now) {
2525                continue;
2526            }
2527            new.remember(m)?;
2528            after += 1;
2529        }
2530        // Rebuild the index fresh (re-clustered) when the source had one.
2531        if let Some((n_probe, n_rerank)) = want_index {
2532            if after > 0 {
2533                new.build_index_with(IndexConfig {
2534                    n_probe,
2535                    n_rerank,
2536                    ..Default::default()
2537                })?;
2538            }
2539        }
2540        new.flush()?;
2541        drop(new);
2542        drop(old);
2543
2544        std::fs::rename(&tmp, path)?;
2545        Ok(CompactReport { before, after })
2546    }
2547}