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