Skip to main content

mnemo_engine/
store.rs

1//! The Mnemo database — ties the storage engine and the memory model together.
2//!
3//! Layout of a populated file:
4//!
5//! ```text
6//!   page 0           : Header (unencrypted)
7//!   pages 1..W       : write-ahead log region
8//!   pages W..        : encrypted record runs + catalog runs + index +
9//!                      snapshot manifest (append-only)
10//! ```
11//!
12//! Durability is provided by a **write-ahead log** ([`crate::wal`]). A `flush`
13//! is one transaction: record data pages are written copy-on-write to fresh
14//! pages, then the new catalog, ANN index, and header are logged to the WAL
15//! and committed with a single fsync. A checkpoint then folds the WAL into the
16//! home pages. A crash before the commit leaves the previous state intact; a
17//! crash after it is repaired by replaying the WAL on open.
18//!
19//! Because pages are only ever appended, every past transaction's runs survive
20//! on disk. Each flush also appends an entry to a **snapshot manifest**, so
21//! [`Mnemo::restore_to`] can reinstate any past state. Stale pages — and the
22//! history along with them — are reclaimed by [`Mnemo::compact_file`].
23
24use std::collections::HashMap;
25use std::fs::{File, OpenOptions};
26use std::io::{Read, Seek, SeekFrom, Write};
27use std::path::{Path, PathBuf};
28
29use fs4::fs_std::FileExt;
30use serde::{Deserialize, Serialize};
31use 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, KeyScope, Memory, MemoryPatch, MemoryType, Metric, Scope, ScoreWeights,
41    WriteOutcome, MNEMO_META_INVALID_AT, MNEMO_META_KEY, MNEMO_META_SUPERSEDED_BY,
42    MNEMO_META_SUPERSEDES, 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    /// **Slot resolution (v0.8.1 bugfix, lane A).** The lookup slot
2085    /// matches a keyed `remember` with the same arguments, so the
2086    /// obvious `history(key, None)` walks the private slot of the
2087    /// default agent — the same slot
2088    /// `remember_keyed(Memory::new(...), key)` writes to when the
2089    /// caller omits an explicit `agent_id` / `scope`. Pre-v0.8.1
2090    /// `agent_id = None` incorrectly looked up the shared slot and
2091    /// returned an empty chain for the common case; the old behavior
2092    /// is still reachable via `history_in(key, KeyScope::Shared)`.
2093    ///
2094    /// * `agent_id = Some(a)` → private slot for that agent.
2095    /// * `agent_id = None` → private slot for the default agent (same
2096    ///   as [`KeyScope::private_default`]).
2097    ///
2098    /// Use [`Mnemo::history_in`] with [`KeyScope::Shared`] for the
2099    /// shared slot.
2100    pub fn history(&mut self, key: &str, agent_id: Option<&str>) -> Result<Vec<Memory>> {
2101        let scope = match agent_id {
2102            Some(a) => KeyScope::Private(a.to_string()),
2103            None => KeyScope::private_default(),
2104        };
2105        self.history_in(key, scope)
2106    }
2107
2108    /// Walk the full chain for a keyed slot identified by an explicit
2109    /// [`KeyScope`]. Returns the same shape as [`Mnemo::history`] —
2110    /// live record first, then the superseded chain newest-first.
2111    ///
2112    /// Use this when you need the **shared** slot (`KeyScope::Shared`),
2113    /// which `history(key, None)` no longer reaches as of v0.8.1.
2114    pub fn history_in(&mut self, key: &str, scope: KeyScope) -> Result<Vec<Memory>> {
2115        self.ensure_key_caches()?;
2116        let slot = crate::store::KeySlot {
2117            scope_owner: match scope {
2118                KeyScope::Private(a) => Some(a),
2119                KeyScope::Shared => None,
2120            },
2121            key: key.to_string(),
2122        };
2123        let live_id = self.key_index.as_ref().and_then(|k| k.get(&slot).copied());
2124        let mut out = Vec::new();
2125        let mut cursor = live_id;
2126        while let Some(id_u) = cursor {
2127            let idx = *self.index.get(&id_u).ok_or_else(|| {
2128                MnemoError::Invalid("history: superseded chain references missing id".into())
2129            })?;
2130            let entry = self.catalog[idx].clone();
2131            if entry.deleted {
2132                break;
2133            }
2134            let m = self.read_memory(&entry)?;
2135            let next = m
2136                .metadata
2137                .get(MNEMO_META_SUPERSEDES)
2138                .and_then(|v| v.as_str())
2139                .and_then(|s| Ulid::from_string(s).ok())
2140                .map(|u| u.0);
2141            out.push(m);
2142            cursor = next;
2143        }
2144        Ok(out)
2145    }
2146
2147    /// Return every live memory (used by tooling and compaction).
2148    pub fn memories(&mut self) -> Result<Vec<Memory>> {
2149        let entries: Vec<CatalogEntry> = self
2150            .catalog
2151            .iter()
2152            .filter(|e| !e.deleted)
2153            .cloned()
2154            .collect();
2155        let mut out = Vec::with_capacity(entries.len());
2156        for e in &entries {
2157            out.push(self.read_memory(e)?);
2158        }
2159        Ok(out)
2160    }
2161
2162    /// Return the database's **self-describing onboarding memories** — the
2163    /// ones tagged `metadata.area = "onboarding"`. This is the engine-level
2164    /// surface for the *single-file philosophy*: an agent who receives a
2165    /// `.mnemo` file (and its passphrase) can call this to learn what the
2166    /// file is, which embedder it expects, the recommended agent id, and any
2167    /// other conventions the file's author chose to record — all without
2168    /// needing any external documentation.
2169    ///
2170    /// Ordering: the canonical manifest (tag `metadata.topic = "manifest"`)
2171    /// always comes first; everything else follows in `importance` descending,
2172    /// then `created_at` ascending for deterministic results.
2173    pub fn about(&mut self) -> Result<Vec<Memory>> {
2174        // Phase 12.1: drop records that have been superseded by a newer
2175        // keyed write. They still live in snapshot history and
2176        // `history(key)` — just not here.
2177        let mut out: Vec<Memory> = self
2178            .memories()?
2179            .into_iter()
2180            .filter(|m| !m.metadata.contains_key(MNEMO_META_SUPERSEDED_BY))
2181            .filter(|m| {
2182                m.metadata
2183                    .get("area")
2184                    .and_then(|v| v.as_str())
2185                    .map(|s| s.eq_ignore_ascii_case("onboarding"))
2186                    .unwrap_or(false)
2187            })
2188            .collect();
2189        out.sort_by(|a, b| {
2190            let a_manifest = is_manifest(a);
2191            let b_manifest = is_manifest(b);
2192            // Manifest topic wins first; then importance desc; then created_at asc.
2193            b_manifest
2194                .cmp(&a_manifest)
2195                .then_with(|| b.importance.total_cmp(&a.importance))
2196                .then_with(|| a.created_at.cmp(&b.created_at))
2197        });
2198        Ok(out)
2199    }
2200
2201    // --- retrieval --------------------------------------------------------
2202
2203    /// Phase-1 brute-force search: rank live memories by raw similarity only.
2204    /// Read-only — does not update access statistics.
2205    pub fn search(
2206        &mut self,
2207        query: &[f32],
2208        top_k: usize,
2209        metric: Metric,
2210    ) -> Result<Vec<(Memory, f32)>> {
2211        if query.len() != self.dimensions {
2212            return Err(MnemoError::DimensionMismatch {
2213                expected: self.dimensions,
2214                got: query.len(),
2215            });
2216        }
2217        let now = memory::now_secs();
2218        let entries: Vec<CatalogEntry> = self
2219            .catalog
2220            .iter()
2221            .filter(|e| !e.deleted)
2222            .cloned()
2223            .collect();
2224
2225        let mut scored: Vec<(Memory, f32)> = Vec::new();
2226        for e in &entries {
2227            let m = self.read_memory(e)?;
2228            if m.is_expired(now) {
2229                continue;
2230            }
2231            let sim = memory::similarity(metric, query, &m.vector);
2232            scored.push((m, sim));
2233        }
2234        scored.sort_by(|a, b| b.1.total_cmp(&a.1));
2235        scored.truncate(top_k);
2236        Ok(scored)
2237    }
2238
2239    /// Phase-5 multi-signal recall: rank by
2240    /// `α·similarity + β·recency + γ·importance + δ·ln(freq)`, with type and
2241    /// agent filtering. Updates `accessed_at` / `access_count` on the
2242    /// returned memories (persisted on the next `flush`).
2243    ///
2244    /// When an ANN index has been built ([`Mnemo::build_index`]), recall runs
2245    /// the tiered IVF→PQ→rerank pipeline: it scores only the similarity-nearest
2246    /// `n_rerank` candidates rather than every memory. Without an index it
2247    /// scores every live memory exactly.
2248    /// One pass of filter + score for [`Mnemo::recall`]. Factored out of
2249    /// `recall` so the outer path can retry with a wider shortlist when
2250    /// the type/agent/supersession/TTL filters starve the ANN-narrowed
2251    /// result below `top_k` (Phase 12.1 short-results-on-indexed-path
2252    /// fix).
2253    ///
2254    /// * `force_exact = false` → use the ANN index when one exists; fall
2255    ///   back to a full scan only if none is built.
2256    /// * `force_exact = true` → skip the ANN index and scan every live
2257    ///   non-superseded record. Used by the outer path as the last-resort
2258    ///   attempt when widening the shortlist would already touch half or
2259    ///   more of the live non-superseded set.
2260    ///
2261    /// Returns an unsorted, un-truncated `Vec<RecallResult>`; the outer
2262    /// `recall` sorts and truncates once.
2263    fn score_once(
2264        &mut self,
2265        req: &RecallRequest,
2266        now: i64,
2267        force_exact: bool,
2268    ) -> Result<Vec<RecallResult>> {
2269        // Candidate set: ANN-narrowed when an index exists and the
2270        // caller hasn't forced an exact scan, else everything live.
2271        let entries: Vec<CatalogEntry> = match (&self.ann, force_exact) {
2272            (Some(ann), false) => {
2273                let ids = ann.query(&req.query, req.n_probe, req.n_rerank);
2274                ids.iter()
2275                    .filter_map(|id| self.index.get(id).copied())
2276                    .map(|i| self.catalog[i].clone())
2277                    .filter(|e| !e.deleted)
2278                    .collect()
2279            }
2280            _ => self
2281                .catalog
2282                .iter()
2283                .filter(|e| !e.deleted)
2284                .cloned()
2285                .collect(),
2286        };
2287
2288        let mut scored: Vec<RecallResult> = Vec::with_capacity(entries.len());
2289        for e in &entries {
2290            // Phase 12.1: drop superseded records.
2291            //
2292            // Fast path: if the lazy `superseded` set has already been
2293            // built by a prior keyed op on this handle, check it before
2294            // decrypt — O(1), no AES work. On a fresh read-only handle
2295            // it is typically `None`, so we fall through to the slow
2296            // path (one `_mnemo.superseded_by` lookup per surviving
2297            // record, O(k) over user metadata), and never trigger the
2298            // full-file scan that `ensure_key_caches()` would do.
2299            if self.superseded.as_ref().is_some_and(|s| s.contains(&e.id)) {
2300                continue;
2301            }
2302            let m = self.read_memory(e)?;
2303            if m.metadata.contains_key(MNEMO_META_SUPERSEDED_BY) {
2304                continue;
2305            }
2306            if m.is_expired(now) {
2307                continue;
2308            }
2309            if let Some(types) = &req.memory_types {
2310                if !types.contains(&m.memory_type) {
2311                    continue;
2312                }
2313            }
2314            if let Some(agent) = &req.agent_id {
2315                let visible = m.agent_id == *agent || m.scope == Scope::Shared;
2316                if !visible {
2317                    continue;
2318                }
2319            }
2320            // Phase 12.6: pre-scoring filter (session_id + metadata
2321            // conditions). Cheap when `req.filter.is_empty()`; the
2322            // filter struct's `matches` is a short-circuit AND.
2323            if !req.filter.is_empty() && !req.filter.matches(&m) {
2324                continue;
2325            }
2326            let sim = memory::similarity(req.metric, &req.query, &m.vector);
2327            let age_secs = (now - m.created_at) as f32;
2328            let score = req
2329                .weights
2330                .score(sim, age_secs, m.importance, m.access_count);
2331            scored.push(RecallResult {
2332                memory: m,
2333                score,
2334                similarity: sim,
2335            });
2336        }
2337        Ok(scored)
2338    }
2339
2340    /// Phase-5 multi-signal recall under the v0.7.0 scoring-v2 formula
2341    /// (see [`crate::ScoreWeights`]). Thin wrapper over
2342    /// [`Mnemo::recall_budgeted`] that drops the `truncated` flag;
2343    /// use the budgeted form if you need to know whether
2344    /// [`RecallRequest::max_bytes`] cut anything.
2345    pub fn recall(&mut self, req: &RecallRequest) -> Result<Vec<RecallResult>> {
2346        Ok(self.recall_budgeted(req)?.hits)
2347    }
2348
2349    /// Phase 12.8 budgeted recall: scoring v2 (see [`crate::ScoreWeights`])
2350    /// with an optional UTF-8-content byte cap. Ranks by
2351    /// `sim + max(sim, 0) · (β·recency + γ·(importance − 0.5) + δ·tanh(uses/5))`,
2352    /// with type / agent / supersession / TTL filters and the 12.6
2353    /// `Filter` applied before the scorer. Updates `accessed_at` /
2354    /// `access_count` on the returned memories (persisted on the next
2355    /// `flush`).
2356    ///
2357    /// When an ANN index has been built ([`Mnemo::build_index`]),
2358    /// recall runs the tiered IVF→PQ→rerank pipeline. If the filters
2359    /// (type / agent / supersession / TTL / 12.6) reduce the ANN
2360    /// shortlist below `top_k`, Phase 12.1's short-results-on-indexed-
2361    /// path fix kicks in: the query is retried with `n_probe` and
2362    /// `n_rerank` doubled, and if that would already touch half or
2363    /// more of the live non-superseded records, the ANN pass is
2364    /// skipped for an exact scan. Without an index, recall scores
2365    /// every live non-superseded memory.
2366    ///
2367    /// Phase 12.8: after `top_k` is applied, if
2368    /// [`RecallRequest::max_bytes`] is set, hits are dropped in rank
2369    /// order once the running sum of `content` UTF-8 bytes would
2370    /// exceed the budget. The top-1 hit is always kept even if it
2371    /// alone exceeds the budget. [`RecallResponse::truncated`] is
2372    /// `true` iff at least one hit was dropped by the budget.
2373    pub fn recall_budgeted(&mut self, req: &RecallRequest) -> Result<RecallResponse> {
2374        if req.query.len() != self.dimensions {
2375            return Err(MnemoError::DimensionMismatch {
2376                expected: self.dimensions,
2377                got: req.query.len(),
2378            });
2379        }
2380        // Access-tracking bumps `accessed_at` / `access_count` in the
2381        // catalog, which is a mutation. On a read-only handle the caller
2382        // must pass `track_access(false)` (via CLI `--no-track`).
2383        if self.read_only && req.track_access {
2384            return Err(MnemoError::ReadOnly);
2385        }
2386        // Phase 12.1 **fix** (lane A v0.8.0 REVIEW): do NOT call
2387        // `ensure_key_caches()` here. A fresh read-only handle that
2388        // only ever recalls would otherwise pay a full-file decrypt
2389        // on its first query to build the lazy superseded set. Instead
2390        // `score_once` checks `_mnemo.superseded_by` per-record after
2391        // decrypt (slow path) and consults the set only when it has
2392        // already been built by a prior keyed op on this handle (fast
2393        // path). `live_non_superseded` is not knowable without a
2394        // decrypt pass either, so we approximate by `live_count`; the
2395        // exact-scan gate uses this as a conservative upper bound,
2396        // which errs toward widening (cheaper) rather than skipping
2397        // straight to a full scan.
2398        let live_count = self.catalog.iter().filter(|e| !e.deleted).count();
2399        let live_non_superseded = live_count;
2400        let now = memory::now_secs();
2401
2402        // Scoring v2 (shipped v0.7.0). Single-pass filter + score:
2403        //   * `age_secs = now - m.created_at`, so a recall doesn't bump
2404        //     its own results' recency on later queries.
2405        //   * `ScoreWeights::score` applies the similarity-gated,
2406        //     importance-centered, saturating-usage formula documented
2407        //     on `ScoreWeights`.
2408        //
2409        // Phase 12.1 + 12.6 short-results-on-indexed-path fix: when the
2410        // ANN shortlist is narrower than the live count, the filters
2411        // (type / agent / supersession / TTL / 12.6) can reduce the
2412        // result below `top_k`. If that happens, re-query the index
2413        // with `n_probe` / `n_rerank` doubled from their *effective*
2414        // base; if that would already cover ≥50% of the live set,
2415        // skip the index and do an exact scan.
2416        //
2417        // Lane A v0.8.0 REVIEW fix: widen from
2418        // `IvfPqIndex::effective_defaults()` when the caller didn't
2419        // pin the knobs. The pre-fix code only doubled when `req.n_*`
2420        // was already `Some(_)`, which skipped the widen step for the
2421        // default (unpinned) recall path and jumped straight to an
2422        // exact scan on moderately selective filters.
2423        let mut scored = self.score_once(req, now, false)?;
2424        let mut used_exact_scan = self.ann.is_none();
2425        if scored.len() < req.top_k && live_non_superseded > req.top_k {
2426            if let Some(ann) = &self.ann {
2427                let (default_probe, default_rerank) = ann.effective_defaults();
2428                let base_probe = req.n_probe.unwrap_or(default_probe);
2429                let base_rerank = req.n_rerank.unwrap_or(default_rerank);
2430                let widen_probe = base_probe.saturating_mul(2);
2431                let widen_rerank = base_rerank.saturating_mul(2);
2432                // Attempt 2: widen once, 2× from the effective base.
2433                // Only worth another ANN pass if either knob actually
2434                // grew (saturating_mul on usize::MAX would cap).
2435                if widen_probe > base_probe || widen_rerank > base_rerank {
2436                    let widened = RecallRequest {
2437                        n_probe: Some(widen_probe),
2438                        n_rerank: Some(widen_rerank),
2439                        ..req.clone()
2440                    };
2441                    scored = self.score_once(&widened, now, false)?;
2442                }
2443                // Attempt 3: if still short after widening, pay the
2444                // exact scan — it's the only path that guarantees
2445                // `top_k` correct hits when the filter is more
2446                // selective than `widen_rerank / live_count`. Before
2447                // the fix this was gated on `widened_rerank * 2 >=
2448                // live_non_superseded`, which silently returned fewer
2449                // than `top_k` hits on selective filters that fell in
2450                // the gap. Exact scan is bounded (one decrypt per
2451                // live record); the previous widen is an optimization.
2452                if scored.len() < req.top_k {
2453                    scored = self.score_once(req, now, true)?;
2454                    used_exact_scan = true;
2455                }
2456            }
2457        }
2458        scored.sort_by(|a, b| b.score.total_cmp(&a.score));
2459        scored.truncate(req.top_k);
2460
2461        // Update access statistics for everything we surfaced — but only
2462        // mutate the catalog entry, not the full record on disk. Pre-v5
2463        // this loop called `self.put(r.memory.clone())`, which serialized
2464        // and rewrote every result (vector included) to fresh pages,
2465        // turning a top-K recall into K full record rewrites. Now the
2466        // catalog is the source of truth for these two fields and the
2467        // record body stays untouched until the next real edit.
2468        if req.track_access {
2469            for r in &mut scored {
2470                if let Some(&idx) = self.index.get(&r.memory.id.0) {
2471                    let entry = &mut self.catalog[idx];
2472                    entry.accessed_at = now;
2473                    entry.access_count = entry.access_count.saturating_add(1);
2474                    // Propagate the just-updated values onto the returned
2475                    // Memory so the caller sees the post-recall state.
2476                    r.memory.accessed_at = entry.accessed_at;
2477                    r.memory.access_count = entry.access_count;
2478                }
2479            }
2480            // Mark the catalog dirty so the next flush persists these
2481            // bumps. The records themselves are not dirty.
2482            if !scored.is_empty() {
2483                self.dirty_catalog = true;
2484            }
2485        }
2486
2487        // Phase 12.8: apply the `max_bytes` budget *after* tracking has
2488        // seen the full top-k list. Access stats record what the scorer
2489        // chose; the byte budget shouldn't suppress the signal.
2490        let truncated = if let Some(budget) = req.max_bytes {
2491            let orig_len = scored.len();
2492            let mut running = 0usize;
2493            let mut cut_at: Option<usize> = None;
2494            for (i, r) in scored.iter().enumerate() {
2495                let w = r.memory.content.len();
2496                if i == 0 {
2497                    // Always keep the top hit, even if it's over budget
2498                    // on its own. Avoid "I asked for top_k results and
2499                    // got zero" when a single memory exceeds the cap.
2500                    running += w;
2501                    continue;
2502                }
2503                if running.saturating_add(w) > budget {
2504                    cut_at = Some(i);
2505                    break;
2506                }
2507                running += w;
2508            }
2509            if let Some(idx) = cut_at {
2510                scored.truncate(idx);
2511                scored.len() < orig_len
2512            } else {
2513                false
2514            }
2515        } else {
2516            false
2517        };
2518
2519        Ok(RecallResponse {
2520            hits: scored,
2521            truncated,
2522            used_exact_scan,
2523        })
2524    }
2525
2526    // --- approximate index ------------------------------------------------
2527
2528    /// Build an IVF+PQ approximate-nearest-neighbour index over every live
2529    /// memory, using default tuning. After this, [`Mnemo::recall`] runs the
2530    /// tiered pipeline instead of an exact scan. Persisted on the next
2531    /// `flush`. Returns a snapshot of the index shape.
2532    pub fn build_index(&mut self) -> Result<IndexInfo> {
2533        self.build_index_with(IndexConfig::default())
2534    }
2535
2536    /// Build the ANN index with explicit tuning.
2537    pub fn build_index_with(&mut self, cfg: IndexConfig) -> Result<IndexInfo> {
2538        if self.read_only {
2539            return Err(MnemoError::ReadOnly);
2540        }
2541        let mems = self.memories()?;
2542        if mems.is_empty() {
2543            return Err(MnemoError::Invalid(
2544                "cannot build an index over an empty database".into(),
2545            ));
2546        }
2547        let items: Vec<(u128, &[f32])> =
2548            mems.iter().map(|m| (m.id.0, m.vector.as_slice())).collect();
2549        let idx = IvfPqIndex::build(self.dimensions, &items, cfg)?;
2550        let info = idx.info();
2551        self.ann = Some(idx);
2552        self.dirty_index = true;
2553        Ok(info)
2554    }
2555
2556    /// Rebuild the ANN index from scratch — re-clusters centroids and retrains
2557    /// the PQ codebook, undoing the cluster drift that accumulates as memories
2558    /// are inserted against fixed centroids.
2559    pub fn rebuild_index(&mut self) -> Result<IndexInfo> {
2560        // ANN-1.1: re-derive every parameter (partitions, subspaces,
2561        // n_probe, n_rerank) for the current corpus size by passing
2562        // `IndexConfig::default()`. Previously this freezed the
2563        // stored `(n_probe, n_rerank)` from the current index,
2564        // defeating the point of rebuilding on a grown corpus. If
2565        // you really want to pin specific values, call
2566        // `build_index_with(IndexConfig { … })` directly.
2567        self.build_index_with(IndexConfig::default())
2568    }
2569
2570    /// Drop the ANN index; recall reverts to exact scans. Persisted on flush.
2571    /// No-op on a read-only handle (signature is infallible — callers of
2572    /// `build_index` / `flush` on a read-only handle already get a typed
2573    /// `ReadOnly` error, which is the enforceable boundary).
2574    pub fn drop_index(&mut self) {
2575        if self.read_only {
2576            return;
2577        }
2578        if self.ann.is_some() {
2579            self.ann = None;
2580            self.dirty_index = true;
2581        }
2582    }
2583
2584    /// Whether an ANN index is currently loaded.
2585    pub fn has_index(&self) -> bool {
2586        self.ann.is_some()
2587    }
2588
2589    // --- durability & maintenance ----------------------------------------
2590
2591    /// Seal a serialized buffer into a fresh run of encrypted home pages and
2592    /// append a WAL frame for each. Returns the run's `(start_page, pages)`.
2593    /// The pages are *not* written to their home locations here — they go to
2594    /// the WAL and are folded in later by [`Mnemo::checkpoint`].
2595    fn seal_run(&mut self, bytes: &[u8], frames: &mut Vec<wal::Frame>) -> Result<(u64, u32)> {
2596        let pc = bytes.len().div_ceil(PAYLOAD).max(1);
2597        let start = self.header.next_page;
2598        self.header.next_page += pc as u64;
2599        for i in 0..pc {
2600            let lo = i * PAYLOAD;
2601            let hi = ((i + 1) * PAYLOAD).min(bytes.len());
2602            let page_no = start + i as u64;
2603            let sealed = self.pager.seal_page(page_no, &bytes[lo..hi])?;
2604            frames.push((page_no, sealed.to_vec()));
2605        }
2606        Ok((start, pc as u32))
2607    }
2608
2609    /// Grow the WAL region if `frame_pages` worth of frames would not fit.
2610    /// Called at the top of `flush`, where the WAL is always spent (every
2611    /// flush ends with a checkpoint), so relocating it cannot strand a
2612    /// committed transaction.
2613    fn ensure_wal_capacity(&mut self, frame_pages: usize) -> Result<()> {
2614        let need = wal::txn_byte_len(frame_pages, PAGE_SIZE);
2615        if need <= self.header.wal_pages * PAGE_SIZE as u64 {
2616            return Ok(());
2617        }
2618        // Allocate a new, larger region at the file tail (~1.5x headroom).
2619        let req = need.div_ceil(PAGE_SIZE as u64);
2620        let new_pages = (req + req / 2 + 4).max(DEFAULT_WAL_PAGES);
2621        let old_wal_start = self.header.wal_start;
2622        let new_start = self.header.next_page;
2623        self.header.next_page += new_pages;
2624        self.header.wal_start = new_start;
2625        self.header.wal_pages = new_pages;
2626        self.header.wal_seq += 1;
2627        // DUR-1 (R-002.1): WAL-log the header change into the *new* region
2628        // first, checkpoint page 0, then invalidate the old region. A crash
2629        // anywhere here either keeps the old state (if page 0 wasn't yet
2630        // overwritten) or recovers to the new state via the torn-header
2631        // scan in `Mnemo::open`, which prefers the highest-txn-id valid
2632        // committed header frame.
2633        self.relocate_header_through_wal(old_wal_start)
2634    }
2635
2636    /// Shared helper for every post-`create` page-0 commit: WAL-log the
2637    /// new header into the current WAL region, checkpoint page 0, and
2638    /// invalidate `old_wal_start` (the previous region, or the current
2639    /// region when it hasn't moved). The caller is expected to have
2640    /// mutated `self.header` to its target values *including a bumped
2641    /// `wal_seq`* (one more than the last on-disk `wal_seq`); this
2642    /// routine applies the seal, writes the frame + commit, folds page 0,
2643    /// then zeroes the old region's first frame header.
2644    ///
2645    /// If `old_wal_start == self.header.wal_start`, the invalidation is
2646    /// skipped — the region didn't move, so overwriting its first frame
2647    /// would destroy the just-committed one.
2648    fn relocate_header_through_wal(&mut self, old_wal_start: u64) -> Result<()> {
2649        self.header.write_counter = self.pager.write_counter;
2650        let mut page = self.header.to_page();
2651        self.header.apply_seal(&mut page, self.pager.dek())?;
2652        let txn_id = self.header.wal_seq;
2653        let frames = vec![(0u64, page.to_vec())];
2654        wal::commit(
2655            self.pager.file_mut(),
2656            self.header.wal_start,
2657            self.header.wal_pages,
2658            txn_id,
2659            &frames,
2660        )?;
2661        self.pager.write_raw(0, &page)?;
2662        self.pager.sync()?;
2663        if old_wal_start != self.header.wal_start {
2664            wal::invalidate_region(self.pager.file_mut(), old_wal_start)?;
2665        }
2666        Ok(())
2667    }
2668
2669    /// Fold a committed transaction's WAL frames into their home pages.
2670    fn checkpoint(&mut self, frames: &[wal::Frame]) -> Result<()> {
2671        for (page_no, bytes) in frames {
2672            let mut img = [0u8; PAGE_SIZE];
2673            img.copy_from_slice(bytes);
2674            self.pager.write_sealed(*page_no, &img)?;
2675        }
2676        self.pager.sync()?;
2677        Ok(())
2678    }
2679
2680    /// Persist all pending changes as one **write-ahead-logged transaction**.
2681    ///
2682    /// Sequence:
2683    ///
2684    /// 1. **Prepare.** Serialize the control plane (catalog, ANN index),
2685    ///    grow the WAL if needed, and **lease** counter and page slots —
2686    ///    bump `header.write_counter` and `header.next_page` by upper
2687    ///    bounds on this transaction's consumption and persist that
2688    ///    *leased* header to disk before any encrypted page is written.
2689    /// 2. **Data pages.** [`Pager::flush`] writes dirty record pages
2690    ///    copy-on-write under fresh nonces and fsyncs them. The orphan-page
2691    ///    nonce-reuse window (see [Phase 1.1 of the improvement plan])
2692    ///    is closed because the leased header on disk already records a
2693    ///    `write_counter` past anything this step can produce.
2694    /// 3. **Control plane.** Seal the new catalog / ANN index / snapshot
2695    ///    manifest into fresh home page runs.
2696    /// 4. **Commit.** Log all of step 3's frames plus the final header
2697    ///    frame into the WAL and fsync — the durability point.
2698    /// 5. **Checkpoint.** Fold the WAL into the home pages.
2699    ///
2700    /// A crash before step 4 leaves the previous state intact. A crash
2701    /// after step 4 is repaired by [`Mnemo::open`] replaying the WAL.
2702    /// Safe to call repeatedly.
2703    pub fn flush(&mut self) -> Result<()> {
2704        if self.read_only {
2705            return Err(MnemoError::ReadOnly);
2706        }
2707        // Snapshot the thread-local fsync counter. Every `sync_all` the
2708        // pager / wal / checkpoint path makes between here and the end
2709        // of this function bumps the same counter; `delta` lands on
2710        // `self.fsyncs_last_flush` so `Stats::fsyncs_last_flush` can
2711        // report it. The planner wants this to decide whether cutting
2712        // fsync count is a viable write-latency win before we touch
2713        // the write path's durability model.
2714        let fsync_before = crate::fsync_count();
2715        if !self.dirty_catalog && !self.dirty_index && !self.dirty_cache {
2716            // No control-plane changes pending. Because every `put` that
2717            // dirties a data page also dirties the catalog, and every
2718            // `cache_put` that dirties a data page also dirties
2719            // `dirty_cache`, there can't be dirty data pages either —
2720            // nothing to do.
2721            self.fsyncs_last_flush = 0;
2722            return Ok(());
2723        }
2724
2725        // FIX-1 R-001.8 part 3: snapshot the memory/index-dirty flags
2726        // before step 3b decides whether to push a restore point.
2727        // Only flushes that change the memory catalog or ANN index
2728        // need a snapshot — a cache-only flush (`dirty_cache` set,
2729        // `dirty_catalog` + `dirty_index` both false) must not push,
2730        // because snapshots don't capture cache state and every push
2731        // counts against `max_snapshots` (default 256), so a cache-
2732        // heavy workload would otherwise evict every real memory
2733        // restore point within ~256 cache writes.
2734        let is_memory_or_index_flush = self.dirty_catalog || self.dirty_index;
2735
2736        // 1. Serialize control plane, ensure WAL capacity, lease counter +
2737        //    page slots, and persist the leased header *before* any data
2738        //    page hits the disk.
2739        let ctx = self.prepare_for_flush()?;
2740
2741        // 2. Record (vector) data pages: copy-on-write to fresh pages, fsynced.
2742        //    Safe: the lease guarantees that even if we crash here, reopen
2743        //    will see a `write_counter` past every nonce this step uses.
2744        self.pager.flush()?;
2745
2746        // 3. Seal the catalog / index control pages into fresh home runs.
2747        let mut frames: Vec<wal::Frame> = Vec::new();
2748        if let Some(bytes) = &ctx.cat_bytes {
2749            let (start, pages) = self.seal_run(bytes, &mut frames)?;
2750            self.header.catalog_start = start;
2751            self.header.catalog_pages = pages as u64;
2752            self.header.catalog_len = bytes.len() as u64;
2753        }
2754        if self.dirty_index {
2755            match &ctx.idx_bytes {
2756                Some(bytes) => {
2757                    let (start, pages) = self.seal_run(bytes, &mut frames)?;
2758                    self.header.index_start = start;
2759                    self.header.index_pages = pages as u64;
2760                    self.header.index_len = bytes.len() as u64;
2761                }
2762                None => {
2763                    self.header.index_start = 0;
2764                    self.header.index_pages = 0;
2765                    self.header.index_len = 0;
2766                }
2767            }
2768        }
2769        // v8+ result-cache directory (Phase 10.1). Seal the serialized
2770        // directory into a fresh page run and update the pointer in the
2771        // header. The seal AAD already covers cache_start/pages/len for
2772        // v8+ (see `header_seal_aad`), so tampering with the pointer
2773        // trips the header seal on the next open.
2774        if self.dirty_cache {
2775            match &ctx.cache_bytes {
2776                Some(bytes) if !self.cache_dir.is_empty() => {
2777                    let (start, pages) = self.seal_run(bytes, &mut frames)?;
2778                    self.header.cache_start = start;
2779                    self.header.cache_pages = pages as u64;
2780                    self.header.cache_len = bytes.len() as u64;
2781                }
2782                _ => {
2783                    // Either the directory serialized to nothing (all
2784                    // entries deleted after tombstone-only migration) or
2785                    // it's empty — collapse the pointer to zero.
2786                    self.header.cache_start = 0;
2787                    self.header.cache_pages = 0;
2788                    self.header.cache_len = 0;
2789                }
2790            }
2791        }
2792
2793        // 3b. Record this transaction as a restorable snapshot. The manifest
2794        //     update is staged in a local and adopted only once the commit
2795        //     below succeeds, so a failed flush leaves no phantom entry.
2796        //
2797        //     The cap (`self.max_snapshots`) is applied to the staged local
2798        //     before the new entry lands, so after appending the manifest
2799        //     holds at most `max_snapshots` entries — the most-recent N.
2800        //     `max_snapshots == 0` disables the cap. Pages referenced only
2801        //     by pruned snapshots stay on disk until `compact_file`.
2802        //
2803        //     FIX-1 R-001.8 part 3: skip the snapshot push on a
2804        //     cache-only flush. Snapshots don't capture the cache
2805        //     anyway, and every push counts against `max_snapshots`
2806        //     (default 256) + rewrites the manifest run — so a
2807        //     cache-heavy workload would otherwise evict every real
2808        //     memory restore point within ~256 cache writes. On a
2809        //     cache-only flush we reuse the current manifest
2810        //     verbatim; the manifest run already lives at
2811        //     `self.header.manifest_{start,pages,len}` (set by the
2812        //     last memory-touching flush, or at `create` time).
2813        let live = self.len() as u64;
2814        let txn_id = self.header.wal_seq + 1;
2815        let mut manifest = self.manifest.clone();
2816        if is_memory_or_index_flush {
2817            if self.max_snapshots > 0 {
2818                // Drop the oldest entries so there's room for the new one.
2819                while manifest.len() >= self.max_snapshots {
2820                    manifest.remove(0);
2821                }
2822            }
2823            manifest.push(Snapshot {
2824                txn_id,
2825                created_at: memory::now_secs(),
2826                catalog_start: self.header.catalog_start,
2827                catalog_pages: self.header.catalog_pages,
2828                catalog_len: self.header.catalog_len,
2829                index_start: self.header.index_start,
2830                index_pages: self.header.index_pages,
2831                index_len: self.header.index_len,
2832                memory_count: live,
2833            });
2834            let man_bytes =
2835                rmp_serde::to_vec(&manifest).map_err(|e| MnemoError::Serialize(e.to_string()))?;
2836            let (m_start, m_pages) = self.seal_run(&man_bytes, &mut frames)?;
2837            self.header.manifest_start = m_start;
2838            self.header.manifest_pages = m_pages as u64;
2839            self.header.manifest_len = man_bytes.len() as u64;
2840        }
2841
2842        // 3c. The header is the transaction's final frame; stamp the new id
2843        //     and the *actual* (post-flush) write counter, which is <= the
2844        //     leased value persisted in `prepare_for_flush`. This is the
2845        //     value that ultimately replaces the leased header on checkpoint.
2846        self.header.vector_count = live;
2847        self.header.write_counter = self.pager.write_counter;
2848        self.header.wal_seq = txn_id;
2849        let mut hpage = self.header.to_page();
2850        self.header.apply_seal(&mut hpage, self.pager.dek())?;
2851        frames.push((0, hpage.to_vec()));
2852
2853        // 4. COMMIT — log the transaction and fsync. Nothing at a home page
2854        //    other than the leased header has changed yet; this single
2855        //    fsync is what makes the rest durable.
2856        let (wal_start, wal_pages) = (self.header.wal_start, self.header.wal_pages);
2857        wal::commit(self.pager.file_mut(), wal_start, wal_pages, txn_id, &frames)?;
2858
2859        // 5. Checkpoint — fold the WAL into the home pages.
2860        self.checkpoint(&frames)?;
2861
2862        self.manifest = manifest;
2863        self.dirty_catalog = false;
2864        self.dirty_index = false;
2865        self.dirty_cache = false;
2866        // Batched policy: successful flush consumes the pending batch.
2867        self.cache_batch.reset();
2868        // Record how many fsyncs this flush issued, for Stats reporting.
2869        // Thread-local counter guarantees we see only this handle's own
2870        // sync_all calls between the snapshot above and this read.
2871        self.fsyncs_last_flush =
2872            (crate::fsync_count().wrapping_sub(fsync_before)).min(u32::MAX as u64) as u32;
2873        Ok(())
2874    }
2875
2876    /// Pre-flush prelude shared by [`Mnemo::flush`] and the
2877    /// `__crash_partial_flush_for_testing` hook.
2878    ///
2879    /// Serializes the control plane (catalog, ANN index, cache directory)
2880    /// and grows the WAL if needed. That's it: this used to persist a
2881    /// leased header too (SEC-1 obsoleted it; see DUR-1 R-002.1 note in
2882    /// the body for the full rationale), but after DUR-1 the only page-0
2883    /// writes outside `create` go through the WAL via
2884    /// [`Mnemo::relocate_header_through_wal`].
2885    ///
2886    /// The historical pre-stamping used to close the nonce-reuse window
2887    /// flagged in Phase 1.1 of the improvement plan: a crash anywhere
2888    /// between here and the WAL commit left a file whose on-disk header
2889    /// recorded a `write_counter` past every nonce the rest of this
2890    /// transaction could produce, so the next reopen started from a
2891    /// counter that guaranteed uniqueness against any orphan data pages
2892    /// still on disk. SEC-1's per-session random nonce prefix gives the
2893    /// same guarantee without needing a persisted counter, which is why
2894    /// this routine no longer writes page 0.
2895    fn prepare_for_flush(&mut self) -> Result<FlushPrelude> {
2896        // Serialize.
2897        let cat_bytes = if self.dirty_catalog {
2898            Some(
2899                rmp_serde::to_vec(&self.catalog)
2900                    .map_err(|e| MnemoError::Serialize(e.to_string()))?,
2901            )
2902        } else {
2903            None
2904        };
2905        let idx_bytes: Option<Vec<u8>> = if self.dirty_index {
2906            match &self.ann {
2907                Some(ann) => {
2908                    Some(rmp_serde::to_vec(ann).map_err(|e| MnemoError::Serialize(e.to_string()))?)
2909                }
2910                None => None,
2911            }
2912        } else {
2913            None
2914        };
2915        // Serialize the result-cache directory when it's dirty. An
2916        // empty directory serializes to a small header + 0-element
2917        // array; still cheap enough that we don't special-case it.
2918        let cache_bytes: Option<Vec<u8>> = if self.dirty_cache {
2919            Some(
2920                rmp_serde::to_vec(&self.cache_dir)
2921                    .map_err(|e| MnemoError::Serialize(e.to_string()))?,
2922            )
2923        } else {
2924            None
2925        };
2926
2927        let cat_pc = cat_bytes
2928            .as_ref()
2929            .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
2930        let idx_pc = idx_bytes
2931            .as_ref()
2932            .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
2933        let cache_pc = cache_bytes
2934            .as_ref()
2935            .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
2936        // Upper bound on the manifest run: one extra entry, <=82 bytes each
2937        // (rmp_serde encodes a Snapshot as a 9-element fixarray of i64/u64,
2938        // worst case 81 bytes plus a 1-byte array header).
2939        let man_upper = 9 + (self.manifest.len() + 1) * 82;
2940        let man_pc_est = man_upper.div_ceil(PAYLOAD).max(1);
2941
2942        // Grow WAL if needed (DUR-1: relocation is WAL-logged and does
2943        // its own fsyncs). The cache directory rides in the same
2944        // transaction so its page count contributes to the reservation.
2945        self.ensure_wal_capacity(cat_pc + idx_pc + cache_pc + man_pc_est + 1)?;
2946
2947        // DUR-1 (R-002.1): the pre-flush "leased header" write that used
2948        // to live here is gone. It existed to close the Phase 1.1 nonce
2949        // reuse window (reopen must see a `write_counter` past any nonce
2950        // this flush could produce before data pages hit disk). SEC-1
2951        // superseded that: nonces now carry a per-session random prefix
2952        // and `write_counter` is no longer a nonce input, so the leased
2953        // write bought us nothing but an extra fsync per flush. Orphan
2954        // data pages past `header.next_page` on a crashed flush are
2955        // still safe — the next session's random prefix makes any later
2956        // write at the same page_no uniquely nonced regardless of
2957        // `write_counter`.
2958
2959        Ok(FlushPrelude {
2960            cat_bytes,
2961            idx_bytes,
2962            cache_bytes,
2963        })
2964    }
2965
2966    /// Flush and close. Equivalent to `flush()` on a read-write handle; on
2967    /// a read-only handle it's a no-op (no dirty state to persist). The
2968    /// file is released on drop, which also drops the OS advisory lock.
2969    pub fn close(&mut self) -> Result<()> {
2970        if self.read_only {
2971            return Ok(());
2972        }
2973        self.flush()
2974    }
2975
2976    /// **TEST ONLY — do not use.** Reproduces a crash inside
2977    /// [`Mnemo::flush`] *after* the prelude has leased counter+page slots
2978    /// and persisted the leased header, but *before* the WAL is committed.
2979    /// Runs `prepare_for_flush` + `pager.flush` and returns — leaving
2980    /// data pages and a leased header on disk, with no commit frame.
2981    ///
2982    /// Used by `tests/integration.rs::nonce_unique_after_crashed_data_flush`
2983    /// to verify that this window (Phase 1.1 of the improvement plan) does
2984    /// not enable AES-GCM nonce reuse. Calling this in production code
2985    /// strands data pages; never expose it from a binding.
2986    #[doc(hidden)]
2987    pub fn __crash_partial_flush_for_testing(&mut self) -> Result<()> {
2988        if self.read_only {
2989            return Err(MnemoError::ReadOnly);
2990        }
2991        if !self.dirty_catalog && !self.dirty_index {
2992            return Ok(());
2993        }
2994        let _prelude = self.prepare_for_flush()?;
2995        self.pager.flush()
2996    }
2997
2998    /// Bound the in-memory page cache to `pages` decrypted pages.
2999    ///
3000    /// The cache holds decrypted page payloads to speed repeated reads. By
3001    /// default it is capped at 8192 pages (~64 MiB); lower the cap to trade
3002    /// hit rate for a smaller footprint, or raise it for a hotter cache. The
3003    /// cap governs *clean* pages — pages with un-flushed writes are always
3004    /// retained until [`Mnemo::flush`], regardless of the cap.
3005    pub fn set_cache_capacity(&mut self, pages: usize) {
3006        self.pager.set_cache_capacity(pages);
3007    }
3008
3009    /// Page-cache occupancy: `(pages_cached, capacity)`. Renamed from
3010    /// `cache_stats` in v0.4.0: the result-cache API (Phase 10.1) took
3011    /// that name for its per-namespace [`CacheStats`] summary, and the
3012    /// page-cache pair is the older/less-user-facing of the two, so it
3013    /// gets the more-specific name.
3014    pub fn page_cache_stats(&self) -> (usize, usize) {
3015        self.pager.cache_stats()
3016    }
3017
3018    // === Result cache (Phase 10.1 + 10.3) ============================
3019
3020    /// Insert or overwrite a cache entry at `(namespace, key)`.
3021    ///
3022    /// The engine SHA-256-hashes `key` — callers pass the raw input
3023    /// string. Persists to disk on the next `flush()` under the current
3024    /// flush policy (see [`Mnemo::set_cache_flush_policy`]); in
3025    /// [`CacheFlushPolicy::Batched`] the engine may call `flush()`
3026    /// internally when a threshold trips.
3027    ///
3028    /// On overflow of the namespace's [`CacheBudget`] (default 10,000
3029    /// entries / 64 MiB) the LRU entry is tombstoned before the new
3030    /// insert. Tombstones are reclaimed by
3031    /// [`Mnemo::compact_file`], not the runtime flush.
3032    ///
3033    /// Returns [`MnemoError::ReadOnly`] on a read-only handle.
3034    pub fn cache_put(
3035        &mut self,
3036        namespace: &str,
3037        key: &str,
3038        value: &[u8],
3039        opts: CachePutOpts,
3040    ) -> Result<()> {
3041        if self.read_only {
3042            return Err(MnemoError::ReadOnly);
3043        }
3044        let key_hash = result_cache::hash_key(key);
3045        let key_preview = result_cache::key_preview(key);
3046        let now = memory::now_secs();
3047        let size_bytes = value.len() as u64;
3048
3049        let entry = CacheEntry {
3050            namespace: namespace.to_string(),
3051            key_hash,
3052            key_preview: key_preview.clone(),
3053            value: value.to_vec(),
3054            content_type: opts.content_type.clone(),
3055            created_at: now,
3056            accessed_at: now,
3057            access_count: 0,
3058            ttl_secs: opts.ttl_secs,
3059            cost_hint_ms: opts.cost_hint_ms,
3060            size_bytes,
3061            vector: None,
3062            model: None,
3063        };
3064        let bytes = rmp_serde::to_vec(&entry).map_err(|e| MnemoError::Serialize(e.to_string()))?;
3065        let (start_page, page_count) = self.write_record(&bytes)?;
3066
3067        // Tombstone any prior entry for this key. Its pages remain on
3068        // disk until `compact_file`, exactly like memory-record
3069        // overwrites — the tombstone just hides it from lookups.
3070        if let Some(prev_pos) = self.cache_idx.get(namespace, &key_hash) {
3071            self.cache_dir[prev_pos].deleted = true;
3072            self.cache_idx.remove(namespace, &key_hash);
3073        }
3074
3075        let dir_entry = CacheDirectoryEntry {
3076            namespace: namespace.to_string(),
3077            key_hash,
3078            start_page,
3079            page_count,
3080            len: bytes.len() as u32,
3081            deleted: false,
3082            accessed_at: now,
3083            access_count: 0,
3084            size_bytes,
3085            ttl_secs: opts.ttl_secs,
3086            created_at: now,
3087            vector: None,
3088            model: None,
3089        };
3090        let pos = self.cache_dir.len();
3091        self.cache_dir.push(dir_entry);
3092        self.cache_idx.insert(namespace, key_hash, pos);
3093
3094        // Enforce budget — evict LRU tombstones until the namespace
3095        // fits under both caps.
3096        self.evict_to_budget(namespace)?;
3097
3098        self.dirty_cache = true;
3099        self.cache_batch.record_dirty();
3100
3101        // Batched policy: auto-flush if a threshold tripped. Successful
3102        // `flush()` calls `cache_batch.reset()` for us.
3103        if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
3104            self.flush()?;
3105        }
3106        Ok(())
3107    }
3108
3109    /// Look up `(namespace, key)`. Returns `None` on miss (either the
3110    /// key was never inserted or the entry is TTL-expired). On hit
3111    /// updates catalog-side access stats (mirrors the recall v5-trick);
3112    /// on a read-only handle the stats update is silently skipped.
3113    ///
3114    /// A miss is recorded as a miss in this handle's counters even if
3115    /// the key was TTL-expired — from the caller's perspective the two
3116    /// are indistinguishable.
3117    pub fn cache_get(&mut self, namespace: &str, key: &str) -> Result<Option<CachedValue>> {
3118        // FIX-1 R-001.8 part 4: under `Batched`, honor `max_age` on
3119        // reads too. Pre-FIX-1 the age check only fired on `cache_put`,
3120        // so a workload that wrote a few entries and then switched to
3121        // pure reads could sit with its pending batch past the
3122        // nominal max_age indefinitely. `max_dirty` is already
3123        // maintained per-write; this adds the symmetric time-bound
3124        // check on the read path. Only runs on read-write handles —
3125        // flush() returns ReadOnly for shared-lock readers.
3126        if !self.read_only && self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
3127            self.flush()?;
3128        }
3129        let key_hash = result_cache::hash_key(key);
3130        let pos = match self.cache_idx.get(namespace, &key_hash) {
3131            Some(p) => p,
3132            None => {
3133                self.cache_counters.misses += 1;
3134                return Ok(None);
3135            }
3136        };
3137        let now = memory::now_secs();
3138        // Snapshot the directory entry fields we need before any
3139        // mutation — the borrow checker gets grumpy otherwise.
3140        let (start_page, page_count, len, ttl_secs, expired, created_at) = {
3141            let e = &self.cache_dir[pos];
3142            (
3143                e.start_page,
3144                e.page_count,
3145                e.len,
3146                e.ttl_secs,
3147                e.is_expired(now),
3148                e.created_at,
3149            )
3150        };
3151        if expired {
3152            // Treat expired as a miss for the caller. Don't tombstone
3153            // here — cache_purge does that. Just skip.
3154            self.cache_counters.misses += 1;
3155            return Ok(None);
3156        }
3157        // Decrypt and decode the record body.
3158        let mut buf = Vec::with_capacity(len as usize);
3159        for i in 0..page_count as u64 {
3160            buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
3161        }
3162        buf.truncate(len as usize);
3163        let entry: CacheEntry =
3164            rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
3165
3166        // Bump access stats catalog-only (v5 trick — no full record
3167        // rewrite on a hit). Skipped on read-only handles so a shared-lock
3168        // reader can't dirty the catalog.
3169        if !self.read_only {
3170            self.cache_dir[pos].accessed_at = now;
3171            self.cache_dir[pos].access_count = self.cache_dir[pos].access_count.saturating_add(1);
3172            self.dirty_cache = true;
3173        }
3174        self.cache_counters.hits += 1;
3175        // On a mutating handle: access_count reflects the just-bumped
3176        // value. On a read-only handle: the on-disk directory value is
3177        // unchanged, so returning it verbatim is honest (the caller's
3178        // reads did not, and cannot, alter the persisted counter).
3179        //
3180        // FIX-1 R-002.2 part 1: `accessed_at` must also come from the
3181        // directory on a read-only handle, not from `entry.accessed_at`
3182        // in the record body. The record body is written once at insert
3183        // time and never updated on a hit — every live `accessed_at`
3184        // update lands in the directory (see the `!self.read_only`
3185        // branch above). The pre-FIX-1 code returned
3186        // `entry.accessed_at`, which is really the entry's **creation
3187        // time** by then, so a read-only reader saw "last accessed =
3188        // when it was inserted" even after many successful hits by
3189        // concurrent writers.
3190        let access_count = self.cache_dir[pos].access_count;
3191        let accessed_at = if self.read_only {
3192            self.cache_dir[pos].accessed_at
3193        } else {
3194            now
3195        };
3196        Ok(Some(CachedValue {
3197            value: entry.value,
3198            content_type: entry.content_type,
3199            created_at,
3200            accessed_at,
3201            access_count,
3202            ttl_secs,
3203        }))
3204    }
3205
3206    /// Tombstone the entry at `(namespace, key)`. Silently succeeds if
3207    /// no such entry exists. Space is reclaimed by `compact_file`.
3208    pub fn cache_delete(&mut self, namespace: &str, key: &str) -> Result<()> {
3209        if self.read_only {
3210            return Err(MnemoError::ReadOnly);
3211        }
3212        let key_hash = result_cache::hash_key(key);
3213        if let Some(pos) = self.cache_idx.get(namespace, &key_hash) {
3214            self.cache_dir[pos].deleted = true;
3215            self.cache_idx.remove(namespace, &key_hash);
3216            self.dirty_cache = true;
3217            self.cache_batch.record_dirty();
3218            if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
3219                self.flush()?;
3220            }
3221        }
3222        Ok(())
3223    }
3224
3225    /// Purge cache entries. If `namespace` is `Some`, restrict to that
3226    /// namespace; otherwise purge across the whole cache. If
3227    /// `expired_only` is `true`, only TTL-expired entries are removed;
3228    /// otherwise every live entry in the target scope is tombstoned.
3229    /// Returns the number of entries newly tombstoned.
3230    pub fn cache_purge(&mut self, namespace: Option<&str>, expired_only: bool) -> Result<usize> {
3231        if self.read_only {
3232            return Err(MnemoError::ReadOnly);
3233        }
3234        let now = memory::now_secs();
3235        let mut purged = 0usize;
3236        // Iterate the directory by index; mark matching entries as
3237        // deleted and pull them out of the in-memory index.
3238        for pos in 0..self.cache_dir.len() {
3239            if self.cache_dir[pos].deleted {
3240                continue;
3241            }
3242            if let Some(ns) = namespace {
3243                if self.cache_dir[pos].namespace != ns {
3244                    continue;
3245                }
3246            }
3247            if expired_only && !self.cache_dir[pos].is_expired(now) {
3248                continue;
3249            }
3250            let key_hash = self.cache_dir[pos].key_hash;
3251            let ns = self.cache_dir[pos].namespace.clone();
3252            self.cache_dir[pos].deleted = true;
3253            self.cache_idx.remove(&ns, &key_hash);
3254            purged += 1;
3255        }
3256        if purged > 0 {
3257            self.dirty_cache = true;
3258            self.cache_batch.record_dirty();
3259            if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
3260                self.flush()?;
3261            }
3262        }
3263        Ok(purged)
3264    }
3265
3266    /// Statistics for `namespace` (or the whole cache if `None`).
3267    /// `hits`/`misses`/`evictions` are per-handle in-memory counters
3268    /// that reset on reopen; `entries`/`bytes` reflect the durable
3269    /// on-disk state (live entries only, tombstones excluded).
3270    pub fn cache_stats(&self, namespace: Option<&str>) -> CacheStats {
3271        let mut entries = 0usize;
3272        let mut bytes = 0u64;
3273        for e in &self.cache_dir {
3274            if e.deleted {
3275                continue;
3276            }
3277            if let Some(ns) = namespace {
3278                if e.namespace != ns {
3279                    continue;
3280                }
3281            }
3282            entries += 1;
3283            bytes += e.size_bytes;
3284        }
3285        let hits = self.cache_counters.hits;
3286        let misses = self.cache_counters.misses;
3287        let total = hits + misses;
3288        let hit_rate = if total > 0 {
3289            hits as f64 / total as f64
3290        } else {
3291            0.0
3292        };
3293        CacheStats {
3294            entries,
3295            bytes,
3296            hits,
3297            misses,
3298            hit_rate,
3299            evictions: self.cache_counters.evictions,
3300        }
3301    }
3302
3303    /// Set (or update) the eviction budget for `namespace`. Applies to
3304    /// subsequent `cache_put` calls; does not retroactively evict
3305    /// entries that already sit under a looser cap.
3306    pub fn set_cache_budget(&mut self, namespace: &str, budget: CacheBudget) {
3307        self.cache_budgets.insert(namespace.to_string(), budget);
3308    }
3309
3310    /// Set the per-handle cache flush policy. Defaults to
3311    /// [`CacheFlushPolicy::Strict`]. Memory writes are never batched;
3312    /// this affects only cache mutations.
3313    pub fn set_cache_flush_policy(&mut self, policy: CacheFlushPolicy) {
3314        self.cache_flush_policy = policy;
3315        // Switching mid-run: if we're moving to Strict and the batch is
3316        // non-empty, the pending count still counts toward dirty_cache
3317        // which the next flush() will consume.
3318    }
3319
3320    /// Evict LRU live entries in `namespace` until it fits under both
3321    /// budget caps. Called from `cache_put` after the insert; the new
3322    /// entry is included in the count so we always leave room.
3323    fn evict_to_budget(&mut self, namespace: &str) -> Result<()> {
3324        let budget = self
3325            .cache_budgets
3326            .get(namespace)
3327            .copied()
3328            .unwrap_or_default();
3329        loop {
3330            // Sum current live count + bytes in the namespace.
3331            let mut live_count = 0usize;
3332            let mut live_bytes = 0u64;
3333            for e in &self.cache_dir {
3334                if !e.deleted && e.namespace == namespace {
3335                    live_count += 1;
3336                    live_bytes += e.size_bytes;
3337                }
3338            }
3339            if live_count <= budget.max_entries && live_bytes <= budget.max_bytes {
3340                return Ok(());
3341            }
3342            // Find the LRU (oldest accessed_at) live entry.
3343            let victim_pos = self
3344                .cache_dir
3345                .iter()
3346                .enumerate()
3347                .filter(|(_, e)| !e.deleted && e.namespace == namespace)
3348                .min_by_key(|(_, e)| e.accessed_at)
3349                .map(|(i, _)| i);
3350            let Some(pos) = victim_pos else {
3351                // Nothing to evict but still over budget — the single
3352                // just-inserted entry itself is bigger than
3353                // max_bytes. Nothing we can safely do here without
3354                // rejecting the put; leave it and return.
3355                return Ok(());
3356            };
3357            let key_hash = self.cache_dir[pos].key_hash;
3358            let ns = self.cache_dir[pos].namespace.clone();
3359            self.cache_dir[pos].deleted = true;
3360            self.cache_idx.remove(&ns, &key_hash);
3361            self.cache_counters.evictions += 1;
3362        }
3363    }
3364
3365    // === Semantic cache (Phase 10.2) ==================================
3366
3367    /// Insert or overwrite a **semantic** cache entry — same as
3368    /// [`Mnemo::cache_put`] but with a required embedding vector and
3369    /// model identifier. The vector is stored on the directory entry
3370    /// so [`Mnemo::cache_get_semantic`] can score without decrypting
3371    /// record bodies.
3372    ///
3373    /// Vector dimensionality is validated against the database's
3374    /// configured `dimensions` — a mismatch returns
3375    /// [`MnemoError::DimensionMismatch`], the same signal `remember`
3376    /// gives for the memory catalog.
3377    ///
3378    /// The `model` string is matched **exactly** on lookup — a
3379    /// different model's cache is invisible to a given query. Pick
3380    /// canonical identifiers (`"text-embedding-3-small"`,
3381    /// `"bge-large-en-v1.5"`, ...) and stick to them.
3382    pub fn cache_put_semantic(
3383        &mut self,
3384        namespace: &str,
3385        key: &str,
3386        vector: Vec<f32>,
3387        value: &[u8],
3388        opts: SemanticCachePutOpts,
3389    ) -> Result<()> {
3390        if self.read_only {
3391            return Err(MnemoError::ReadOnly);
3392        }
3393        if vector.len() != self.dimensions {
3394            return Err(MnemoError::DimensionMismatch {
3395                expected: self.dimensions,
3396                got: vector.len(),
3397            });
3398        }
3399        let key_hash = result_cache::hash_key(key);
3400        let key_preview = result_cache::key_preview(key);
3401        let now = memory::now_secs();
3402        let size_bytes = value.len() as u64;
3403
3404        let entry = CacheEntry {
3405            namespace: namespace.to_string(),
3406            key_hash,
3407            key_preview: key_preview.clone(),
3408            value: value.to_vec(),
3409            content_type: opts.content_type.clone(),
3410            created_at: now,
3411            accessed_at: now,
3412            access_count: 0,
3413            ttl_secs: opts.ttl_secs,
3414            cost_hint_ms: opts.cost_hint_ms,
3415            size_bytes,
3416            vector: Some(vector.clone()),
3417            model: Some(opts.model.clone()),
3418        };
3419        let bytes = rmp_serde::to_vec(&entry).map_err(|e| MnemoError::Serialize(e.to_string()))?;
3420        let (start_page, page_count) = self.write_record(&bytes)?;
3421
3422        if let Some(prev_pos) = self.cache_idx.get(namespace, &key_hash) {
3423            self.cache_dir[prev_pos].deleted = true;
3424            self.cache_idx.remove(namespace, &key_hash);
3425        }
3426
3427        let dir_entry = CacheDirectoryEntry {
3428            namespace: namespace.to_string(),
3429            key_hash,
3430            start_page,
3431            page_count,
3432            len: bytes.len() as u32,
3433            deleted: false,
3434            accessed_at: now,
3435            access_count: 0,
3436            size_bytes,
3437            ttl_secs: opts.ttl_secs,
3438            created_at: now,
3439            vector: Some(vector),
3440            model: Some(opts.model),
3441        };
3442        let pos = self.cache_dir.len();
3443        self.cache_dir.push(dir_entry);
3444        self.cache_idx.insert(namespace, key_hash, pos);
3445
3446        self.evict_to_budget(namespace)?;
3447
3448        self.dirty_cache = true;
3449        self.cache_batch.record_dirty();
3450        if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
3451            self.flush()?;
3452        }
3453        Ok(())
3454    }
3455
3456    /// Semantic recall over the cache: score every live entry in
3457    /// `namespace` whose `model` matches against `query_vector` using
3458    /// cosine similarity; return the top-1 hit and its similarity iff
3459    /// `sim >= threshold`. `None` otherwise.
3460    ///
3461    /// Uses [`crate::result_cache::DEFAULT_SEMANTIC_THRESHOLD`] if you
3462    /// want a sensible default — pass it explicitly (`0.97` today).
3463    ///
3464    /// Exact-key entries (those without a vector) are transparently
3465    /// skipped, as are entries whose model doesn't match — a
3466    /// different-model hit would be a semantic bug, not a win.
3467    /// TTL-expired entries are also skipped.
3468    ///
3469    /// Access-stat updates on hit follow the same policy as
3470    /// [`Mnemo::cache_get`]: bumped on read-write handles, silently
3471    /// skipped on read-only. Hits and misses are recorded in the
3472    /// per-handle counters just like exact-key lookups.
3473    pub fn cache_get_semantic(
3474        &mut self,
3475        namespace: &str,
3476        query_vector: &[f32],
3477        threshold: f32,
3478        model: &str,
3479    ) -> Result<Option<(CachedValue, f32)>> {
3480        if query_vector.len() != self.dimensions {
3481            return Err(MnemoError::DimensionMismatch {
3482                expected: self.dimensions,
3483                got: query_vector.len(),
3484            });
3485        }
3486        // FIX-1 R-001.8 part 4: same Batched max_age check as
3487        // `cache_get`. See the comment there.
3488        if !self.read_only && self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
3489            self.flush()?;
3490        }
3491        let now = memory::now_secs();
3492
3493        // Scan namespace live entries, filter by model, compute
3494        // cosine, keep the best. Exact-key entries have `vector: None`
3495        // and are skipped by the filter.
3496        let mut best: Option<(usize, f32)> = None;
3497        for (i, e) in self.cache_dir.iter().enumerate() {
3498            if e.deleted || e.namespace != namespace || e.is_expired(now) {
3499                continue;
3500            }
3501            let Some(ref v) = e.vector else { continue };
3502            let Some(ref m) = e.model else { continue };
3503            if m != model {
3504                continue;
3505            }
3506            let sim = memory::cosine(query_vector, v);
3507            if best.map(|(_, s)| sim > s).unwrap_or(true) {
3508                best = Some((i, sim));
3509            }
3510        }
3511
3512        let Some((pos, sim)) = best else {
3513            self.cache_counters.misses += 1;
3514            return Ok(None);
3515        };
3516        if sim < threshold {
3517            self.cache_counters.misses += 1;
3518            return Ok(None);
3519        }
3520
3521        // Extract fields before any mutation to keep the borrow checker
3522        // happy.
3523        let (start_page, page_count, len, ttl_secs, created_at) = {
3524            let e = &self.cache_dir[pos];
3525            (e.start_page, e.page_count, e.len, e.ttl_secs, e.created_at)
3526        };
3527
3528        // Decrypt and decode the record body.
3529        let mut buf = Vec::with_capacity(len as usize);
3530        for i in 0..page_count as u64 {
3531            buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
3532        }
3533        buf.truncate(len as usize);
3534        let entry: CacheEntry =
3535            rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
3536
3537        if !self.read_only {
3538            self.cache_dir[pos].accessed_at = now;
3539            self.cache_dir[pos].access_count = self.cache_dir[pos].access_count.saturating_add(1);
3540            self.dirty_cache = true;
3541        }
3542        self.cache_counters.hits += 1;
3543        let access_count = self.cache_dir[pos].access_count;
3544        // FIX-1 R-002.2 part 1: same read-only `accessed_at` fix as
3545        // `cache_get`. The directory carries the live value; the record
3546        // body only holds the insert-time snapshot.
3547        let accessed_at = if self.read_only {
3548            self.cache_dir[pos].accessed_at
3549        } else {
3550            now
3551        };
3552        Ok(Some((
3553            CachedValue {
3554                value: entry.value,
3555                content_type: entry.content_type,
3556                created_at,
3557                accessed_at,
3558                access_count,
3559                ttl_secs,
3560            },
3561            sim,
3562        )))
3563    }
3564
3565    /// Begin a conversation [`Session`](crate::Session) for `agent_id`.
3566    ///
3567    /// The session borrows the database for its lifetime, records turns as
3568    /// working memory, and consolidates them into episodic memory when closed.
3569    pub fn session(&mut self, agent_id: impl Into<String>) -> crate::session::Session<'_> {
3570        crate::session::Session::new(self, agent_id.into())
3571    }
3572
3573    // --- snapshots & point-in-time recovery ------------------------------
3574
3575    /// Every committed transaction, oldest first — the restore points
3576    /// available to [`Mnemo::restore_to`] and [`Mnemo::restore_to_time`].
3577    ///
3578    /// Each `flush` appends one snapshot. Because the storage engine is
3579    /// append-only, the pages a past flush wrote are still on disk, so any
3580    /// listed snapshot can be reinstated exactly. The history reaches back to
3581    /// the last [`Mnemo::compact_file`], which reclaims space by collapsing
3582    /// it.
3583    pub fn snapshots(&self) -> Vec<SnapshotInfo> {
3584        let mut v: Vec<SnapshotInfo> = self
3585            .manifest
3586            .iter()
3587            .map(|s| SnapshotInfo {
3588                txn_id: s.txn_id,
3589                created_at: s.created_at,
3590                memory_count: s.memory_count,
3591            })
3592            .collect();
3593        v.sort_by_key(|s| s.txn_id);
3594        v
3595    }
3596
3597    /// Load a past snapshot's state and commit it as a new transaction.
3598    fn apply_snapshot(&mut self, snap: &Snapshot) -> Result<()> {
3599        let catalog = load_catalog(
3600            &mut self.pager,
3601            snap.catalog_start,
3602            snap.catalog_pages,
3603            snap.catalog_len,
3604        )?;
3605        let ann = load_index(
3606            &mut self.pager,
3607            snap.index_start,
3608            snap.index_pages,
3609            snap.index_len,
3610            self.dimensions,
3611        )?;
3612        self.index = build_id_index(&catalog);
3613        self.catalog = catalog;
3614        self.ann = ann;
3615        // Re-commit as a fresh transaction: crash-safe, and itself recorded
3616        // as a new snapshot so a restore can always be undone.
3617        self.dirty_catalog = true;
3618        self.dirty_index = true;
3619        self.flush()
3620    }
3621
3622    /// Restore the database to the snapshot produced by transaction `txn_id`.
3623    ///
3624    /// The restore is itself a new committed transaction (and a new
3625    /// snapshot), so it is crash-safe and reversible — restoring forward to a
3626    /// later snapshot afterwards works just as well.
3627    pub fn restore_to(&mut self, txn_id: u64) -> Result<SnapshotInfo> {
3628        if self.read_only {
3629            return Err(MnemoError::ReadOnly);
3630        }
3631        let snap = self
3632            .manifest
3633            .iter()
3634            .find(|s| s.txn_id == txn_id)
3635            .cloned()
3636            .ok_or_else(|| MnemoError::NotFound(format!("snapshot for transaction {txn_id}")))?;
3637        self.apply_snapshot(&snap)?;
3638        Ok(SnapshotInfo {
3639            txn_id: snap.txn_id,
3640            created_at: snap.created_at,
3641            memory_count: snap.memory_count,
3642        })
3643    }
3644
3645    /// Restore the database to the latest snapshot committed at or before
3646    /// `unix_secs`. Returns [`MnemoError::NotFound`] if no snapshot is that
3647    /// old. Like [`Mnemo::restore_to`], the restore is a new transaction.
3648    pub fn restore_to_time(&mut self, unix_secs: i64) -> Result<SnapshotInfo> {
3649        if self.read_only {
3650            return Err(MnemoError::ReadOnly);
3651        }
3652        let snap = self
3653            .manifest
3654            .iter()
3655            .filter(|s| s.created_at <= unix_secs)
3656            .max_by_key(|s| s.txn_id)
3657            .cloned()
3658            .ok_or_else(|| {
3659                MnemoError::NotFound(format!("no snapshot at or before time {unix_secs}"))
3660            })?;
3661        self.apply_snapshot(&snap)?;
3662        Ok(SnapshotInfo {
3663            txn_id: snap.txn_id,
3664            created_at: snap.created_at,
3665            memory_count: snap.memory_count,
3666        })
3667    }
3668
3669    /// Return the KDF (Argon2id) parameters this file currently uses to
3670    /// wrap its DEK. Needed by callers that want to [`rekey`] while
3671    /// keeping the file's original protection level (e.g. the Python
3672    /// binding's `rekey(fast=False)`, whose docstring promises exactly
3673    /// that). FIX-1 R-001.5.
3674    pub fn kdf_params(&self) -> KdfParams {
3675        self.kdf
3676    }
3677
3678    /// Flush any pending writes, then stream the (encrypted) file to
3679    /// `dest` atomically: writes to a sibling temp file in `dest`'s
3680    /// parent directory, `sync_all`s it, renames onto `dest`, and
3681    /// fsyncs the parent directory on Unix so a crash never leaves a
3682    /// partial export.
3683    ///
3684    /// Refuses when `dest` resolves to the file this handle has open
3685    /// — that case would otherwise `File::create`-truncate the live
3686    /// database to 0 bytes (`fs::copy` on Unix reads *after* the
3687    /// `O_TRUNC`, so a naive export-to-self destroys the source).
3688    /// Unix: compares `(dev, ino)` of both paths via `MetadataExt`,
3689    /// catching symlinks and hard links. Windows: compares
3690    /// canonicalized paths; a rename onto an open file also fails on
3691    /// Windows because of the mandatory lock the handle holds, so the
3692    /// rename step is a second line of defense.
3693    ///
3694    /// Reads the bytes from the already-open file handle, not by
3695    /// `fs::copy`. On Windows that matters: `LockFileEx` locks are
3696    /// mandatory, so a second handle opened by `fs::copy` would fail
3697    /// while this writer still holds the lock.
3698    ///
3699    /// FIX-1 R-001.5 + R-002.2 part 3.
3700    pub fn export_encrypted(&mut self, dest: impl AsRef<Path>) -> Result<()> {
3701        let dest = dest.as_ref();
3702
3703        // Step 1 — refuse export to the open file itself. We compare
3704        // by identity, not string, so symlinks and hard links to
3705        // `self.path` are caught too. The check runs before any flush
3706        // so a self-targeted export is a pure no-op.
3707        if paths_refer_to_same_file(&self.path, dest).unwrap_or(false) {
3708            return Err(MnemoError::Invalid(format!(
3709                "export_encrypted refuses to overwrite the open database at {}",
3710                dest.display()
3711            )));
3712        }
3713
3714        // Step 2 — flush, so on-disk bytes match the live state.
3715        if !self.read_only {
3716            self.flush()?;
3717        }
3718
3719        // Step 3 — stream from the open handle into a sibling temp.
3720        // Using the open handle (instead of a second open on `self.path`)
3721        // means a Windows mandatory lock can't block the copy.
3722        let parent = dest.parent().ok_or_else(|| {
3723            MnemoError::Invalid(format!(
3724                "export_encrypted: dest {} has no parent directory",
3725                dest.display()
3726            ))
3727        })?;
3728        let tmp = temp_sibling(dest);
3729        // Scope the temp handle so it closes before the rename on
3730        // Windows (which refuses to rename over an open file).
3731        {
3732            let src = self.pager.file_mut();
3733            src.seek(SeekFrom::Start(0))?;
3734            let mut out = OpenOptions::new()
3735                .write(true)
3736                .create(true)
3737                .truncate(true)
3738                .open(&tmp)?;
3739            let mut buf = vec![0u8; 1 << 20]; // 1 MiB
3740            loop {
3741                let n = src.read(&mut buf)?;
3742                if n == 0 {
3743                    break;
3744                }
3745                out.write_all(&buf[..n])?;
3746            }
3747            out.sync_all()?;
3748            crate::bump_fsync();
3749        }
3750
3751        // Step 4 — atomic rename onto `dest`.
3752        if let Err(e) = std::fs::rename(&tmp, dest) {
3753            // Best-effort cleanup; propagate the rename error.
3754            let _ = std::fs::remove_file(&tmp);
3755            return Err(MnemoError::Io(e));
3756        }
3757
3758        // Step 5 — fsync the parent directory on Unix so the rename
3759        // itself is durable. On Windows directory syncs aren't
3760        // supported; the rename above is atomic enough on NTFS.
3761        #[cfg(unix)]
3762        {
3763            if let Ok(dir) = File::open(parent) {
3764                let _ = dir.sync_all();
3765                crate::bump_fsync();
3766            }
3767        }
3768        #[cfg(not(unix))]
3769        {
3770            let _ = parent;
3771        }
3772
3773        Ok(())
3774    }
3775
3776    /// Change the passphrase. Cheap: re-derives the KEK and re-wraps the DEK;
3777    /// the encrypted pages are never rewritten.
3778    pub fn rekey(&mut self, new_passphrase: &str, kdf: KdfParams) -> Result<()> {
3779        if self.read_only {
3780            return Err(MnemoError::ReadOnly);
3781        }
3782        self.flush()?;
3783        let new_salt = crypto::random_salt();
3784        let new_kek = crypto::derive_kek(new_passphrase.as_bytes(), &new_salt, kdf)?;
3785        let new_nonce = crypto::random_nonce();
3786        let wrapped = crypto::wrap_dek(&new_kek, &new_nonce, self.pager.dek())?;
3787        if wrapped.len() != WRAPPED_DEK_LEN {
3788            return Err(MnemoError::Crypto("unexpected wrapped DEK length".into()));
3789        }
3790        let mut wrapped_dek = [0u8; WRAPPED_DEK_LEN];
3791        wrapped_dek.copy_from_slice(&wrapped);
3792
3793        self.header.salt = new_salt;
3794        self.header.dek_nonce = new_nonce;
3795        self.header.wrapped_dek = wrapped_dek;
3796        self.header.m_cost = kdf.m_cost;
3797        self.header.t_cost = kdf.t_cost;
3798        self.header.p_cost = kdf.p_cost;
3799        self.kdf = kdf;
3800
3801        // DUR-1 (R-002.1): WAL-log the new page 0 before touching it.
3802        // Allocate a fresh WAL region at the tail so that a crash during
3803        // the rekey leaves either the old region (old passphrase works)
3804        // or the new region (new passphrase works) fully committed —
3805        // overwriting the existing region would risk destroying both.
3806        let old_wal_start = self.header.wal_start;
3807        let new_pages = self.header.wal_pages;
3808        let new_start = self.header.next_page;
3809        self.header.next_page += new_pages;
3810        self.header.wal_start = new_start;
3811        self.header.wal_pages = new_pages;
3812        self.header.wal_seq += 1;
3813        self.relocate_header_through_wal(old_wal_start)
3814    }
3815
3816    /// Decrypt and validate every live record. Returns the count verified
3817    /// across memories **and** cache entries (FIX-1 R-001.1: now that the
3818    /// cache survives compaction, `verify()` has to walk it too — the
3819    /// pre-FIX-1 implementation only checked `self.catalog`, so a
3820    /// corrupted cache record went unnoticed).
3821    pub fn verify(&mut self) -> Result<usize> {
3822        let entries: Vec<CatalogEntry> = self
3823            .catalog
3824            .iter()
3825            .filter(|e| !e.deleted)
3826            .cloned()
3827            .collect();
3828        let mut n = entries.len();
3829        for e in &entries {
3830            let bytes = self.read_record(e)?;
3831            let _m: Memory = rmp_serde::from_slice(&bytes)
3832                .map_err(|err| MnemoError::Serialize(err.to_string()))?;
3833        }
3834
3835        // Also decode every live cache entry. Tombstoned entries are
3836        // skipped — their pages are orphan ciphertext awaiting
3837        // `compact_file`, same as deleted memories. Expired entries
3838        // are *still* on disk until compaction, so we verify them too.
3839        let cache_positions: Vec<(u64, u32, u32)> = self
3840            .cache_dir
3841            .iter()
3842            .filter(|e| !e.deleted)
3843            .map(|e| (e.start_page, e.page_count, e.len))
3844            .collect();
3845        for (start_page, page_count, len) in cache_positions {
3846            let mut buf = Vec::with_capacity(len as usize);
3847            for i in 0..page_count as u64 {
3848                buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
3849            }
3850            buf.truncate(len as usize);
3851            let _c: CacheEntry = rmp_serde::from_slice(&buf)
3852                .map_err(|err| MnemoError::Serialize(err.to_string()))?;
3853            n += 1;
3854        }
3855        Ok(n)
3856    }
3857
3858    /// Summary statistics for the open database.
3859    pub fn stats(&mut self) -> Result<Stats> {
3860        let deleted = self.catalog.iter().filter(|e| e.deleted).count();
3861        // FIX-1 R-001.8 part 2: build the agent set lazily on first
3862        // call per handle, then maintain in `put` / invalidate in
3863        // `delete`. Pre-FIX-1 every `stats()` call decrypted every
3864        // live memory (O(N) AES-GCM decrypts just to list a usually-
3865        // tiny set of distinct agent IDs). On a 1M-memory database
3866        // this was in the multi-second range on hardware AES.
3867        if self.agent_set.is_none() {
3868            let built: std::collections::BTreeSet<String> =
3869                self.memories()?.into_iter().map(|m| m.agent_id).collect();
3870            self.agent_set = Some(built);
3871        }
3872        let agents: Vec<String> = self
3873            .agent_set
3874            .as_ref()
3875            .expect("agent_set just filled above")
3876            .iter()
3877            .cloned()
3878            .collect();
3879        let file_bytes = self.header.next_page * PAGE_SIZE as u64;
3880        Ok(Stats {
3881            memories: self.len(),
3882            deleted,
3883            dimensions: self.dimensions,
3884            file_bytes,
3885            agents,
3886            encrypted: self.header.flags & FLAG_ENCRYPTED != 0,
3887            created_at: self.header.created_at,
3888            index: self.ann.as_ref().map(|a| a.info()),
3889            wal_pages: self.header.wal_pages,
3890            fsyncs_last_flush: self.fsyncs_last_flush,
3891        })
3892    }
3893
3894    /// Rewrite the file, dropping tombstoned and expired memories and
3895    /// reclaiming stale pages left by updates. Done out-of-place via a temp
3896    /// file and an atomic rename.
3897    ///
3898    /// FIX-1 R-001.1 + R-002.2 part 2: the result cache now survives
3899    /// compaction. Every live, non-expired directory entry is copied
3900    /// into the new file verbatim — `created_at`, `ttl_secs`,
3901    /// `cost_hint_ms`, `key_preview`, `size_bytes`, `content_type`,
3902    /// `vector` and `model` come from the record body; `accessed_at`
3903    /// and `access_count` come from the *directory* (the record body
3904    /// only holds the insert-time snapshot — the directory is the
3905    /// source of truth for the live counts, as of v5's recall trick).
3906    /// Tombstoned and TTL-expired entries are dropped. Pre-FIX-1
3907    /// compaction kept only the memory catalog and silently emptied
3908    /// the cache; the Planner's interim README caveat (c417d2e) called
3909    /// that out — this PR removes the surprise.
3910    pub fn compact_file(path: &str, passphrase: &str) -> Result<CompactReport> {
3911        let mut old = Mnemo::open(path, passphrase)?;
3912        let dims = old.dimensions;
3913        let kdf = old.kdf;
3914        let tmp = format!("{path}.compact-tmp");
3915
3916        let mut new = Mnemo::create(
3917            &tmp,
3918            passphrase,
3919            MnemoConfig {
3920                dimensions: dims,
3921                kdf,
3922                ..Default::default()
3923            },
3924        )?;
3925        // ANN-1.1: compaction is "re-tune for the current size" by
3926        // default. Pass IndexConfig::default() so partitions,
3927        // subspaces, n_probe and n_rerank are all re-derived from the
3928        // compacted vector count. Previously we froze the pre-compact
3929        // (n_probe, n_rerank), which left the new index on stale
3930        // numbers for the smaller post-compact corpus.
3931        let want_index = old.ann.is_some();
3932        let now = memory::now_secs();
3933        let all = old.memories()?;
3934        let before = all.len();
3935        let mut after = 0;
3936        for m in all {
3937            if m.is_expired(now) {
3938                continue;
3939            }
3940            new.remember(m)?;
3941            after += 1;
3942        }
3943        if want_index && after > 0 {
3944            new.build_index_with(IndexConfig::default())?;
3945        }
3946
3947        // Count + copy cache entries. We walk `old.cache_dir` by index
3948        // so each hop reads the directory's live `accessed_at` /
3949        // `access_count` directly rather than going through
3950        // `cache_get` (which would also treat read-only semantics,
3951        // tombstones, etc. — we want the raw, authoritative values).
3952        let mut cache_before = 0usize;
3953        let mut cache_after = 0usize;
3954        let positions: Vec<usize> = (0..old.cache_dir.len()).collect();
3955        for pos in positions {
3956            let keep = {
3957                let e = &old.cache_dir[pos];
3958                !e.deleted && !e.is_expired(now)
3959            };
3960            if keep {
3961                cache_before += 1;
3962                new.copy_cache_entry_from(&mut old, pos)?;
3963                cache_after += 1;
3964            }
3965        }
3966
3967        new.flush()?;
3968        drop(new);
3969        drop(old);
3970
3971        std::fs::rename(&tmp, path)?;
3972        Ok(CompactReport {
3973            before,
3974            after,
3975            cache_before,
3976            cache_after,
3977        })
3978    }
3979
3980    /// Copy one live cache entry from `old` at directory position
3981    /// `pos` into `self` (a freshly-created compaction target).
3982    /// Decrypts the old record body, substitutes the old *directory*'s
3983    /// `accessed_at` + `access_count` (the record body's copies are
3984    /// stale as of v5's directory-only hit path), re-encodes, writes
3985    /// a fresh record run, and pushes a matching directory entry.
3986    ///
3987    /// Caller is expected to have already filtered out tombstoned /
3988    /// expired entries and to call this on a read-write handle.
3989    pub(crate) fn copy_cache_entry_from(&mut self, old: &mut Mnemo, pos: usize) -> Result<()> {
3990        if self.read_only {
3991            return Err(MnemoError::ReadOnly);
3992        }
3993        // Snapshot the directory fields we need, so the borrow on
3994        // `old.cache_dir` ends before we touch `old.pager`.
3995        let (start_page, page_count, len, dir_accessed_at, dir_access_count) = {
3996            let e = &old.cache_dir[pos];
3997            (
3998                e.start_page,
3999                e.page_count,
4000                e.len,
4001                e.accessed_at,
4002                e.access_count,
4003            )
4004        };
4005
4006        // Read + decrypt the old record body.
4007        let mut buf = Vec::with_capacity(len as usize);
4008        for i in 0..page_count as u64 {
4009            buf.extend_from_slice(&old.pager.read_page(start_page + i)?);
4010        }
4011        buf.truncate(len as usize);
4012        let mut entry: CacheEntry =
4013            rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
4014
4015        // Overwrite the stale body copies with the directory's live
4016        // values before persisting. Everything else — `created_at`,
4017        // `ttl_secs`, `cost_hint_ms`, `key_preview`, `size_bytes`,
4018        // `content_type`, `value`, `vector`, `model` — stays as the
4019        // body already recorded it.
4020        entry.accessed_at = dir_accessed_at;
4021        entry.access_count = dir_access_count;
4022
4023        let bytes =
4024            rmp_serde::to_vec(&entry).map_err(|err| MnemoError::Serialize(err.to_string()))?;
4025        let (new_start_page, new_page_count) = self.write_record(&bytes)?;
4026
4027        let dir_entry = CacheDirectoryEntry {
4028            namespace: entry.namespace.clone(),
4029            key_hash: entry.key_hash,
4030            start_page: new_start_page,
4031            page_count: new_page_count,
4032            len: bytes.len() as u32,
4033            deleted: false,
4034            accessed_at: dir_accessed_at,
4035            access_count: dir_access_count,
4036            size_bytes: entry.size_bytes,
4037            ttl_secs: entry.ttl_secs,
4038            created_at: entry.created_at,
4039            vector: entry.vector.clone(),
4040            model: entry.model.clone(),
4041        };
4042        let new_pos = self.cache_dir.len();
4043        self.cache_idx
4044            .insert(&entry.namespace, entry.key_hash, new_pos);
4045        self.cache_dir.push(dir_entry);
4046        self.dirty_cache = true;
4047        Ok(())
4048    }
4049}