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}