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