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::fs::{File, OpenOptions};
25use std::io::{Read, Seek, SeekFrom, Write};
26use std::path::{Path, PathBuf};
27
28use fs4::fs_std::FileExt;
29use serde::{Deserialize, Serialize};
30use serde_json::{Map, Value};
31use ulid::Ulid;
32
33use crate::crypto::{self, KdfParams};
34use crate::embed::{self, Embedder};
35use crate::error::{MnemoError, Result};
36use crate::format::{Header, FLAG_ENCRYPTED, PAGE_SIZE, PAYLOAD, VERSION, WRAPPED_DEK_LEN};
37use crate::index::{IndexConfig, IndexInfo, IvfPqIndex};
38use crate::memory::{
39    self, status as essence_status, CommitBatch, CommitReport, EssenceOp, EssenceOpKind,
40    EssenceScope, EssenceSection, ExportedTurn, Filter, FilterTerm, IngestReport, KeyScope, Memory,
41    MemoryPatch, MemoryType, Metric, Scope, ScoreWeights, ThreadTurn, TurnPolicy, WriteOutcome,
42    CONTINUATION_TIMELINE_ENTRIES, ESSENCE_DEFAULT_RENDER_BYTES, ESSENCE_TRANSITIONS,
43    MNEMO_META_CONTINUES, MNEMO_META_INVALID_AT, MNEMO_META_KEY, MNEMO_META_KIND,
44    MNEMO_META_KIND_ESSENCE, MNEMO_META_KIND_THREAD_LINK, MNEMO_META_KIND_TURN, MNEMO_META_LENGTH,
45    MNEMO_META_PART, MNEMO_META_REFS, MNEMO_META_ROLE, MNEMO_META_SECTION, MNEMO_META_SEQ,
46    MNEMO_META_SHA256, MNEMO_META_STATUS, MNEMO_META_STATUS_HISTORY, MNEMO_META_SUPERSEDED_BY,
47    MNEMO_META_SUPERSEDES, MNEMO_META_THREAD, MNEMO_META_TURN_ID, MNEMO_META_UPDATED_AT,
48    THREAD_TOOL_RESULT_TRUNCATE_BYTES, THREAD_TURN_CHUNK_BYTES,
49};
50use crate::pager::Pager;
51use crate::result_cache::{
52    self, BatchState, CacheBudget, CacheDirectoryEntry, CacheEntry, CacheFlushPolicy, CachePutOpts,
53    CacheStats, CachedValue, CounterState, DirectoryIndex, SemanticCachePutOpts,
54};
55use crate::wal;
56
57/// Default initial size of the write-ahead log region, in pages (64 KiB).
58/// The region grows automatically when a transaction's control plane outgrows
59/// it, so this is a *floor* on the fresh-file footprint, not a cap on
60/// transaction size. Was 64 pages (512 KiB) in v0.1.0 — lowered because that
61/// reservation dominated small-file size (~62% of a 31-memory dogfood file).
62const DEFAULT_WAL_PAGES: u64 = 8;
63
64/// Hard floor on the WAL reservation. A single header-page flush already
65/// needs one full DATA frame plus a COMMIT frame; two pages are the bare
66/// minimum that lets the first transaction commit without immediate growth.
67const MIN_WAL_PAGES: u64 = 2;
68
69/// Default cap on retained snapshot manifest entries. Each `flush` appends
70/// one entry; without a cap, long-running processes accumulate entries
71/// forever, growing per-flush manifest-serialize cost as O(total flushes)
72/// and the manifest run size on disk in lockstep. 256 keeps roughly a
73/// week of hourly snapshots while staying tiny (256 × ~82 bytes ≈ 21 KiB
74/// of manifest). Set `MnemoConfig::max_snapshots = 0` to disable the cap.
75pub const DEFAULT_MAX_SNAPSHOTS: usize = 256;
76
77/// PERF-1: the ANN index is derived data, so `flush` persists it only every
78/// this many commits that changed it (or see
79/// [`INDEX_PERSIST_EVERY_CHANGES`], [`Mnemo::flush_index`], [`Mnemo::close`]).
80/// Between persists, only the catalog is rewritten.
81pub const INDEX_PERSIST_EVERY_COMMITS: u32 = 64;
82
83/// PERF-1: …or once this many vectors were added, changed or removed since
84/// the last persist. Records written after the persisted index are
85/// re-assigned when the index is next loaded, so this bounds that work.
86pub const INDEX_PERSIST_EVERY_CHANGES: usize = 1_000;
87
88/// PERF-1: when the in-memory ANN index was last persisted, and whether the
89/// next flush must persist it.
90#[derive(Debug, Default, Clone, Copy)]
91struct IndexPersist {
92    /// Persist on the next flush (build, rebuild, drop, `flush_index`,
93    /// `close`, snapshot restore).
94    forced: bool,
95    /// Commits since the last persist that left the index dirty.
96    commits: u32,
97    /// Vectors added, re-assigned or removed since the last persist.
98    changes: usize,
99}
100
101/// R-004.2: where the thread layers and the onboarding briefing live,
102/// built by the handle's one lazy scan (`ensure_key_caches`, together with
103/// the 12.1 key caches) and then kept current by `put` and `delete`. It
104/// lets `about`, `essence`, `ingest_thread`, `thread_export` and
105/// thread-pinned recall read just their own records instead of decrypting
106/// the whole file each time.
107#[derive(Debug, Default)]
108struct LayerCache {
109    /// Live essence item ids.
110    essence: std::collections::HashSet<u128>,
111    /// Live ids tagged `metadata.area = "onboarding"` (superseded versions
112    /// included; `about` drops those when it reads them).
113    onboarding: std::collections::HashSet<u128>,
114    /// Per `_mnemo.thread`: its records and archive state.
115    threads: std::collections::HashMap<String, ThreadCache>,
116    /// Thread → (the thread it continues, the link record's id).
117    links: std::collections::HashMap<String, (String, u128)>,
118}
119
120/// One thread's records (turns, essence items, link), the turn ids already
121/// archived (with how many records each has: long turns are chunked), and
122/// the highest `_mnemo.seq` seen.
123#[derive(Debug, Default)]
124struct ThreadCache {
125    records: std::collections::HashSet<u128>,
126    turns: std::collections::HashMap<String, u32>,
127    max_seq: Option<u64>,
128}
129
130fn meta_str<'a>(m: &'a Memory, key: &str) -> Option<&'a str> {
131    m.metadata.get(key).and_then(|v| v.as_str())
132}
133
134fn is_onboarding(m: &Memory) -> bool {
135    meta_str(m, "area").is_some_and(|s| s.eq_ignore_ascii_case("onboarding"))
136}
137
138impl LayerCache {
139    fn add(&mut self, m: &Memory) {
140        let id = m.id.0;
141        let kind = meta_str(m, MNEMO_META_KIND);
142        if kind == Some(MNEMO_META_KIND_ESSENCE) {
143            self.essence.insert(id);
144        }
145        if is_onboarding(m) {
146            self.onboarding.insert(id);
147        }
148        let Some(thread) = meta_str(m, MNEMO_META_THREAD) else {
149            return;
150        };
151        let tc = self.threads.entry(thread.to_string()).or_default();
152        tc.records.insert(id);
153        if kind == Some(MNEMO_META_KIND_TURN) {
154            if let Some(turn_id) = meta_str(m, MNEMO_META_TURN_ID) {
155                *tc.turns.entry(turn_id.to_string()).or_insert(0) += 1;
156            }
157            if let Some(seq) = m.metadata.get(MNEMO_META_SEQ).and_then(|v| v.as_u64()) {
158                tc.max_seq = Some(tc.max_seq.map_or(seq, |s| s.max(seq)));
159            }
160        }
161        if kind == Some(MNEMO_META_KIND_THREAD_LINK) {
162            if let Some(from) = meta_str(m, MNEMO_META_CONTINUES) {
163                self.links
164                    .entry(thread.to_string())
165                    .or_insert_with(|| (from.to_string(), id));
166            }
167        }
168    }
169
170    /// Undo [`LayerCache::add`] for the stored version of a record that is
171    /// being overwritten or deleted. A thread with no records left is
172    /// dropped, so a forgotten thread's `_mnemo.seq` restarts at 0. Partial
173    /// deletes keep the old maximum (sequence gaps are harmless).
174    fn remove(&mut self, m: &Memory) {
175        let id = m.id.0;
176        self.essence.remove(&id);
177        self.onboarding.remove(&id);
178        let Some(thread) = meta_str(m, MNEMO_META_THREAD) else {
179            return;
180        };
181        if self.links.get(thread).is_some_and(|(_, link)| *link == id) {
182            self.links.remove(thread);
183        }
184        let Some(tc) = self.threads.get_mut(thread) else {
185            return;
186        };
187        tc.records.remove(&id);
188        if meta_str(m, MNEMO_META_KIND) == Some(MNEMO_META_KIND_TURN) {
189            if let Some(turn_id) = meta_str(m, MNEMO_META_TURN_ID) {
190                if let Some(n) = tc.turns.get_mut(turn_id) {
191                    *n -= 1;
192                    if *n == 0 {
193                        tc.turns.remove(turn_id);
194                    }
195                }
196            }
197        }
198        if tc.records.is_empty() {
199            self.threads.remove(thread);
200        }
201    }
202}
203
204/// The thread a recall filter pins with `_mnemo.thread` equality, if any.
205fn filter_thread_pin(f: &Filter) -> Option<&str> {
206    f.terms.iter().find_map(|t| match t {
207        FilterTerm::Eq(k, Value::String(s)) if k == MNEMO_META_THREAD => Some(s.as_str()),
208        _ => None,
209    })
210}
211
212/// Whether a recall filter can only match records that aren't in the ANN
213/// (R-004.2): it pins `_mnemo.thread`, or restricts `_mnemo.kind` to turns,
214/// essence items or thread links. Such recalls scan exactly instead.
215fn filter_bypasses_ann(f: &Filter) -> bool {
216    let non_ann = |v: &Value| {
217        matches!(
218            v.as_str(),
219            Some(MNEMO_META_KIND_TURN)
220                | Some(MNEMO_META_KIND_ESSENCE)
221                | Some(MNEMO_META_KIND_THREAD_LINK)
222        )
223    };
224    filter_thread_pin(f).is_some()
225        || f.terms.iter().any(|t| match t {
226            FilterTerm::Eq(k, v) => k == MNEMO_META_KIND && non_ann(v),
227            FilterTerm::In(k, vs) => {
228                k == MNEMO_META_KIND && !vs.is_empty() && vs.iter().all(non_ann)
229            }
230            _ => false,
231        })
232}
233
234/// Chunk size for the torn-header scan's streaming read of the file.
235/// We walk the file 1 MiB at a time looking for 8 KiB-aligned `MWAL`
236/// magic. There is **no** cap on how far a given candidate's frame
237/// chain may extend — a 1M-vector IVF+PQ relocation can push a single
238/// transaction well past 16 MiB — so each candidate is read frame by
239/// frame via [`wal::scan_committed_txn`] rather than slurped into a
240/// bounded buffer. (DUR-1 R-002.1 part 2.)
241const TORN_HEADER_SCAN_CHUNK: usize = 1 << 20; // 1 MiB
242
243/// Walk the file for every 8 KiB-aligned page whose first four bytes are
244/// `MWAL`, treating each as a candidate WAL-region start, and return the
245/// frames of the highest-`txn_id` committed transaction whose header
246/// frame fully validates under `passphrase`.
247///
248/// A candidate is accepted only if:
249/// * `wal::scan_committed_txn` returns a committed transaction (chained
250///   CRC verifies, single `txn_id`, `KIND_COMMIT` closes it);
251/// * that transaction contains a `(0, header_page)` frame;
252/// * the header parses with `wal_start == candidate_page_no` and
253///   `wal_seq == txn_id` (so a frame cannot forge a region it was not
254///   written into);
255/// * deriving the KEK from `passphrase + header.salt`, unwrapping the
256///   wrapped DEK, and validating the header seal all succeed.
257///
258/// Returns `Ok(None)` when the scan finds no valid candidate — the
259/// caller is expected to propagate the original parse error so
260/// non-mnemo files still report `BadMagic`.
261///
262/// DUR-1 (R-002.1 part 2): there is no fixed scan window — the whole
263/// file is scanned. Reading at page granularity (not byte granularity)
264/// is sufficient because every WAL region in the format is page-aligned
265/// and every WAL commit writes its first frame at the region's start.
266fn scan_header_recovery_candidates(
267    file: &mut File,
268    secret: &[u8],
269) -> Result<Option<Vec<wal::Frame>>> {
270    let file_len = file.seek(SeekFrom::End(0))?;
271    if file_len < PAGE_SIZE as u64 * 2 {
272        return Ok(None); // No room for even one WAL page past the header.
273    }
274
275    let mut best: Option<(u64, Vec<wal::Frame>)> = None;
276    let page_sz = PAGE_SIZE as u64;
277    let mut offset = page_sz; // Skip page 0.
278
279    // Stream through the file in 1 MiB chunks, scanning each chunk for
280    // page-aligned `MWAL` candidates. Each candidate's frame chain is
281    // then read incrementally via `wal::scan_committed_txn`, which has
282    // no window cap — a 1M-vector IVF+PQ relocation can span tens of
283    // MiB in one committed transaction.
284    let mut chunk = vec![0u8; TORN_HEADER_SCAN_CHUNK];
285    while offset < file_len {
286        let want = (TORN_HEADER_SCAN_CHUNK as u64).min(file_len - offset) as usize;
287        chunk.resize(want, 0);
288        file.seek(SeekFrom::Start(offset))?;
289        let mut filled = 0;
290        while filled < want {
291            match file.read(&mut chunk[filled..])? {
292                0 => break,
293                n => filled += n,
294            }
295        }
296        if filled == 0 {
297            break;
298        }
299        let mut inner = 0usize;
300        while inner + 4 <= filled {
301            if &chunk[inner..inner + 4] == wal::FRAME_MAGIC_BYTES {
302                let page_no = (offset + inner as u64) / page_sz;
303                if let Some((txn_id, frames)) = try_recover_candidate(file, page_no, secret)? {
304                    let keep = match &best {
305                        None => true,
306                        Some((id, _)) => txn_id > *id,
307                    };
308                    if keep {
309                        best = Some((txn_id, frames));
310                    }
311                }
312            }
313            inner += PAGE_SIZE;
314        }
315        offset += filled as u64;
316    }
317
318    Ok(best.map(|(_, f)| f))
319}
320
321/// Try to recover a committed transaction from `candidate_page_no`,
322/// returning `(txn_id, frames)` only if the candidate's
323/// `(0, header_page)` frame fully validates: the frame chain commits
324/// (via [`wal::scan_committed_txn`]'s streaming reader), the header
325/// parses with `wal_start == candidate_page_no` **and**
326/// `wal_seq == txn_id`, the DEK unwraps under `passphrase`, and the
327/// AES-GCM seal validates. DUR-1 (R-002.1 item 2) requires the
328/// `wal_seq == txn_id` check and no fixed scan window — the streaming
329/// reader handles transactions of arbitrary size.
330fn try_recover_candidate(
331    file: &mut File,
332    candidate_page_no: u64,
333    secret: &[u8],
334) -> Result<Option<(u64, Vec<wal::Frame>)>> {
335    let Some((txn_id, frames)) = wal::scan_committed_txn(file, candidate_page_no)? else {
336        return Ok(None);
337    };
338
339    // Must contain a header frame at page 0.
340    let header_bytes = match frames.iter().find(|(p, _)| *p == 0) {
341        Some((_, b)) => b,
342        None => return Ok(None),
343    };
344    if header_bytes.len() != PAGE_SIZE {
345        return Ok(None);
346    }
347    let mut hbuf = [0u8; PAGE_SIZE];
348    hbuf.copy_from_slice(header_bytes);
349    let candidate = match Header::from_page(&hbuf) {
350        Ok(h) => h,
351        Err(_) => return Ok(None),
352    };
353
354    // The frame's claimed region must match where we found it, and the
355    // header frame's `wal_seq` must match the committed transaction's
356    // `txn_id` — this is what pins the frame to the region it was
357    // written into and keeps a stale checkpointed-and-copied frame
358    // from being accepted as a new region's root.
359    if candidate.wal_start != candidate_page_no || candidate.wal_seq != txn_id {
360        return Ok(None);
361    }
362    // DEK unwrap under this candidate's own salt + KDF params.
363    let kdf = KdfParams {
364        m_cost: candidate.m_cost,
365        t_cost: candidate.t_cost,
366        p_cost: candidate.p_cost,
367    };
368    let kek = match crypto::derive_kek(secret, &candidate.salt, kdf) {
369        Ok(k) => k,
370        Err(_) => return Ok(None),
371    };
372    let dek = match crypto::unwrap_dek(&kek, &candidate.dek_nonce, &candidate.wrapped_dek) {
373        Ok(d) => d,
374        Err(_) => return Ok(None),
375    };
376    if candidate.validate_seal(&dek).is_err() {
377        return Ok(None);
378    }
379
380    Ok(Some((txn_id, frames)))
381}
382
383/// Configuration for creating a new database.
384#[derive(Clone, Copy, Debug)]
385pub struct MnemoConfig {
386    /// Embedding dimensionality. Every stored vector must match this.
387    pub dimensions: usize,
388    /// Key-derivation parameters.
389    pub kdf: KdfParams,
390    /// Initial size of the WAL region in 8 KiB pages. Default is
391    /// [`DEFAULT_WAL_PAGES`] (8 pages = 64 KiB), which keeps fresh files
392    /// small; the region auto-grows when a transaction's control plane needs
393    /// more space, so raising this is only a hint for the expected steady-state
394    /// transaction size. Write-heavy databases that routinely flush large
395    /// catalogs or ANN indexes can set this higher (e.g. 64 = 512 KiB) to
396    /// avoid early grow events; tiny embedded uses can leave it at the
397    /// default. Clamped to a minimum of [`MIN_WAL_PAGES`] (2 pages).
398    pub wal_pages_initial: u64,
399    /// Maximum number of snapshot manifest entries to retain. Each `flush`
400    /// appends one entry; once the cap is hit, the *oldest* entry is dropped
401    /// from the in-memory manifest before the new one is added, so the
402    /// retained set is always the most-recent N. Defaults to
403    /// [`DEFAULT_MAX_SNAPSHOTS`] (256). Set to `0` to disable the cap and
404    /// retain every snapshot forever (the pre-v0.3 behavior).
405    ///
406    /// Pages referenced *only* by pruned snapshots stay on disk until the
407    /// next [`Mnemo::compact_file`], which reclaims them. Point-in-time
408    /// recovery via [`Mnemo::restore_to`] into a pruned `txn_id` returns
409    /// [`MnemoError::NotFound`].
410    ///
411    /// Apply to an already-open database with [`Mnemo::set_max_snapshots`].
412    pub max_snapshots: usize,
413}
414
415impl Default for MnemoConfig {
416    fn default() -> Self {
417        Self {
418            dimensions: 768,
419            kdf: KdfParams::secure(),
420            wal_pages_initial: DEFAULT_WAL_PAGES,
421            max_snapshots: DEFAULT_MAX_SNAPSHOTS,
422        }
423    }
424}
425
426/// Catalog entry: maps a memory ID to its page run on disk.
427///
428/// **Schema (v5+):** carries `accessed_at` and `access_count` so `recall`
429/// can update access stats by mutating the catalog entry rather than
430/// rewriting the entire record (Phase 2.1 of the improvement plan). The
431/// values in a memory's serialized record body become a stale snapshot
432/// after the first recall; [`Mnemo::read_memory`] re-populates them from
433/// the catalog so consumers always see the live values.
434#[derive(Serialize, Deserialize, Clone, Debug)]
435struct CatalogEntry {
436    /// ULID as a raw `u128`.
437    id: u128,
438    start_page: u64,
439    page_count: u32,
440    /// Exact serialized byte length of the record.
441    len: u32,
442    deleted: bool,
443    /// Unix seconds of the most recent recall hit (or write, for fresh entries).
444    accessed_at: i64,
445    /// How many times `recall` has surfaced this memory.
446    access_count: u32,
447}
448
449/// Frozen v4 catalog entry shape, used only by the v4→v5 migration path in
450/// [`Mnemo::open`]. Its layout matches the catalog encoding written by
451/// pre-v5 builds (5 positional fields). Do not change this struct.
452#[derive(Serialize, Deserialize, Clone, Debug)]
453struct CatalogEntryV4 {
454    id: u128,
455    start_page: u64,
456    page_count: u32,
457    len: u32,
458    deleted: bool,
459}
460
461/// Returned by [`Mnemo::prepare_for_flush`] and consumed by [`Mnemo::flush`].
462/// Carries the already-serialized control-plane bytes through the prelude
463/// so we don't re-serialize after the lease has been persisted.
464struct FlushPrelude {
465    cat_bytes: Option<Vec<u8>>,
466    idx_bytes: Option<Vec<u8>>,
467    /// Serialized result-cache directory bytes when `dirty_cache` is
468    /// set (Phase 10.1). `None` if the cache didn't change since the
469    /// last flush — in that case the header keeps its current
470    /// `cache_start` / `cache_pages` / `cache_len` and no fresh cache
471    /// run is written.
472    cache_bytes: Option<Vec<u8>>,
473}
474
475/// One entry in the append-only snapshot manifest. Because record, catalog,
476/// and index pages are only ever *appended*, the runs a past flush wrote are
477/// still on disk; a `Snapshot` is the set of pointers needed to reconstruct
478/// the database exactly as that flush left it.
479#[derive(Serialize, Deserialize, Clone, Debug)]
480struct Snapshot {
481    txn_id: u64,
482    created_at: i64,
483    catalog_start: u64,
484    catalog_pages: u64,
485    catalog_len: u64,
486    index_start: u64,
487    index_pages: u64,
488    index_len: u64,
489    memory_count: u64,
490}
491
492/// A restorable point in a database's history — see [`Mnemo::snapshots`].
493#[derive(Clone, Copy, Debug)]
494pub struct SnapshotInfo {
495    /// Id of the transaction that produced this snapshot (monotonic from 1).
496    pub txn_id: u64,
497    /// When the snapshot was committed (unix seconds).
498    pub created_at: i64,
499    /// Live memory count captured in the snapshot.
500    pub memory_count: u64,
501}
502
503/// A retrieval request for [`Mnemo::recall`].
504#[derive(Clone, Debug)]
505pub struct RecallRequest {
506    /// Query embedding.
507    pub query: Vec<f32>,
508    /// Maximum results to return.
509    pub top_k: usize,
510    /// Restrict to these memory types (`None` = all types).
511    pub memory_types: Option<Vec<MemoryType>>,
512    /// Restrict to a single agent's view: its own memories plus shared ones.
513    pub agent_id: Option<String>,
514    /// Similarity metric.
515    pub metric: Metric,
516    /// Multi-signal score weights.
517    pub weights: ScoreWeights,
518    /// Index override: partitions to probe (`None` = index default). Ignored
519    /// when no ANN index is present.
520    pub n_probe: Option<usize>,
521    /// Index override: candidates to rerank (`None` = index default). Ignored
522    /// when no ANN index is present.
523    pub n_rerank: Option<usize>,
524    /// Whether to update each result's `accessed_at` and `access_count` in
525    /// the catalog. Defaults to `true`, matching pre-v5 behavior.
526    ///
527    /// Set to `false` for **fully read-only recall** — useful for batch
528    /// scoring, dry-run scoring, or recalls run by introspection tooling
529    /// (e.g. `mnemo about` consumers) where you don't want the score's own
530    /// observation to perturb the database. With `false`, recall does not
531    /// dirty the catalog and the next `flush` is a no-op.
532    pub track_access: bool,
533    /// Phase 12.6 pre-scoring filter: `session_id` + AND-ed metadata
534    /// conditions applied before the candidate is scored. Default is
535    /// [`Filter::default`] (no constraints). When the ANN index is on
536    /// and this filter is selective enough to starve the shortlist
537    /// below `top_k`, Phase 12.1's widen-then-exact-scan logic kicks
538    /// in so the result still carries up to `top_k` correct hits.
539    pub filter: Filter,
540    /// Phase 12.8 post-scoring budget: stop adding hits once the UTF-8
541    /// bytes of their `content` fields (not the serialized record
542    /// body) would exceed this many bytes. `None` (default) means no
543    /// budget and every hit through `top_k` is returned. If the top-1
544    /// hit is already over budget it is still returned — callers that
545    /// need an absolute cap should also pre-size `top_k`.
546    ///
547    /// Counts only the UTF-8 `content` bytes so the number is stable
548    /// across the Rust / Python / CLI / MCP surfaces; the JSON or
549    /// MessagePack envelope is excluded. Roughly 4 bytes per BPE
550    /// token for English text — a 4 KB budget comfortably holds ~1K
551    /// tokens.
552    ///
553    /// Observable through [`RecallResponse::truncated`]
554    /// ([`Mnemo::recall_budgeted`]): `true` iff at least one scored
555    /// hit was dropped by this budget, `false` otherwise.
556    pub max_bytes: Option<usize>,
557}
558
559impl RecallRequest {
560    /// A request with sensible defaults for a query embedding.
561    pub fn new(query: Vec<f32>) -> Self {
562        Self {
563            query,
564            top_k: 10,
565            memory_types: None,
566            agent_id: None,
567            metric: Metric::Cosine,
568            weights: ScoreWeights::default(),
569            n_probe: None,
570            n_rerank: None,
571            track_access: true,
572            filter: Filter::default(),
573            max_bytes: None,
574        }
575    }
576
577    /// Phase 12.6: attach a pre-scoring filter. See [`Filter`] and
578    /// [`FilterTerm`] for term shapes.
579    pub fn filter(mut self, filter: Filter) -> Self {
580        self.filter = filter;
581        self
582    }
583
584    /// Phase 12.8: cap the total UTF-8 bytes of returned `content`
585    /// fields. See [`RecallRequest::max_bytes`].
586    pub fn max_bytes(mut self, n: usize) -> Self {
587        self.max_bytes = Some(n);
588        self
589    }
590
591    /// Build a request by embedding `query_text` with the supplied
592    /// embedder (Phase 3). Shorthand for `embedder.embed(&[text])` +
593    /// [`RecallRequest::new`]; same defaults as `new`. Fails if the
594    /// embedder returns an empty vector list or an embedding failure.
595    pub fn from_text(embedder: &dyn Embedder, query_text: &str) -> Result<Self> {
596        let mut vectors = embedder.embed(&[query_text])?;
597        let v = vectors.pop().ok_or_else(|| {
598            MnemoError::Invalid("embedder returned no vector for the query text".into())
599        })?;
600        Ok(Self::new(v))
601    }
602    /// Set the result cap.
603    pub fn top_k(mut self, k: usize) -> Self {
604        self.top_k = k;
605        self
606    }
607    /// Restrict to specific memory types.
608    pub fn types(mut self, t: Vec<MemoryType>) -> Self {
609        self.memory_types = Some(t);
610        self
611    }
612    /// Restrict to one agent's view.
613    pub fn agent(mut self, agent_id: impl Into<String>) -> Self {
614        self.agent_id = Some(agent_id.into());
615        self
616    }
617    /// Set the similarity metric (default: cosine).
618    pub fn metric(mut self, metric: Metric) -> Self {
619        self.metric = metric;
620        self
621    }
622    /// Replace the multi-signal score weights.
623    pub fn weights(mut self, weights: ScoreWeights) -> Self {
624        self.weights = weights;
625        self
626    }
627    /// Override the number of IVF partitions probed (accuracy/speed dial).
628    pub fn n_probe(mut self, n: usize) -> Self {
629        self.n_probe = Some(n);
630        self
631    }
632    /// Override the number of candidates reranked exactly (accuracy dial).
633    pub fn n_rerank(mut self, n: usize) -> Self {
634        self.n_rerank = Some(n);
635        self
636    }
637    /// Toggle whether recall updates `accessed_at` / `access_count` on
638    /// the returned memories' catalog entries (default `true`). Pass
639    /// `false` for a fully read-only recall — see the field doc.
640    pub fn track_access(mut self, track: bool) -> Self {
641        self.track_access = track;
642        self
643    }
644}
645
646/// One scored result from [`Mnemo::recall`].
647#[derive(Clone, Debug)]
648pub struct RecallResult {
649    /// The retrieved memory.
650    pub memory: Memory,
651    /// Combined multi-signal score.
652    pub score: f32,
653    /// The bare similarity component (before other signals).
654    pub similarity: f32,
655}
656
657/// Phase 12.8 budgeted-recall envelope returned by
658/// [`Mnemo::recall_budgeted`]. Wraps the `Vec<RecallResult>` the
659/// legacy [`Mnemo::recall`] returns so callers can tell whether
660/// [`RecallRequest::max_bytes`] cut anything.
661#[derive(Clone, Debug)]
662pub struct RecallResponse {
663    /// Scored hits, best first, truncated to at most `top_k` and then
664    /// to fit the `max_bytes` budget if set.
665    pub hits: Vec<RecallResult>,
666    /// `true` iff the `max_bytes` budget dropped at least one hit
667    /// that would otherwise have been returned (i.e. the result is
668    /// shorter than what `top_k` alone would produce).
669    pub truncated: bool,
670    /// Phase 12.1 + 12.6 observability: `true` iff recall fell
671    /// through to an exact scan (ANN shortlist too narrow after
672    /// filtering, and the widen pass also came up short, so the
673    /// index was skipped for one full-corpus pass). Monitoring on
674    /// this flag is how callers detect selective-filter regressions
675    /// or supersession chains outgrowing the ANN shortlist.
676    pub used_exact_scan: bool,
677}
678
679/// Where a file's bytes and an open handle's heap go (P-0 device budget,
680/// `examples/device_budget.rs`). **Unstable diagnostic**: fields may
681/// change in any release. File figures come from the header and the
682/// catalog. RAM figures are buffer capacities, with hash maps estimated
683/// from their capacity; they cover the engine's main structures, not
684/// every allocation, and not allocator overhead.
685#[doc(hidden)]
686#[derive(Clone, Debug, Default, serde::Serialize)]
687pub struct BudgetReport {
688    pub file_bytes: u64,
689    pub page_size: u64,
690    /// Page 0.
691    pub header_bytes: u64,
692    /// Whole pages holding live records (superseded versions included).
693    pub record_page_bytes: u64,
694    /// Serialized bytes of those records; the rest of their pages is
695    /// slack.
696    pub record_payload_bytes: u64,
697    /// Pages still holding soft-deleted records (reclaimed by compact).
698    pub deleted_record_page_bytes: u64,
699    pub catalog_run_bytes: u64,
700    pub catalog_payload_bytes: u64,
701    pub index_run_bytes: u64,
702    pub index_payload_bytes: u64,
703    /// First page of the persisted index run (0 = none).
704    pub index_start_page: u64,
705    pub wal_bytes: u64,
706    pub snapshot_manifest_bytes: u64,
707    pub cache_dir_run_bytes: u64,
708    /// Everything else: superseded catalog, index and directory runs from
709    /// earlier flushes, and freed pages, until the next compact.
710    pub unreferenced_bytes: u64,
711    pub live_records: u64,
712    pub ram_catalog: u64,
713    pub ram_id_map: u64,
714    pub ram_ann_centroids: u64,
715    pub ram_ann_codebook: u64,
716    pub ram_ann_posting_ids: u64,
717    pub ram_ann_posting_codes: u64,
718    pub ram_ann_assignment: u64,
719    pub ram_cache_dir: u64,
720    pub ram_key_caches: u64,
721    /// Sum of the `ram_*` fields.
722    pub ram_total: u64,
723}
724
725/// Summary statistics for a database.
726#[derive(Clone, Debug)]
727pub struct Stats {
728    /// Live (non-deleted) memory count.
729    pub memories: usize,
730    /// Tombstoned entries awaiting compaction.
731    pub deleted: usize,
732    /// Embedding dimensionality.
733    pub dimensions: usize,
734    /// File size in bytes.
735    pub file_bytes: u64,
736    /// Distinct agent IDs present.
737    pub agents: Vec<String>,
738    /// Whether pages are encrypted (always true in v1).
739    pub encrypted: bool,
740    /// Creation time (unix seconds).
741    pub created_at: i64,
742    /// ANN index shape, if an index has been built.
743    pub index: Option<IndexInfo>,
744    /// Current size of the write-ahead log region, in 8 KiB pages.
745    pub wal_pages: u64,
746    /// Fsyncs issued by the most recent successful [`Mnemo::flush`] on
747    /// this handle. `0` on a fresh handle before any flush, or when
748    /// `flush()` short-circuited (nothing dirty). Phase 6 instrumentation
749    /// for the Phase 11.1 write-latency work — the planner wants this
750    /// before deciding whether cutting fsync count is a viable
751    /// durability-preserving fix.
752    pub fsyncs_last_flush: u32,
753}
754
755/// Result of a [`Mnemo::compact_file`] run.
756#[derive(Clone, Copy, Debug, Default)]
757pub struct CompactReport {
758    /// Live memories before compaction.
759    pub before: usize,
760    /// Live memories after compaction (expired ones dropped).
761    pub after: usize,
762    /// Live, non-expired cache entries before compaction. FIX-1
763    /// R-001.1: compaction now preserves the result cache instead of
764    /// silently dropping it.
765    pub cache_before: usize,
766    /// Live, non-expired cache entries after compaction (tombstoned
767    /// and TTL-expired ones dropped).
768    pub cache_after: usize,
769}
770
771/// An encrypted, single-file agent memory database.
772pub struct Mnemo {
773    pager: Pager,
774    header: Header,
775    catalog: Vec<CatalogEntry>,
776    index: IdIndex,
777    #[allow(dead_code)]
778    path: PathBuf,
779    dimensions: usize,
780    kdf: KdfParams,
781    /// Set whenever the catalog changes; drives whether `flush` rewrites it.
782    dirty_catalog: bool,
783    /// Optional IVF+PQ approximate-nearest-neighbour index.
784    ann: Option<IvfPqIndex>,
785    /// Set whenever the in-memory ANN index differs from the persisted one.
786    /// PERF-1: `flush` persists it only per [`IndexPersist`]'s policy.
787    dirty_index: bool,
788    /// PERF-1: index persistence policy state.
789    index_persist: IndexPersist,
790    /// PERF-1 step 3: the header points at a persisted ANN index that
791    /// hasn't been loaded yet. It loads on first use (`ensure_ann`).
792    ann_unloaded: bool,
793    /// Append-only manifest of committed snapshots, oldest first. Capped
794    /// at [`Mnemo::max_snapshots`] entries on every flush — the oldest
795    /// entries are pruned first.
796    manifest: Vec<Snapshot>,
797    /// Maximum manifest entries to retain across flushes. `0` disables
798    /// the cap (retain forever). Defaults to [`DEFAULT_MAX_SNAPSHOTS`].
799    max_snapshots: usize,
800    /// If `true`, this handle was opened via [`Mnemo::open_read_only`] and
801    /// every mutating API returns [`MnemoError::ReadOnly`]. The underlying
802    /// file was opened without write permission and holds an OS shared
803    /// lock; the OS would reject a write() anyway, but gating in Rust
804    /// gives callers a clean, typed error.
805    read_only: bool,
806    /// Result-cache directory (Phase 10.1). See [`crate::result_cache`].
807    /// Loaded from `header.cache_start` on open; persisted on `flush`
808    /// via the same seal_run pipeline as catalog/index/manifest.
809    cache_dir: Vec<CacheDirectoryEntry>,
810    /// In-memory index over `cache_dir` for O(1) `(namespace, key_hash)`
811    /// lookups. Rebuilt on open + kept in sync on `cache_put`/`delete`.
812    cache_idx: DirectoryIndex,
813    /// Set whenever `cache_dir` changes; drives whether `flush` rewrites
814    /// the cache directory run.
815    dirty_cache: bool,
816    /// Per-namespace eviction budgets. Missing entries fall back to
817    /// [`CacheBudget::default`]. Not persisted (per-handle policy).
818    cache_budgets: std::collections::HashMap<String, CacheBudget>,
819    /// Flush policy for cache mutations. Memory writes ignore this and
820    /// always behave as `Strict`.
821    cache_flush_policy: CacheFlushPolicy,
822    /// Bookkeeping for `CacheFlushPolicy::Batched` — pending mutation
823    /// count and first-dirty timestamp.
824    cache_batch: BatchState,
825    /// In-memory hit/miss/eviction counters (reset on reopen).
826    cache_counters: CounterState,
827    /// Optional handle-scoped embedder (Phase 3, PR 10a). Registered
828    /// via [`Mnemo::set_embedder`]; consulted by [`Mnemo::remember_text`]
829    /// and [`RecallRequest::from_text`]. `None` means text-side APIs
830    /// return [`MnemoError::Invalid`] with a "no embedder" message.
831    /// Not persisted — the handle is re-registered after every open.
832    embedder: Option<std::sync::Arc<dyn Embedder>>,
833    /// Fsyncs issued by the most recent `flush()` call. Snapshotted
834    /// from the thread-local fsync counter in `lib.rs::fsync_counter`:
835    /// `flush()` reads it on entry, does its work (each `sync_all` the
836    /// pager / wal / recovery path makes bumps the counter), and
837    /// stores the delta here. `0` on a fresh handle before any flush,
838    /// or on a read-only handle where `flush()` short-circuits.
839    /// Exposed through [`Stats::fsyncs_last_flush`] for diagnostics —
840    /// the Phase 6 planner wants this before deciding whether cutting
841    /// fsync count is a viable write-latency win.
842    fsyncs_last_flush: u32,
843    /// FIX-1 R-001.8 part 2: lazy cache of the distinct agent IDs
844    /// across every live memory, built once on first `stats()` call
845    /// per handle. `remember()` adds to it in O(1); `delete()` can't
846    /// cheaply tell whether the removed memory was the last of its
847    /// agent (the agent lives in the encrypted body, not the catalog
848    /// — pre-v9 format), so it just invalidates the set and the next
849    /// `stats()` call rebuilds. Pre-FIX-1 every `stats()` call
850    /// decrypted every live memory to collect agents, turning an
851    /// otherwise-cheap diagnostic into an O(N) decrypt.
852    agent_set: Option<std::collections::BTreeSet<String>>,
853    /// Phase 12.1 lazy caches, built together on the first keyed op this
854    /// handle sees (`remember_keyed`, `update` that touches `_mnemo.key`,
855    /// or any `history(...)` call). Both are maintained incrementally
856    /// after that. `None` → not built yet; `Some(…)` → authoritative.
857    ///
858    /// * `key_index`: `(scope_owner, key) → ULID bytes` of the **live**
859    ///   record for each keyed slot. `scope_owner` is `Some(agent_id)`
860    ///   for a `Private` scope slot and `None` for a `Shared` one, so a
861    ///   private write never collides with a shared one.
862    /// * `superseded`: the ULID bytes of every record that has been
863    ///   superseded (`_mnemo.superseded_by` set on it). `recall` and
864    ///   `about` skip these before any decrypt.
865    ///
866    /// Building both costs one scan over live records (~1s at 100K with
867    /// hardware AES) and happens at most once per handle. A format bump
868    /// to v9 moves the key into the catalog row so the scan goes away.
869    key_index: Option<std::collections::HashMap<KeySlot, u128>>,
870    superseded: Option<std::collections::HashSet<u128>>,
871    /// R-004.2: thread layers + onboarding ids, built by the same lazy scan.
872    layers: Option<LayerCache>,
873}
874
875/// Composite key for the lazy `key_index` on `Mnemo`. A slot is identified
876/// by `(scope_owner, key)`:
877/// * `Scope::Private` writes carry the owning agent → `Some(agent_id)`.
878/// * `Scope::Shared` writes are per-file → `None`.
879///
880/// A private write never supersedes a shared one, and the reverse is also
881/// true. Scope lives on `Memory.scope`, agent on `Memory.agent_id`.
882#[derive(Debug, Clone, PartialEq, Eq, Hash)]
883pub(crate) struct KeySlot {
884    /// `Some(agent_id)` for `Scope::Private`, `None` for `Scope::Shared`.
885    pub(crate) scope_owner: Option<String>,
886    /// The caller-supplied key string, exactly as given (no case-fold).
887    pub(crate) key: String,
888}
889
890/// Open a file and take an OS advisory lock, returning `Locked { path }`
891/// if the requested lock cannot be acquired. On drop of the returned
892/// [`File`], the OS releases the lock. `flock`/`LockFileEx` semantics
893/// (per file-descriptor on Unix, per-open-file-handle on Windows) mean
894/// the lock is scoped to this handle: cloning the file (or the returned
895/// `File`) shares the lock, but a fresh `open()` from any process gets a
896/// distinct handle and will collide.
897fn open_with_lock(path: &Path, write: bool, create: bool) -> Result<File> {
898    let mut opts = OpenOptions::new();
899    opts.read(true);
900    if write {
901        opts.write(true);
902    }
903    if create {
904        opts.create(true).truncate(true);
905    }
906    let file = opts.open(path)?;
907    // On success the returned `Result<()>` is `Ok(())`. On conflict:
908    //   * Unix `flock(2)` → `ErrorKind::WouldBlock` (EWOULDBLOCK / EAGAIN)
909    //   * Windows `LockFileEx` → raw OS code 33 (ERROR_LOCK_VIOLATION)
910    //     or 32 (ERROR_SHARING_VIOLATION), both surfaced by Rust as
911    //     `ErrorKind::Uncategorized` because `std::io::ErrorKind`
912    //     doesn't carry a portable "locked" variant.
913    // Anything else is a real I/O failure (permissions, invalid handle,
914    // ...) and gets bubbled up as `MnemoError::Io`.
915    let result = if write {
916        FileExt::try_lock_exclusive(&file)
917    } else {
918        FileExt::try_lock_shared(&file)
919    };
920    match result {
921        Ok(()) => Ok(file),
922        Err(e) if is_lock_conflict(&e) => Err(MnemoError::Locked {
923            path: path.to_path_buf(),
924        }),
925        Err(e) => Err(MnemoError::Io(e)),
926    }
927}
928
929/// Do two paths resolve to the same on-disk file? Used by
930/// [`Mnemo::export_encrypted`] to refuse overwriting the live
931/// database. On Unix we compare `(dev, ino)` via `MetadataExt`, which
932/// catches hard links and symlinks. On Windows we compare canonicalized
933/// paths, since there is no stable device+inode equivalent in `std`.
934/// Returns `None` if either side can't be stat'd / canonicalized —
935/// the caller treats that as "not the same file" so a missing `dest`
936/// (the normal export case) proceeds.
937fn paths_refer_to_same_file(a: &Path, b: &Path) -> Option<bool> {
938    #[cfg(unix)]
939    {
940        use std::os::unix::fs::MetadataExt;
941        let ma = std::fs::metadata(a).ok()?;
942        let mb = std::fs::metadata(b).ok()?;
943        Some(ma.dev() == mb.dev() && ma.ino() == mb.ino())
944    }
945    #[cfg(not(unix))]
946    {
947        let ca = std::fs::canonicalize(a).ok()?;
948        let cb = std::fs::canonicalize(b).ok()?;
949        Some(ca == cb)
950    }
951}
952
953/// A sibling temp-file path next to `dest`, suffixed with
954/// `.export-tmp`. Lives in `dest`'s own parent directory so the final
955/// [`std::fs::rename`] is same-filesystem (and therefore atomic on
956/// every platform mnemo supports).
957fn temp_sibling(dest: &Path) -> PathBuf {
958    let mut buf = dest.as_os_str().to_os_string();
959    buf.push(".export-tmp");
960    PathBuf::from(buf)
961}
962
963/// Was this I/O error the OS reporting a conflicting file lock, or a
964/// genuine I/O failure? Platform-conditional because
965/// [`std::io::ErrorKind`] carries no portable "resource-locked" variant
966/// as of Rust 1.75 — Windows surfaces `ERROR_LOCK_VIOLATION` /
967/// `ERROR_SHARING_VIOLATION` as `Uncategorized`.
968fn is_lock_conflict(e: &std::io::Error) -> bool {
969    if e.kind() == std::io::ErrorKind::WouldBlock {
970        return true;
971    }
972    #[cfg(windows)]
973    if let Some(code) = e.raw_os_error() {
974        // 33 = ERROR_LOCK_VIOLATION, 32 = ERROR_SHARING_VIOLATION —
975        // both mean "someone else holds an incompatible lock on this
976        // region of the file". See
977        // https://learn.microsoft.com/en-us/windows/win32/debug/system-error-codes--0-499-
978        if code == 33 || code == 32 {
979            return true;
980        }
981    }
982    false
983}
984
985/// Read a run of consecutive encrypted pages and concatenate their plaintext.
986fn read_run_bytes(pager: &mut Pager, start: u64, pages: u64, len: u64) -> Result<Vec<u8>> {
987    let mut buf = Vec::with_capacity(len as usize);
988    for i in 0..pages {
989        buf.extend_from_slice(&pager.read_page(start + i)?);
990    }
991    buf.truncate(len as usize);
992    Ok(buf)
993}
994
995/// Load a v5+ catalog run (an empty `pages` yields an empty catalog).
996fn load_catalog(pager: &mut Pager, start: u64, pages: u64, len: u64) -> Result<Vec<CatalogEntry>> {
997    if pages == 0 {
998        return Ok(Vec::new());
999    }
1000    let buf = read_run_bytes(pager, start, pages, len)?;
1001    rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
1002}
1003
1004/// Load a **v4** catalog run using the frozen [`CatalogEntryV4`] shape.
1005/// Used only by the v4→v5 migration path in [`Mnemo::open`].
1006fn load_catalog_v4(
1007    pager: &mut Pager,
1008    start: u64,
1009    pages: u64,
1010    len: u64,
1011) -> Result<Vec<CatalogEntryV4>> {
1012    if pages == 0 {
1013        return Ok(Vec::new());
1014    }
1015    let buf = read_run_bytes(pager, start, pages, len)?;
1016    rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
1017}
1018
1019/// Walk a v4 catalog and synthesize the v5 shape. For each live entry,
1020/// reads the memory's serialized body and copies its `accessed_at` and
1021/// `access_count` into the catalog row (the v4 record body still carries
1022/// them — they're moving *from* the body *to* the catalog). Deleted
1023/// entries skip the read and zero the fields.
1024fn migrate_v4_catalog(pager: &mut Pager, v4: Vec<CatalogEntryV4>) -> Result<Vec<CatalogEntry>> {
1025    let mut out = Vec::with_capacity(v4.len());
1026    for e in &v4 {
1027        let (accessed_at, access_count) = if e.deleted {
1028            (0i64, 0u32)
1029        } else {
1030            let mut buf = Vec::with_capacity(e.len as usize);
1031            for i in 0..e.page_count as u64 {
1032                buf.extend_from_slice(&pager.read_page(e.start_page + i)?);
1033            }
1034            buf.truncate(e.len as usize);
1035            let mem: Memory = rmp_serde::from_slice(&buf)
1036                .map_err(|err| MnemoError::Serialize(err.to_string()))?;
1037            (mem.accessed_at, mem.access_count)
1038        };
1039        out.push(CatalogEntry {
1040            id: e.id,
1041            start_page: e.start_page,
1042            page_count: e.page_count,
1043            len: e.len,
1044            deleted: e.deleted,
1045            accessed_at,
1046            access_count,
1047        });
1048    }
1049    Ok(out)
1050}
1051
1052/// Load the snapshot manifest (an empty `pages` yields an empty manifest).
1053fn load_manifest(pager: &mut Pager, start: u64, pages: u64, len: u64) -> Result<Vec<Snapshot>> {
1054    if pages == 0 {
1055        return Ok(Vec::new());
1056    }
1057    let buf = read_run_bytes(pager, start, pages, len)?;
1058    rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
1059}
1060
1061/// Load the result-cache directory (v8+). An empty `pages` yields an
1062/// empty directory, which is the natural state for v7 files migrated
1063/// to v8 (no cache entries yet) and for freshly created v8 files.
1064fn load_cache_directory(
1065    pager: &mut Pager,
1066    start: u64,
1067    pages: u64,
1068    len: u64,
1069) -> Result<Vec<CacheDirectoryEntry>> {
1070    if pages == 0 {
1071        return Ok(Vec::new());
1072    }
1073    let buf = read_run_bytes(pager, start, pages, len)?;
1074    rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))
1075}
1076
1077/// Load an ANN index run, validating its dimensionality.
1078fn load_index(
1079    pager: &mut Pager,
1080    start: u64,
1081    pages: u64,
1082    len: u64,
1083    dims: usize,
1084) -> Result<Option<IvfPqIndex>> {
1085    if pages == 0 {
1086        return Ok(None);
1087    }
1088    let buf = read_run_bytes(pager, start, pages, len)?;
1089    let mut idx: IvfPqIndex =
1090        rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
1091    if idx.dims() != dims {
1092        return Err(MnemoError::Invalid(
1093            "ANN index dimensionality does not match the database".into(),
1094        ));
1095    }
1096    idx.rebuild_assignment();
1097    Ok(Some(idx))
1098}
1099
1100/// True if a memory's metadata marks it as the canonical onboarding manifest
1101/// (`metadata.topic == "manifest"`). Used by [`Mnemo::about`] to hoist the
1102/// manifest to the top of the briefing regardless of importance ordering.
1103fn is_manifest(m: &Memory) -> bool {
1104    m.metadata
1105        .get("topic")
1106        .and_then(|v| v.as_str())
1107        .map(|s| s.eq_ignore_ascii_case("manifest"))
1108        .unwrap_or(false)
1109}
1110
1111/// PERF-1 step 4: ULID → catalog row, as an open-addressing table of `u32`
1112/// row numbers: 4 bytes a slot, about 5 bytes an entry, against about 43
1113/// for the `HashMap<u128, usize>` it replaces. A probe compares against the
1114/// catalog's own id at that row, so the ids aren't stored twice. Rows are
1115/// never removed (a delete only tombstones the catalog entry), so there's
1116/// no delete path.
1117#[derive(Debug, Clone, Default)]
1118struct IdIndex {
1119    slots: Vec<u32>,
1120    len: usize,
1121}
1122
1123const NO_ROW: u32 = u32::MAX;
1124
1125impl IdIndex {
1126    fn build(catalog: &[CatalogEntry]) -> IdIndex {
1127        let mut t = IdIndex {
1128            slots: vec![NO_ROW; crate::index::table_capacity_for(catalog.len())],
1129            len: 0,
1130        };
1131        for (row, e) in catalog.iter().enumerate() {
1132            t.insert(e.id, row, catalog);
1133        }
1134        t
1135    }
1136
1137    /// The catalog row holding `id`, if any.
1138    fn get(&self, id: u128, catalog: &[CatalogEntry]) -> Option<usize> {
1139        if self.slots.is_empty() {
1140            return None;
1141        }
1142        let cap = self.slots.len();
1143        let mut i = crate::index::slot_home(id, cap);
1144        loop {
1145            let row = self.slots[i];
1146            if row == NO_ROW {
1147                return None;
1148            }
1149            if catalog[row as usize].id == id {
1150                return Some(row as usize);
1151            }
1152            i = crate::index::next_slot(i, cap);
1153        }
1154    }
1155
1156    /// Map `id` to `row`. `catalog` must hold every row already in the
1157    /// table (the new row may not be pushed yet).
1158    fn insert(&mut self, id: u128, row: usize, catalog: &[CatalogEntry]) {
1159        if (self.len + 1) * 4 > self.slots.len() * 3 {
1160            let cap = (self.slots.len() * 2).max(crate::index::table_capacity_for(self.len + 1));
1161            let old = std::mem::replace(&mut self.slots, vec![NO_ROW; cap]);
1162            self.len = 0;
1163            for r in old.into_iter().filter(|&r| r != NO_ROW) {
1164                self.insert(catalog[r as usize].id, r as usize, catalog);
1165            }
1166        }
1167        let cap = self.slots.len();
1168        let mut i = crate::index::slot_home(id, cap);
1169        loop {
1170            let r = self.slots[i];
1171            if r == NO_ROW {
1172                self.slots[i] = row as u32;
1173                self.len += 1;
1174                return;
1175            }
1176            if catalog[r as usize].id == id {
1177                self.slots[i] = row as u32;
1178                return;
1179            }
1180            i = crate::index::next_slot(i, cap);
1181        }
1182    }
1183
1184    fn heap_bytes(&self) -> usize {
1185        self.slots.capacity() * std::mem::size_of::<u32>()
1186    }
1187}
1188
1189/// Build the ULID → catalog-row lookup.
1190fn build_id_index(catalog: &[CatalogEntry]) -> IdIndex {
1191    IdIndex::build(catalog)
1192}
1193
1194// ---------------------------------------------------------------------------
1195// Phase 12.7a helpers used by `ingest_thread`.
1196// ---------------------------------------------------------------------------
1197
1198/// One incoming turn after chunking or truncation, ready to embed and
1199/// write (Phase 12.7a, split out for 12.7d's batch embed).
1200struct PendingTurn {
1201    turn_id: String,
1202    role: String,
1203    created_at: i64,
1204    parts: Vec<String>,
1205    /// Hex sha256 + original length, set only for a truncated
1206    /// tool-result turn. Applied to the single chunk the truncation
1207    /// yielded.
1208    truncation: Option<(String, u64)>,
1209}
1210
1211/// Output of `Mnemo::plan_ingest`: the turns to write, the first free
1212/// `_mnemo.seq`, and the counts gathered so far (skips, chunking,
1213/// truncation).
1214struct IngestPlan {
1215    pending: Vec<PendingTurn>,
1216    next_seq: u64,
1217    report: IngestReport,
1218}
1219
1220impl IngestPlan {
1221    /// Every text to embed, in write order (turns, then parts).
1222    fn texts(&self) -> Vec<&str> {
1223        self.pending
1224            .iter()
1225            .flat_map(|p| p.parts.iter().map(String::as_str))
1226            .collect()
1227    }
1228
1229    fn text_count(&self) -> usize {
1230        self.pending.iter().map(|p| p.parts.len()).sum()
1231    }
1232}
1233
1234/// Whether a record belongs in the ANN index: facts only. Thread-archive
1235/// turns, essence items and thread links are read by kind and scope, never
1236/// by similarity, so they stay out (R-004.2).
1237fn is_ann_kind(m: &Memory) -> bool {
1238    !matches!(
1239        m.metadata.get(MNEMO_META_KIND).and_then(|v| v.as_str()),
1240        Some(MNEMO_META_KIND_TURN)
1241            | Some(MNEMO_META_KIND_ESSENCE)
1242            | Some(MNEMO_META_KIND_THREAD_LINK)
1243    )
1244}
1245
1246/// Hex-encode a byte slice. Low-dependency alternative to pulling in a
1247/// hex crate for the lone call site in `ingest_thread`'s SHA-256
1248/// recording.
1249fn hex_encode(bytes: &[u8]) -> String {
1250    const HEX: &[u8; 16] = b"0123456789abcdef";
1251    let mut out = String::with_capacity(bytes.len() * 2);
1252    for b in bytes {
1253        out.push(HEX[(b >> 4) as usize] as char);
1254        out.push(HEX[(b & 0x0f) as usize] as char);
1255    }
1256    out
1257}
1258
1259/// Return the longest UTF-8 prefix of `s` whose byte length is `<=
1260/// max_bytes`. Always lands on a `char` boundary.
1261fn take_utf8_prefix(s: &str, max_bytes: usize) -> String {
1262    if s.len() <= max_bytes {
1263        return s.to_string();
1264    }
1265    // Walk forward through char indices; the last index whose byte
1266    // offset fits inside `max_bytes` tells us where to cut.
1267    let mut cut = 0usize;
1268    for (idx, _) in s.char_indices() {
1269        if idx > max_bytes {
1270            break;
1271        }
1272        cut = idx;
1273    }
1274    s[..cut].to_string()
1275}
1276
1277/// Split `s` into chunks, each with `<= max_bytes` of UTF-8 content,
1278/// landing on `char` boundaries. Preserves original ordering; a trailing
1279/// chunk may be shorter than `max_bytes`.
1280/// Phase 12.7b helper: pull the `_mnemo.status` string from an
1281/// essence record. Returns `None` for timeline entries (they have no
1282/// status) and for malformed records.
1283fn status_of(m: &Memory) -> Option<&str> {
1284    m.metadata.get(MNEMO_META_STATUS).and_then(|v| v.as_str())
1285}
1286
1287/// Phase 12.7b helper: render one essence item as a markdown bullet.
1288/// Timeline entries get a date tag; section items get `[open]` /
1289/// `[resolved]` so the reader can scan active work quickly.
1290fn render_essence_item(section: EssenceSection, m: &Memory) -> String {
1291    if section == EssenceSection::Timeline {
1292        let date = format_unix_date(m.created_at);
1293        format!("- [{date}] {}\n", m.content)
1294    } else {
1295        let status = status_of(m).unwrap_or(essence_status::OPEN);
1296        format!("- [{status}] {}\n", m.content)
1297    }
1298}
1299
1300/// Minimal unix-seconds → `YYYY-MM-DD` formatter (no chrono/time
1301/// dependency; mirrors the ISO 8601 parser in `transcript.rs`).
1302fn format_unix_date(secs: i64) -> String {
1303    let days_total = secs.div_euclid(86_400);
1304    let (y, m, d) = civil_from_days(days_total);
1305    format!("{y:04}-{m:02}-{d:02}")
1306}
1307
1308/// Howard Hinnant's `civil_from_days` — inverse of `days_from_civil`,
1309/// adequate for Gregorian dates since 1970. Pure arithmetic, no
1310/// dependencies.
1311fn civil_from_days(z: i64) -> (i64, u32, u32) {
1312    let z = z + 719_468;
1313    let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
1314    let doe = (z - era * 146_097) as u64;
1315    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
1316    let y = yoe as i64 + era * 400;
1317    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
1318    let mp = (5 * doy + 2) / 153;
1319    let d = (doy - (153 * mp + 2) / 5 + 1) as u32;
1320    let m = if mp < 10 { mp + 3 } else { mp - 9 } as u32;
1321    let y = if m <= 2 { y + 1 } else { y };
1322    (y, m, d)
1323}
1324
1325fn chunk_utf8(s: &str, max_bytes: usize) -> Vec<String> {
1326    if s.len() <= max_bytes {
1327        return vec![s.to_string()];
1328    }
1329    let mut out = Vec::new();
1330    let mut rest = s;
1331    while rest.len() > max_bytes {
1332        let head = take_utf8_prefix(rest, max_bytes);
1333        let head_bytes = head.len();
1334        out.push(head);
1335        rest = &rest[head_bytes..];
1336    }
1337    if !rest.is_empty() {
1338        out.push(rest.to_string());
1339    }
1340    out
1341}
1342
1343impl Mnemo {
1344    /// Create a brand-new encrypted database at `path`.
1345    pub fn create(
1346        path: impl Into<PathBuf>,
1347        passphrase: &str,
1348        config: MnemoConfig,
1349    ) -> Result<Mnemo> {
1350        Self::create_secret(path, passphrase.as_bytes(), config)
1351    }
1352
1353    /// PERF-1 step 7 (R-004.5): create a database unlocked by a **raw
1354    /// 256-bit key** instead of a passphrase, for keys held by an OS
1355    /// keychain or keystore. The key must be uniformly random (see
1356    /// [`Mnemo::generate_key`]); it gets no stretching. The header records
1357    /// minimal Argon2id parameters ([`KdfParams::raw_key`]), so opening
1358    /// costs microseconds instead of tens of milliseconds. `config.kdf` is
1359    /// ignored. No format change: the key goes through the same KEK
1360    /// derivation a passphrase does.
1361    pub fn create_with_key(
1362        path: impl Into<PathBuf>,
1363        key: &[u8; crypto::KEY_LEN],
1364        config: MnemoConfig,
1365    ) -> Result<Mnemo> {
1366        let config = MnemoConfig {
1367            kdf: KdfParams::raw_key(),
1368            ..config
1369        };
1370        Self::create_secret(path, key, config)
1371    }
1372
1373    /// A fresh random 256-bit key for [`Mnemo::create_with_key`], from the
1374    /// thread-local CSPRNG (seeded from the OS). Store it in a keychain;
1375    /// it can't be recovered from the file.
1376    pub fn generate_key() -> [u8; crypto::KEY_LEN] {
1377        *crypto::random_dek()
1378    }
1379
1380    fn create_secret(
1381        path: impl Into<PathBuf>,
1382        secret: &[u8],
1383        config: MnemoConfig,
1384    ) -> Result<Mnemo> {
1385        let path: PathBuf = path.into();
1386        if config.dimensions == 0 {
1387            return Err(MnemoError::Invalid("dimensions must be > 0".into()));
1388        }
1389        // Exclusive lock covers the write path; second writer sees `Locked`.
1390        let file = open_with_lock(&path, /*write=*/ true, /*create=*/ true)?;
1391
1392        let salt = crypto::random_salt();
1393        let dek = crypto::random_dek();
1394        let kek = crypto::derive_kek(secret, &salt, config.kdf)?;
1395        let dek_nonce = crypto::random_nonce();
1396        let wrapped = crypto::wrap_dek(&kek, &dek_nonce, &dek)?;
1397        if wrapped.len() != WRAPPED_DEK_LEN {
1398            return Err(MnemoError::Crypto("unexpected wrapped DEK length".into()));
1399        }
1400        let mut wrapped_dek = [0u8; WRAPPED_DEK_LEN];
1401        wrapped_dek.copy_from_slice(&wrapped);
1402
1403        let mut header = Header {
1404            version: VERSION,
1405            page_size: PAGE_SIZE as u32,
1406            flags: FLAG_ENCRYPTED,
1407            dimensions: config.dimensions as u32,
1408            created_at: memory::now_secs(),
1409            write_counter: 0,
1410            // Page 0 is the header; pages 1..=wal_pages are the WAL;
1411            // record/catalog/index pages start after it. Initial WAL size is
1412            // chosen per-config (default 8 pages = 64 KiB) and clamped to a
1413            // sane floor so the first transaction can always commit.
1414            next_page: 1 + config.wal_pages_initial.max(MIN_WAL_PAGES),
1415            catalog_start: 0,
1416            catalog_pages: 0,
1417            catalog_len: 0,
1418            vector_count: 0,
1419            m_cost: config.kdf.m_cost,
1420            t_cost: config.kdf.t_cost,
1421            p_cost: config.kdf.p_cost,
1422            salt,
1423            dek_nonce,
1424            wrapped_dek,
1425            index_start: 0,
1426            index_pages: 0,
1427            index_len: 0,
1428            wal_start: 1,
1429            wal_pages: config.wal_pages_initial.max(MIN_WAL_PAGES),
1430            wal_seq: 0,
1431            manifest_start: 0,
1432            manifest_pages: 0,
1433            manifest_len: 0,
1434            // Regenerated per write by `apply_seal`; `None` until we seal.
1435            seal_nonce: None,
1436            seal_tag: None,
1437            // v8+ result-cache pointer — zero-initialized on create,
1438            // populated only when `cache_put` runs at least once and
1439            // the next `flush` seals a directory run.
1440            cache_start: 0,
1441            cache_pages: 0,
1442            cache_len: 0,
1443        };
1444
1445        let mut pager = Pager::new(file, dek, 0, VERSION);
1446        let mut page = header.to_page();
1447        header.apply_seal(&mut page, pager.dek())?;
1448        pager.write_raw(0, &page)?;
1449        pager.sync()?;
1450
1451        Ok(Mnemo {
1452            pager,
1453            header,
1454            catalog: Vec::new(),
1455            index: IdIndex::default(),
1456            path,
1457            dimensions: config.dimensions,
1458            kdf: config.kdf,
1459            dirty_catalog: false,
1460            ann: None,
1461            dirty_index: false,
1462            index_persist: IndexPersist::default(),
1463            ann_unloaded: false,
1464            manifest: Vec::new(),
1465            max_snapshots: config.max_snapshots,
1466            read_only: false,
1467            cache_dir: Vec::new(),
1468            cache_idx: DirectoryIndex::new(),
1469            dirty_cache: false,
1470            cache_budgets: std::collections::HashMap::new(),
1471            cache_flush_policy: CacheFlushPolicy::default(),
1472            cache_batch: BatchState::default(),
1473            cache_counters: CounterState::default(),
1474            embedder: None,
1475            fsyncs_last_flush: 0,
1476            agent_set: None,
1477            key_index: None,
1478            superseded: None,
1479            layers: None,
1480        })
1481    }
1482
1483    /// Open an existing database. A wrong passphrase fails cleanly with
1484    /// [`MnemoError::WrongPassphrase`]. Takes an OS exclusive lock for
1485    /// the lifetime of the returned handle; a second concurrent open
1486    /// (from any process) returns [`MnemoError::Locked`].
1487    ///
1488    /// For a caller that only needs to read (and can tolerate the
1489    /// pending-recovery / pending-migration cases documented on
1490    /// [`Mnemo::open_read_only`]), prefer that entry point — it takes a
1491    /// shared lock and coexists with other readers.
1492    pub fn open(path: impl Into<PathBuf>, passphrase: &str) -> Result<Mnemo> {
1493        Self::open_secret(path, passphrase.as_bytes())
1494    }
1495
1496    /// PERF-1 step 7: open a database created by [`Mnemo::create_with_key`]
1497    /// (or rekeyed with [`Mnemo::rekey_with_key`]) with its raw key.
1498    pub fn open_with_key(path: impl Into<PathBuf>, key: &[u8; crypto::KEY_LEN]) -> Result<Mnemo> {
1499        Self::open_secret(path, key)
1500    }
1501
1502    fn open_secret(path: impl Into<PathBuf>, secret: &[u8]) -> Result<Mnemo> {
1503        let path: PathBuf = path.into();
1504        let mut file = open_with_lock(&path, /*write=*/ true, /*create=*/ false)?;
1505
1506        let mut hbuf = [0u8; PAGE_SIZE];
1507        file.seek(SeekFrom::Start(0))?;
1508        file.read_exact(&mut hbuf)?;
1509        let mut header = match Header::from_page(&hbuf) {
1510            Ok(h) => h,
1511            Err(e) => {
1512                // DUR-1 (R-002.1 part 2+3): page 0 did not parse — any
1513                // error is treated as a torn-header candidate, not just
1514                // `HeaderChecksum`. The whole 294-byte header lives in
1515                // the first 512-byte sector, so the realistic failure is
1516                // the sector zeroed or garbled, which returns `BadMagic`
1517                // or `UnsupportedVersion` *before* the CRC gate.
1518                //
1519                // Scan the file for valid committed header-frame WAL
1520                // regions, pick the highest `txn_id`, and replay it. If
1521                // nothing valid is found, return the *original* error so
1522                // non-mnemo files still report `BadMagic`.
1523                let scan = scan_header_recovery_candidates(&mut file, secret)?;
1524                let frames = scan.ok_or(e)?;
1525                for (page_no, bytes) in &frames {
1526                    if bytes.len() != PAGE_SIZE {
1527                        return Err(MnemoError::Invalid("WAL frame is not page-sized".into()));
1528                    }
1529                    file.seek(SeekFrom::Start(page_no * PAGE_SIZE as u64))?;
1530                    file.write_all(bytes)?;
1531                }
1532                file.sync_all()?;
1533                crate::bump_fsync();
1534                file.seek(SeekFrom::Start(0))?;
1535                file.read_exact(&mut hbuf)?;
1536                Header::from_page(&hbuf)?
1537            }
1538        };
1539
1540        // Crash recovery: replay a committed-but-uncheckpointed transaction.
1541        // Each frame is a finished page image; the header frame, replayed to
1542        // page 0, supersedes the header just read.
1543        if let Some(frames) = wal::recover(
1544            &mut file,
1545            header.wal_start,
1546            header.wal_pages,
1547            header.wal_seq,
1548        )? {
1549            for (page_no, bytes) in &frames {
1550                if bytes.len() != PAGE_SIZE {
1551                    return Err(MnemoError::Invalid("WAL frame is not page-sized".into()));
1552                }
1553                file.seek(SeekFrom::Start(page_no * PAGE_SIZE as u64))?;
1554                file.write_all(bytes)?;
1555            }
1556            file.sync_all()?;
1557            crate::bump_fsync();
1558            // Re-read the now-current header from the replayed page 0.
1559            file.seek(SeekFrom::Start(0))?;
1560            file.read_exact(&mut hbuf)?;
1561            header = Header::from_page(&hbuf)?;
1562        }
1563
1564        let kdf = KdfParams {
1565            m_cost: header.m_cost,
1566            t_cost: header.t_cost,
1567            p_cost: header.p_cost,
1568        };
1569        let kek = crypto::derive_kek(secret, &header.salt, kdf)?;
1570        let dek = crypto::unwrap_dek(&kek, &header.dek_nonce, &header.wrapped_dek)?;
1571
1572        // Validate the v7 header seal now that we have the DEK. Returns
1573        // `MnemoError::HeaderTampered` if any of the sealed mutable fields
1574        // were rewritten after the last legitimate flush. No-op for v6 and
1575        // older — pre-v7 files don't carry a seal.
1576        //
1577        // DUR-1 (Planner decision 2026-11-01, Option B): a seal failure is
1578        // **always** reported as tampering, never auto-healed. R-002.1
1579        // part 3's seal-mismatch WAL fallback is withdrawn: the seal
1580        // exists for tamper-evidence; silently loading a stale committed
1581        // header from the WAL would erase that evidence. A real crash
1582        // cannot produce "CRC valid + seal invalid" either, because the
1583        // whole 294-byte header lies inside sector 0 which storage
1584        // writes atomically — a torn sector breaks magic, version or
1585        // CRC first, and *those* are healed by the scan above.
1586        header.validate_seal(&dek)?;
1587
1588        // Build the pager at the file's *current* version so initial reads
1589        // use the matching AAD scheme — `Pager::page_aad` returns empty AAD
1590        // for v4/v5 (no page binding) and `page_no.to_le_bytes()` for v6+.
1591        let on_disk_version = header.version;
1592        let mut pager = Pager::new(file, dek, header.write_counter, on_disk_version);
1593        let dimensions = header.dimensions as usize;
1594
1595        // === v4→v5 catalog migration =====================================
1596        // v6 reads directly; v5 reads directly; v4 replays the catalog into
1597        // the v5 shape by reading each memory's body to populate the new
1598        // `accessed_at` / `access_count` fields. All page reads here still
1599        // use the v4/v5 AAD scheme (none).
1600        let (catalog, migrated_from_v4) = if on_disk_version >= 5 {
1601            (
1602                load_catalog(
1603                    &mut pager,
1604                    header.catalog_start,
1605                    header.catalog_pages,
1606                    header.catalog_len,
1607                )?,
1608                false,
1609            )
1610        } else if on_disk_version == 4 {
1611            let v4 = load_catalog_v4(
1612                &mut pager,
1613                header.catalog_start,
1614                header.catalog_pages,
1615                header.catalog_len,
1616            )?;
1617            (migrate_v4_catalog(&mut pager, v4)?, true)
1618        } else {
1619            return Err(MnemoError::UnsupportedVersion(on_disk_version));
1620        };
1621        let index = build_id_index(&catalog);
1622        // PERF-1 step 3: the ANN index loads on first use, not at open. A
1623        // v5→v6 page migration is the exception: it must read the index
1624        // now, under the old page AAD, to rewrite it.
1625        let ann = if on_disk_version < 6 {
1626            load_index(
1627                &mut pager,
1628                header.index_start,
1629                header.index_pages,
1630                header.index_len,
1631                dimensions,
1632            )?
1633        } else {
1634            None
1635        };
1636        let ann_unloaded = ann.is_none() && header.index_pages > 0;
1637        let mut manifest = load_manifest(
1638            &mut pager,
1639            header.manifest_start,
1640            header.manifest_pages,
1641            header.manifest_len,
1642        )?;
1643
1644        // === v5→v6 page-encryption migration =============================
1645        // Pre-v6 data pages were encrypted under no AAD; v6 binds the home
1646        // `page_no` as AAD so a page-transplant attack fails authentication
1647        // on read. Re-stage every live record page through `write_page` so
1648        // the next flush re-encrypts it under v6 AAD at the same home slot.
1649        // Tombstoned entries skip — they're reclaimed by `compact_file`.
1650        // Old catalog / ANN / manifest pages stay as orphan ciphertext;
1651        // the next flush writes the new runs at fresh slots.
1652        let migrated_pages_to_v6 = on_disk_version < 6;
1653        if migrated_pages_to_v6 {
1654            for e in &catalog {
1655                if e.deleted {
1656                    continue;
1657                }
1658                for i in 0..e.page_count as u64 {
1659                    let payload = pager.read_page(e.start_page + i)?;
1660                    pager.write_page(e.start_page + i, &payload)?;
1661                }
1662            }
1663            pager.set_version(6);
1664        }
1665
1666        // === v6→v7 header-seal migration =================================
1667        // v7 adds an AES-GCM seal at the tail of the header page that
1668        // authenticates every mutable header field. Pre-v7 files have no
1669        // seal — we just bump the version in memory and mark `dirty_catalog`
1670        // so the next flush writes a sealed v7 header. Pages are untouched
1671        // (page crypto is unchanged from v6).
1672        let migrated_header_to_v7 = on_disk_version < 7;
1673        if migrated_header_to_v7 {
1674            pager.set_version(7);
1675        }
1676
1677        // === v7→v8 cache-directory migration =============================
1678        // v8 adds three u64 header fields (cache_start/pages/len) for the
1679        // result-cache directory (Phase 10.1). Pre-v8 files have those
1680        // three bytes as zero (the on-disk region beyond the v7 seal is
1681        // zero-initialized), which is the legal "empty cache directory"
1682        // state — no data migration required. We bump the version in
1683        // memory and mark dirty_catalog so the next flush writes a
1684        // v8-sealed header with the new AAD shape covering the cache
1685        // pointer. Pages are untouched.
1686        let migrated_header_to_v8 = on_disk_version < 8;
1687        if migrated_header_to_v8 {
1688            pager.set_version(VERSION);
1689        }
1690
1691        let migrated = migrated_from_v4
1692            || migrated_pages_to_v6
1693            || migrated_header_to_v7
1694            || migrated_header_to_v8;
1695        if migrated {
1696            // Stamp the version forward in memory; the next flush persists
1697            // it via the WAL-committed header frame (and the v7 seal).
1698            header.version = VERSION;
1699            // Snapshots written under an older format reference page runs
1700            // encrypted under the old AAD scheme; this build can't decrypt
1701            // them under the new pager. Drop the manifest so the next
1702            // flush records a fresh snapshot of the migrated state.
1703            // Point-in-time recovery into the pre-migration past is
1704            // sacrificed for migration simplicity. Old pages remain on
1705            // disk until `compact_file`. (For v6→v7 alone — which doesn't
1706            // change page crypto — the old snapshots would technically
1707            // still be readable, but dropping them uniformly keeps the
1708            // migration policy simple.)
1709            manifest.clear();
1710        }
1711
1712        // The ANN index pages on disk were sealed under the *old* AAD scheme
1713        // if we migrated pages to v6; force a rewrite at fresh v6-encrypted
1714        // slots on the next flush so a future reopen doesn't try to read
1715        // them under the new AAD and fail. (Catalog and manifest already
1716        // get rewritten via dirty_catalog and the manifest.clear() above.)
1717        let dirty_index = migrated_pages_to_v6 && ann.is_some();
1718
1719        // Load the cache directory (v8+ only). Pre-v8 files have
1720        // cache_pages == 0 by header layout, so this trivially yields
1721        // an empty directory without touching the disk.
1722        let cache_dir = load_cache_directory(
1723            &mut pager,
1724            header.cache_start,
1725            header.cache_pages,
1726            header.cache_len,
1727        )?;
1728        let cache_idx = DirectoryIndex::build(&cache_dir);
1729
1730        Ok(Mnemo {
1731            pager,
1732            header,
1733            catalog,
1734            index,
1735            path,
1736            dimensions,
1737            kdf,
1738            dirty_catalog: migrated,
1739            ann,
1740            dirty_index,
1741            // A v5→v6 page migration leaves the index sealed under the old
1742            // page AAD, so it must be rewritten on the next flush.
1743            index_persist: IndexPersist {
1744                forced: dirty_index,
1745                ..IndexPersist::default()
1746            },
1747            ann_unloaded,
1748            manifest,
1749            // `Mnemo::open` has no [`MnemoConfig`] to read; default the cap
1750            // to [`DEFAULT_MAX_SNAPSHOTS`]. Override with
1751            // [`Mnemo::set_max_snapshots`] right after open if you need
1752            // unlimited retention or a different cap for this handle.
1753            max_snapshots: DEFAULT_MAX_SNAPSHOTS,
1754            read_only: false,
1755            cache_dir,
1756            cache_idx,
1757            dirty_cache: false,
1758            cache_budgets: std::collections::HashMap::new(),
1759            cache_flush_policy: CacheFlushPolicy::default(),
1760            cache_batch: BatchState::default(),
1761            cache_counters: CounterState::default(),
1762            embedder: None,
1763            fsyncs_last_flush: 0,
1764            agent_set: None,
1765            key_index: None,
1766            superseded: None,
1767            layers: None,
1768        })
1769    }
1770
1771    /// Open an existing database in **read-only** mode. Takes an OS shared
1772    /// lock; multiple read-only handles can coexist, but any attempt to
1773    /// open the same file read-write elsewhere returns
1774    /// [`MnemoError::Locked`]. All mutating methods on the returned handle
1775    /// return [`MnemoError::ReadOnly`], including `recall` with
1776    /// `track_access(true)`.
1777    ///
1778    /// Two conditions block a read-only open and return
1779    /// [`MnemoError::NeedsWriteOpen`]:
1780    ///
1781    /// 1. The file has a committed-but-uncheckpointed WAL transaction that
1782    ///    would ordinarily be replayed on open. A read-only handle
1783    ///    cannot write the recovered pages back.
1784    /// 2. The on-disk format version is older than the current
1785    ///    `format::VERSION`, requiring an on-disk migration.
1786    ///
1787    /// In either case, open the file read-write once (which recovers /
1788    /// migrates in place), close the handle cleanly, and reopen
1789    /// read-only. This is documented in the README's Durability section.
1790    pub fn open_read_only(path: impl Into<PathBuf>, passphrase: &str) -> Result<Mnemo> {
1791        Self::open_read_only_secret(path, passphrase.as_bytes())
1792    }
1793
1794    /// PERF-1 step 7: [`Mnemo::open_read_only`] for a raw-key database.
1795    pub fn open_read_only_with_key(
1796        path: impl Into<PathBuf>,
1797        key: &[u8; crypto::KEY_LEN],
1798    ) -> Result<Mnemo> {
1799        Self::open_read_only_secret(path, key)
1800    }
1801
1802    fn open_read_only_secret(path: impl Into<PathBuf>, secret: &[u8]) -> Result<Mnemo> {
1803        let path: PathBuf = path.into();
1804        let mut file = open_with_lock(&path, /*write=*/ false, /*create=*/ false)?;
1805
1806        let mut hbuf = [0u8; PAGE_SIZE];
1807        file.seek(SeekFrom::Start(0))?;
1808        file.read_exact(&mut hbuf)?;
1809        // A torn header on a read-only open cannot be healed (that requires
1810        // writing recovered frames back). Surface as `NeedsWriteOpen` so the
1811        // caller knows to reopen read-write once. DUR-1 (R-002.1 part 4):
1812        // any `Header::from_page` failure (`BadMagic`,
1813        // `UnsupportedVersion`, `HeaderChecksum`) qualifies — but only
1814        // when the file-wide scan actually finds a valid committed
1815        // header frame to recover from, so a non-mnemo file still
1816        // reports `BadMagic` cleanly. A `validate_seal` failure is NOT
1817        // routed here (Option B: tampering is never auto-healed).
1818        let header = match Header::from_page(&hbuf) {
1819            Ok(h) => h,
1820            Err(parse_err) => {
1821                let recoverable = scan_header_recovery_candidates(&mut file, secret)?;
1822                return Err(match (parse_err, recoverable) {
1823                    (_, Some(_)) => MnemoError::NeedsWriteOpen {
1824                        reason: "torn header page needs recovery from WAL",
1825                    },
1826                    (e, None) => e,
1827                });
1828            }
1829        };
1830
1831        // Refuse to migrate on read-only opens. v6→v7 is header-only
1832        // (no page writes required), but even that bumps `version` on
1833        // disk and re-seals the header — which needs write access. Keep
1834        // the policy uniform across all sub-version bumps.
1835        if header.version < VERSION {
1836            return Err(MnemoError::NeedsWriteOpen {
1837                reason: "pending format-version migration",
1838            });
1839        }
1840
1841        // Refuse if there is a committed-but-uncheckpointed transaction
1842        // waiting for `wal::recover` to fold it into home pages.
1843        if wal::recover(
1844            &mut file,
1845            header.wal_start,
1846            header.wal_pages,
1847            header.wal_seq,
1848        )?
1849        .is_some()
1850        {
1851            return Err(MnemoError::NeedsWriteOpen {
1852                reason: "pending WAL recovery",
1853            });
1854        }
1855
1856        let kdf = KdfParams {
1857            m_cost: header.m_cost,
1858            t_cost: header.t_cost,
1859            p_cost: header.p_cost,
1860        };
1861        let kek = crypto::derive_kek(secret, &header.salt, kdf)?;
1862        let dek = crypto::unwrap_dek(&kek, &header.dek_nonce, &header.wrapped_dek)?;
1863        header.validate_seal(&dek)?;
1864
1865        let mut pager = Pager::new(file, dek, header.write_counter, header.version);
1866        let dimensions = header.dimensions as usize;
1867
1868        let catalog = load_catalog(
1869            &mut pager,
1870            header.catalog_start,
1871            header.catalog_pages,
1872            header.catalog_len,
1873        )?;
1874        let index = build_id_index(&catalog);
1875        // PERF-1 step 3: the ANN index loads on first use, not at open.
1876        let ann: Option<IvfPqIndex> = None;
1877        let ann_unloaded = header.index_pages > 0;
1878        let manifest = load_manifest(
1879            &mut pager,
1880            header.manifest_start,
1881            header.manifest_pages,
1882            header.manifest_len,
1883        )?;
1884        let cache_dir = load_cache_directory(
1885            &mut pager,
1886            header.cache_start,
1887            header.cache_pages,
1888            header.cache_len,
1889        )?;
1890        let cache_idx = DirectoryIndex::build(&cache_dir);
1891
1892        Ok(Mnemo {
1893            pager,
1894            header,
1895            catalog,
1896            index,
1897            path,
1898            dimensions,
1899            kdf,
1900            // No mutation is possible; leave both dirty flags false so
1901            // `close()` is a genuine no-op even if we later add an
1902            // auto-flush on drop.
1903            dirty_catalog: false,
1904            ann,
1905            dirty_index: false,
1906            index_persist: IndexPersist::default(),
1907            ann_unloaded,
1908            manifest,
1909            max_snapshots: DEFAULT_MAX_SNAPSHOTS,
1910            read_only: true,
1911            cache_dir,
1912            cache_idx,
1913            dirty_cache: false,
1914            cache_budgets: std::collections::HashMap::new(),
1915            cache_flush_policy: CacheFlushPolicy::default(),
1916            cache_batch: BatchState::default(),
1917            cache_counters: CounterState::default(),
1918            embedder: None,
1919            fsyncs_last_flush: 0,
1920            agent_set: None,
1921            key_index: None,
1922            superseded: None,
1923            layers: None,
1924        })
1925    }
1926
1927    /// Override the manifest snapshot cap on this open handle. `0` disables
1928    /// the cap; any positive value keeps the most-recent `max` snapshots and
1929    /// drops the rest on the next flush. See [`MnemoConfig::max_snapshots`].
1930    pub fn set_max_snapshots(&mut self, max: usize) {
1931        self.max_snapshots = max;
1932    }
1933
1934    /// Embedding dimensionality this database expects.
1935    pub fn dimensions(&self) -> usize {
1936        self.dimensions
1937    }
1938
1939    /// Number of live (non-deleted) memories.
1940    pub fn len(&self) -> usize {
1941        self.catalog.iter().filter(|e| !e.deleted).count()
1942    }
1943
1944    /// True if the database holds no live memories.
1945    pub fn is_empty(&self) -> bool {
1946        self.len() == 0
1947    }
1948
1949    // --- internal page/record helpers -------------------------------------
1950
1951    fn write_record(&mut self, bytes: &[u8]) -> Result<(u64, u32)> {
1952        let pc = bytes.len().div_ceil(PAYLOAD).max(1);
1953        let start = self.header.next_page;
1954        self.header.next_page += pc as u64;
1955        for i in 0..pc {
1956            let lo = i * PAYLOAD;
1957            let hi = ((i + 1) * PAYLOAD).min(bytes.len());
1958            self.pager.write_page(start + i as u64, &bytes[lo..hi])?;
1959        }
1960        Ok((start, pc as u32))
1961    }
1962
1963    fn read_record(&mut self, e: &CatalogEntry) -> Result<Vec<u8>> {
1964        let mut buf = Vec::with_capacity(e.len as usize);
1965        for i in 0..e.page_count as u64 {
1966            buf.extend_from_slice(&self.pager.read_page(e.start_page + i)?);
1967        }
1968        buf.truncate(e.len as usize);
1969        Ok(buf)
1970    }
1971
1972    fn read_memory(&mut self, e: &CatalogEntry) -> Result<Memory> {
1973        let bytes = self.read_record(e)?;
1974        let mut m: Memory =
1975            rmp_serde::from_slice(&bytes).map_err(|err| MnemoError::Serialize(err.to_string()))?;
1976        // The catalog is the source of truth for access stats — the values
1977        // in the record body are a stale snapshot from when the record was
1978        // last written. Overwrite them here so consumers always see live
1979        // values regardless of when the record was last serialized.
1980        m.accessed_at = e.accessed_at;
1981        m.access_count = e.access_count;
1982        Ok(m)
1983    }
1984
1985    /// Serialize and store a memory (insert or overwrite by ID).
1986    fn put(&mut self, mut m: Memory) -> Result<Ulid> {
1987        if m.vector.len() != self.dimensions {
1988            return Err(MnemoError::DimensionMismatch {
1989                expected: self.dimensions,
1990                got: m.vector.len(),
1991            });
1992        }
1993        // Reject all-zero vectors (Phase 3 PR 10a; breaking vs 0.4.x).
1994        // A zero vector scores exactly 0.0 on every cosine query — it
1995        // can't contribute signal, but recency / importance can still
1996        // pull it into top-k as noise. We'd rather fail loudly at
1997        // `remember` than let a mis-embedded entry pollute recall
1998        // quality for the life of the database. Scaffold manifests now
1999        // use a sentinel unit vector so this check doesn't trip them.
2000        if embed::is_zero_vector(&m.vector) {
2001            return Err(MnemoError::Invalid(
2002                "all-zero vector refused: it scores 0.0 on every cosine query and \
2003                 pollutes recall; check your embedder returned a real embedding"
2004                    .into(),
2005            ));
2006        }
2007        if m.id == Ulid::nil() {
2008            m.id = Ulid::new();
2009        }
2010        let id_u: u128 = m.id.0;
2011        // R-004.2: an overwrite with the handle caches built reads the
2012        // stored version first (one decrypt), so the caches drop exactly
2013        // what it contributed.
2014        if self.layers.is_some() {
2015            if let Some(idx) = self.index.get(id_u, &self.catalog) {
2016                let old_entry = self.catalog[idx].clone();
2017                if !old_entry.deleted {
2018                    let old = self.read_memory(&old_entry)?;
2019                    self.note_removed(&old);
2020                }
2021            }
2022        }
2023        let bytes = rmp_serde::to_vec(&m).map_err(|e| MnemoError::Serialize(e.to_string()))?;
2024        let (start, pc) = self.write_record(&bytes)?;
2025        // For overwrites preserve the existing access stats; for inserts
2026        // seed them from the memory (which carries them in its body).
2027        let (prev_accessed, prev_count) = match self.index.get(id_u, &self.catalog) {
2028            Some(idx) => (
2029                self.catalog[idx].accessed_at,
2030                self.catalog[idx].access_count,
2031            ),
2032            None => (m.accessed_at, m.access_count),
2033        };
2034        let entry = CatalogEntry {
2035            id: id_u,
2036            start_page: start,
2037            page_count: pc,
2038            len: bytes.len() as u32,
2039            deleted: false,
2040            accessed_at: prev_accessed,
2041            access_count: prev_count,
2042        };
2043        match self.index.get(id_u, &self.catalog) {
2044            Some(idx) => self.catalog[idx] = entry,
2045            None => {
2046                self.index.insert(id_u, self.catalog.len(), &self.catalog);
2047                self.catalog.push(entry);
2048            }
2049        }
2050        self.dirty_catalog = true;
2051
2052        // Keep the ANN index complete: assign the (new or changed) vector to
2053        // its nearest partition. Centroids/codebook stay fixed until rebuild.
2054        // R-004.2: only facts go in; turns, essence items and thread links
2055        // are read by kind and scope.
2056        if let Some(ann) = &mut self.ann {
2057            if is_ann_kind(&m) {
2058                ann.add(id_u, &m.vector);
2059                self.dirty_index = true;
2060                self.index_persist.changes += 1;
2061            } else if ann.contains(id_u) {
2062                ann.remove(id_u);
2063                self.dirty_index = true;
2064                self.index_persist.changes += 1;
2065            }
2066        }
2067        self.note_added(&m);
2068
2069        // FIX-1 R-001.8 part 2: keep the lazy agent set in sync. If
2070        // it hasn't been built yet (no `stats()` call this session),
2071        // skip — the first `stats()` will build it from scratch. If
2072        // it has, add this memory's agent_id. An overwrite with the
2073        // same id and same agent is a no-op (BTreeSet::insert returns
2074        // false); an overwrite with a *changed* agent_id adds the new
2075        // agent but doesn't drop the old (we can't tell without a
2076        // decrypt whether the old agent still has live memories).
2077        // That's conservative in the right direction — `stats()` may
2078        // over-report by one agent until the next delete invalidates.
2079        if let Some(set) = &mut self.agent_set {
2080            set.insert(m.agent_id.clone());
2081        }
2082        Ok(m.id)
2083    }
2084
2085    // --- public CRUD ------------------------------------------------------
2086
2087    /// Register a text embedder for this handle. Enables
2088    /// [`Mnemo::remember_text`] and the text-side `RecallRequest`
2089    /// convenience path. Does two checks before storing the embedder:
2090    ///
2091    /// 1. **Dimension check.** The embedder's `identity().dimensions`
2092    ///    must equal [`Mnemo::dimensions`] (the width this database
2093    ///    was created with). Mismatch returns
2094    ///    [`MnemoError::DimensionMismatch`].
2095    /// 2. **Manifest identity check.** If the scaffold / onboarding
2096    ///    manifest already records an embedder — the canonical shape
2097    ///    is `metadata.embedder = { name, dimensions, normalize }`,
2098    ///    with the legacy flat `embedder_name` / `embedder_dimensions`
2099    ///    / `embedder_normalize` keys accepted as an alias — the new
2100    ///    embedder's identity must match exactly. A different name —
2101    ///    even at the same dimensions — returns
2102    ///    [`MnemoError::Invalid`] with
2103    ///    [`EmbedderIdentity::MISMATCH_MSG`] in the message. If the
2104    ///    manifest has no embedder metadata yet, this call upserts
2105    ///    the nested object into the first onboarding/manifest memory
2106    ///    (write + flush so a subsequent reopen sees the identity).
2107    ///
2108    /// **Read-only handles.** `set_embedder` on an `open_read_only`
2109    /// handle performs both checks and adopts the embedder in memory;
2110    /// it never writes. Readers (CLI `--text`, agent-side `recall_text`)
2111    /// can therefore use the text path next to a writer that holds
2112    /// the exclusive lock.
2113    ///
2114    /// On both failure paths the handle's embedder slot is left
2115    /// unchanged. On success the embedder is wrapped in an `Arc<dyn
2116    /// Embedder>` and consulted by every text-side call made through
2117    /// this handle. The slot is per-handle and not persisted — a
2118    /// fresh open must re-register (same identity, so the manifest
2119    /// check round-trips cleanly).
2120    pub fn set_embedder<E: Embedder + 'static>(&mut self, embedder: E) -> Result<()> {
2121        let identity = embedder.identity();
2122        if identity.dimensions != self.dimensions {
2123            return Err(MnemoError::DimensionMismatch {
2124                expected: self.dimensions,
2125                got: identity.dimensions,
2126            });
2127        }
2128
2129        // Find a scaffold/onboarding manifest to inspect (or upsert
2130        // into). `about()` already filters + sorts the onboarding set.
2131        let onboarding = self.about()?;
2132        let mut upsert_target: Option<usize> = None;
2133        for (i, m) in onboarding.iter().enumerate() {
2134            if let Some(saw) = embed::read_identity_from_metadata(&m.metadata) {
2135                if saw != identity {
2136                    return Err(embed::identity_mismatch_err(&saw, &identity));
2137                }
2138                // Match — nothing to persist; adopt + done. Works the
2139                // same on read-only and read-write handles.
2140                self.embedder = Some(std::sync::Arc::new(embedder));
2141                return Ok(());
2142            }
2143            if upsert_target.is_none() {
2144                upsert_target = Some(i);
2145            }
2146        }
2147
2148        // No existing identity on the manifest. On a read-write handle
2149        // upsert into the first onboarding memory so later opens see
2150        // it; on a read-only handle skip the write (readers must not
2151        // mutate the file), but still adopt the embedder in memory.
2152        if !self.read_only {
2153            if let Some(idx) = upsert_target {
2154                let mut m = onboarding[idx].clone();
2155                embed::write_identity_into_metadata(&mut m.metadata, &identity);
2156                // Reuse the memory's existing id so this is an in-place
2157                // update of the scaffold manifest, not a new entry.
2158                self.put(m)?;
2159                self.flush()?;
2160            }
2161        }
2162
2163        self.embedder = Some(std::sync::Arc::new(embedder));
2164        Ok(())
2165    }
2166
2167    /// Borrow the currently-registered embedder, if any. Returns
2168    /// `None` when no embedder has been set via
2169    /// [`Mnemo::set_embedder`]. Bindings use this to drive their own
2170    /// text-side convenience methods without duplicating the identity
2171    /// check.
2172    pub fn embedder(&self) -> Option<&dyn Embedder> {
2173        self.embedder.as_deref()
2174    }
2175
2176    /// Shared handle to the registered embedder, cloned so the caller
2177    /// can drop the borrow on this `Mnemo` before invoking `embed()`.
2178    /// FIX-1 / PY-1 (R-001.11): the Python binding needs this so an
2179    /// embedder callback that re-enters `Mnemo` (e.g. a Python
2180    /// embedder that calls `db.dimensions()` or `db.cache_get(...)`)
2181    /// doesn't hit PyO3's "Already borrowed" — we clone the Arc under
2182    /// a short `&self` borrow, release that borrow, then invoke
2183    /// `embed()` with no `&mut Mnemo` held.
2184    pub fn embedder_arc(&self) -> Option<std::sync::Arc<dyn Embedder>> {
2185        self.embedder.clone()
2186    }
2187
2188    /// Embed `content` with the registered embedder and `remember` the
2189    /// result. Returns the new memory's ULID. Convenience path for
2190    /// callers who have a string, want it stored, and don't need any
2191    /// of the per-memory options [`Memory`] exposes (agent, importance,
2192    /// metadata, ttl, scope) — build a [`Memory`] by hand and call
2193    /// [`Mnemo::remember`] for those.
2194    ///
2195    /// Fails with [`MnemoError::Invalid`] if no embedder has been
2196    /// registered on this handle (call [`Mnemo::set_embedder`] first);
2197    /// with [`MnemoError::DimensionMismatch`] if the embedder returned
2198    /// the wrong width for this database.
2199    pub fn remember_text(&mut self, content: &str, memory_type: MemoryType) -> Result<Ulid> {
2200        if self.read_only {
2201            return Err(MnemoError::ReadOnly);
2202        }
2203        let embedder = self.embedder.as_ref().ok_or_else(|| {
2204            MnemoError::Invalid(
2205                "no embedder registered on this handle; call set_embedder first".into(),
2206            )
2207        })?;
2208        let mut vectors = embedder.embed(&[content])?;
2209        let vector = vectors.pop().ok_or_else(|| {
2210            MnemoError::Invalid("embedder returned no vector for the input text".into())
2211        })?;
2212        if vector.len() != self.dimensions {
2213            return Err(MnemoError::DimensionMismatch {
2214                expected: self.dimensions,
2215                got: vector.len(),
2216            });
2217        }
2218        let memory = Memory::new(content.to_string(), memory_type, vector);
2219        self.remember(memory)
2220    }
2221
2222    /// Store a memory. Returns its ULID. If the memory's ID is nil a fresh
2223    /// one is assigned; an existing ID overwrites in place.
2224    pub fn remember(&mut self, memory: Memory) -> Result<Ulid> {
2225        if self.read_only {
2226            return Err(MnemoError::ReadOnly);
2227        }
2228        self.put(memory)
2229    }
2230
2231    /// Fetch a memory by ID.
2232    pub fn get(&mut self, id: &Ulid) -> Result<Memory> {
2233        let idx = self
2234            .index
2235            .get(id.0, &self.catalog)
2236            .ok_or_else(|| MnemoError::NotFound(id.to_string()))?;
2237        let entry = self.catalog[idx].clone();
2238        if entry.deleted {
2239            return Err(MnemoError::NotFound(id.to_string()));
2240        }
2241        self.read_memory(&entry)
2242    }
2243
2244    /// Soft-delete a memory (tombstoned; space reclaimed by `compact`).
2245    pub fn delete(&mut self, id: &Ulid) -> Result<()> {
2246        if self.read_only {
2247            return Err(MnemoError::ReadOnly);
2248        }
2249        let idx = self
2250            .index
2251            .get(id.0, &self.catalog)
2252            .ok_or_else(|| MnemoError::NotFound(id.to_string()))?;
2253        // R-004.2 / 12.1: with the handle caches built, read the record
2254        // being deleted (one decrypt) so the key slot, superseded set and
2255        // thread layers drop exactly what it contributed.
2256        let entry = self.catalog[idx].clone();
2257        if self.layers.is_some() && !entry.deleted {
2258            let old = self.read_memory(&entry)?;
2259            self.note_removed(&old);
2260        }
2261        if !self.catalog[idx].deleted {
2262            self.catalog[idx].deleted = true;
2263            self.dirty_catalog = true;
2264        }
2265        if let Some(ann) = &mut self.ann {
2266            if ann.contains(id.0) {
2267                ann.remove(id.0);
2268                self.dirty_index = true;
2269                self.index_persist.changes += 1;
2270            }
2271        }
2272        // FIX-1 R-001.8 part 2: we can't tell whether this delete
2273        // removed the last memory of its agent without a decrypt
2274        // (the catalog entry doesn't carry `agent_id` in format v8),
2275        // so just invalidate the lazy set. The next `stats()` call
2276        // rebuilds from live memories.
2277        self.agent_set = None;
2278        // Phase 12.1: delete of a keyed record removes it from the slot,
2279        // and delete of a superseded record frees a `history` entry. When
2280        // the caches are built, `note_removed` above did that exactly;
2281        // otherwise they're built from scratch on first use anyway.
2282        Ok(())
2283    }
2284
2285    /// Keep the handle caches (12.1 key slots, superseded set, R-004.2
2286    /// layers) current after `put` writes `m`. No-op until they're built.
2287    fn note_added(&mut self, m: &Memory) {
2288        let id = m.id.0;
2289        if let (Some(keys), Some(sup)) = (self.key_index.as_mut(), self.superseded.as_mut()) {
2290            if m.metadata.contains_key(MNEMO_META_SUPERSEDED_BY) {
2291                sup.insert(id);
2292            } else if let Some(Value::String(k)) = m.metadata.get(MNEMO_META_KEY) {
2293                keys.insert(Self::key_slot_for(m, k), id);
2294            }
2295        }
2296        if let Some(layers) = self.layers.as_mut() {
2297            layers.add(m);
2298        }
2299    }
2300
2301    /// Undo [`Mnemo::note_added`] for the stored version of a record that's
2302    /// about to be overwritten or deleted.
2303    fn note_removed(&mut self, m: &Memory) {
2304        let id = m.id.0;
2305        if let (Some(keys), Some(sup)) = (self.key_index.as_mut(), self.superseded.as_mut()) {
2306            sup.remove(&id);
2307            if let Some(Value::String(k)) = m.metadata.get(MNEMO_META_KEY) {
2308                let slot = Self::key_slot_for(m, k);
2309                if keys.get(&slot) == Some(&id) {
2310                    keys.remove(&slot);
2311                }
2312            }
2313        }
2314        if let Some(layers) = self.layers.as_mut() {
2315            layers.remove(m);
2316        }
2317    }
2318
2319    /// The live catalog entries for `ids` (ids that are gone are skipped).
2320    fn live_entries(&self, ids: impl IntoIterator<Item = u128>) -> Vec<CatalogEntry> {
2321        ids.into_iter()
2322            .filter_map(|id| {
2323                self.index
2324                    .get(id, &self.catalog)
2325                    .map(|i| self.catalog[i].clone())
2326            })
2327            .filter(|e| !e.deleted)
2328            .collect()
2329    }
2330
2331    /// Catalog entries of the essence items, from the handle caches.
2332    fn essence_entries(&mut self) -> Result<Vec<CatalogEntry>> {
2333        self.ensure_key_caches()?;
2334        let ids: Vec<u128> = self
2335            .layers
2336            .as_ref()
2337            .map(|l| l.essence.iter().copied().collect())
2338            .unwrap_or_default();
2339        Ok(self.live_entries(ids))
2340    }
2341
2342    /// Catalog entries of every record filed under `thread`, from the
2343    /// handle caches.
2344    fn thread_entries(&mut self, thread: &str) -> Result<Vec<CatalogEntry>> {
2345        self.ensure_key_caches()?;
2346        let ids: Vec<u128> = self
2347            .layers
2348            .as_ref()
2349            .and_then(|l| l.threads.get(thread))
2350            .map(|tc| tc.records.iter().copied().collect())
2351            .unwrap_or_default();
2352        Ok(self.live_entries(ids))
2353    }
2354
2355    // --- Phase 12.1: edit in place + keyed supersession -----------------
2356
2357    /// Scope-aware key-slot identifier for `memory` and a caller key.
2358    ///
2359    /// * [`Scope::Private`] → owned by the memory's `agent_id`.
2360    /// * [`Scope::Shared`]  → one slot across the whole file.
2361    fn key_slot_for(memory: &Memory, key: &str) -> crate::store::KeySlot {
2362        let scope_owner = match memory.scope {
2363            Scope::Private => Some(memory.agent_id.clone()),
2364            Scope::Shared => None,
2365        };
2366        crate::store::KeySlot {
2367            scope_owner,
2368            key: key.to_string(),
2369        }
2370    }
2371
2372    /// The handle's one lazy scan: builds the 12.1 `key_index` +
2373    /// `superseded` caches and the R-004.2 layer cache (essence ids,
2374    /// onboarding ids, threads, links) from live records, and prunes
2375    /// non-fact ids from a loaded ANN index (files written before v0.10.0
2376    /// indexed every kind). Decrypts every live record once per handle;
2377    /// `put` and `delete` keep the result current after that.
2378    fn ensure_key_caches(&mut self) -> Result<()> {
2379        if self.key_index.is_some() && self.superseded.is_some() && self.layers.is_some() {
2380            return Ok(());
2381        }
2382        let entries: Vec<CatalogEntry> = self
2383            .catalog
2384            .iter()
2385            .filter(|e| !e.deleted)
2386            .cloned()
2387            .collect();
2388        let mut keys: std::collections::HashMap<crate::store::KeySlot, u128> =
2389            std::collections::HashMap::new();
2390        let mut superseded: std::collections::HashSet<u128> = std::collections::HashSet::new();
2391        let mut layers = LayerCache::default();
2392        let mut non_facts: Vec<u128> = Vec::new();
2393        for e in &entries {
2394            let m = self.read_memory(e)?;
2395            layers.add(&m);
2396            if !is_ann_kind(&m) {
2397                non_facts.push(e.id);
2398            }
2399            if m.metadata.get(MNEMO_META_SUPERSEDED_BY).is_some() {
2400                superseded.insert(e.id);
2401                continue;
2402            }
2403            if let Some(Value::String(k)) = m.metadata.get(MNEMO_META_KEY) {
2404                let slot = Self::key_slot_for(&m, k);
2405                // Catalog order is insertion order; a later live record
2406                // with the same slot would only exist if a prior build
2407                // of this code allowed it. Keep the later id (newer
2408                // ULID), in case of a tie.
2409                keys.entry(slot)
2410                    .and_modify(|cur| {
2411                        if *cur < e.id {
2412                            *cur = e.id;
2413                        }
2414                    })
2415                    .or_insert(e.id);
2416            }
2417        }
2418        self.key_index = Some(keys);
2419        self.superseded = Some(superseded);
2420        self.layers = Some(layers);
2421        if let Some(ann) = &mut self.ann {
2422            let mut pruned = 0usize;
2423            for id in non_facts {
2424                if ann.contains(id) {
2425                    ann.remove(id);
2426                    pruned += 1;
2427                }
2428            }
2429            if pruned > 0 {
2430                self.dirty_index = true;
2431                self.index_persist.changes += pruned;
2432            }
2433        }
2434        Ok(())
2435    }
2436
2437    /// Stage-identical check for a 12.1 keyed no-op. The incoming write
2438    /// is byte-equal to the live record iff every user-visible field
2439    /// (content, type, importance, vector, user metadata) matches. The
2440    /// `_mnemo.*` reserved keys on the live record are **ignored** for
2441    /// this comparison — a repeat of an identical write re-issued by an
2442    /// extractor must be a NOOP even if the live record already carries
2443    /// supersession bookkeeping from a prior op.
2444    fn is_identical_keyed_write(incoming: &Memory, live: &Memory) -> bool {
2445        if incoming.content != live.content {
2446            return false;
2447        }
2448        if incoming.memory_type != live.memory_type {
2449            return false;
2450        }
2451        if (incoming.importance - live.importance).abs() > f32::EPSILON {
2452            return false;
2453        }
2454        if incoming.vector != live.vector {
2455            return false;
2456        }
2457        // Strip `_mnemo.*` from both sides before comparing user metadata.
2458        let strip = |m: &Map<String, Value>| -> Map<String, Value> {
2459            m.iter()
2460                .filter(|(k, _)| !k.starts_with(crate::memory::MNEMO_META_NS))
2461                .map(|(k, v)| (k.clone(), v.clone()))
2462                .collect()
2463        };
2464        strip(&incoming.metadata) == strip(&live.metadata)
2465    }
2466
2467    /// Store a memory under a keyed upsert slot. Returns the live record's
2468    /// id and a [`WriteOutcome`] saying whether this write created a new
2469    /// slot (`Added`), replaced a prior record (`Superseded`), or matched
2470    /// the existing live record byte-for-byte (`Unchanged`, no file write).
2471    ///
2472    /// The slot is identified by `(scope_owner, key)`:
2473    /// * `Scope::Private` memories share a slot per `(agent_id, key)`.
2474    /// * `Scope::Shared` memories share one slot per `key` across the file.
2475    ///
2476    /// On supersession the prior record is **not** deleted — it remains
2477    /// retrievable through [`Mnemo::history`] — but it is excluded from
2478    /// default [`Mnemo::recall`] and [`Mnemo::about`]. A caller that wants
2479    /// the full lineage of a key walks `history(key)`.
2480    ///
2481    /// This is the atomic 12.1 "supersede-or-noop" write path. Plain
2482    /// [`Mnemo::remember`] without a key is unchanged.
2483    pub fn remember_keyed(
2484        &mut self,
2485        mut memory: Memory,
2486        key: &str,
2487    ) -> Result<(Ulid, WriteOutcome)> {
2488        if self.read_only {
2489            return Err(MnemoError::ReadOnly);
2490        }
2491        if key.is_empty() {
2492            return Err(MnemoError::Invalid(
2493                "remember_keyed: 'key' must not be empty".into(),
2494            ));
2495        }
2496        self.ensure_key_caches()?;
2497        let slot = Self::key_slot_for(&memory, key);
2498        let existing = self.key_index.as_ref().and_then(|k| k.get(&slot).copied());
2499
2500        // Caller may already have set `_mnemo.key` on the memory; rewrite
2501        // it to our canonical value regardless so the record is
2502        // self-identifying.
2503        memory
2504            .metadata
2505            .insert(MNEMO_META_KEY.into(), Value::String(key.to_string()));
2506
2507        // NOOP check: if a live record exists for this slot and its
2508        // user-visible fields match, return the existing id without a
2509        // write.
2510        if let Some(prev_id_u) = existing {
2511            let prev_idx = self
2512                .index
2513                .get(prev_id_u, &self.catalog)
2514                .ok_or_else(|| MnemoError::Invalid("key_index drift".into()))?;
2515            let prev_entry = self.catalog[prev_idx].clone();
2516            let prev_mem = self.read_memory(&prev_entry)?;
2517            if Self::is_identical_keyed_write(&memory, &prev_mem) {
2518                return Ok((Ulid(prev_id_u), WriteOutcome::Unchanged));
2519            }
2520
2521            // SUPERSESSION: insert the new record, then rewrite the old
2522            // with `_mnemo.superseded_by` + `_mnemo.invalid_at`.
2523            let now = memory::now_secs();
2524            memory.id = Ulid::nil(); // force a fresh ULID in `put`
2525            memory.metadata.insert(
2526                MNEMO_META_SUPERSEDES.into(),
2527                Value::String(Ulid(prev_id_u).to_string()),
2528            );
2529            let new_id = self.put(memory)?;
2530            // Mark the prior record.
2531            let mut prev_mut = prev_mem;
2532            prev_mut.metadata.insert(
2533                MNEMO_META_SUPERSEDED_BY.into(),
2534                Value::String(new_id.to_string()),
2535            );
2536            prev_mut
2537                .metadata
2538                .insert(MNEMO_META_INVALID_AT.into(), Value::Number(now.into()));
2539            // keep prev_mut.id so `put` overwrites in place.
2540            self.put(prev_mut)?;
2541            // Update caches incrementally.
2542            if let Some(k) = self.key_index.as_mut() {
2543                k.insert(slot, new_id.0);
2544            }
2545            if let Some(s) = self.superseded.as_mut() {
2546                s.insert(prev_id_u);
2547            }
2548            return Ok((new_id, WriteOutcome::Superseded));
2549        }
2550
2551        // ADDED: no prior record for this slot.
2552        memory.id = Ulid::nil();
2553        let new_id = self.put(memory)?;
2554        if let Some(k) = self.key_index.as_mut() {
2555            k.insert(slot, new_id.0);
2556        }
2557        Ok((new_id, WriteOutcome::Added))
2558    }
2559
2560    /// Text-side keyed write: embed `content` through the registered
2561    /// embedder, then [`Mnemo::remember_keyed`]. Same returns.
2562    pub fn remember_keyed_text(
2563        &mut self,
2564        content: &str,
2565        memory_type: MemoryType,
2566        key: &str,
2567    ) -> Result<(Ulid, WriteOutcome)> {
2568        if self.read_only {
2569            return Err(MnemoError::ReadOnly);
2570        }
2571        let embedder = self.embedder.as_ref().ok_or_else(|| {
2572            MnemoError::Invalid(
2573                "no embedder registered on this handle; call set_embedder first".into(),
2574            )
2575        })?;
2576        let mut vectors = embedder.embed(&[content])?;
2577        let vector = vectors.pop().ok_or_else(|| {
2578            MnemoError::Invalid("embedder returned no vector for the input text".into())
2579        })?;
2580        if vector.len() != self.dimensions {
2581            return Err(MnemoError::DimensionMismatch {
2582                expected: self.dimensions,
2583                got: vector.len(),
2584            });
2585        }
2586        let memory = Memory::new(content.to_string(), memory_type, vector);
2587        self.remember_keyed(memory, key)
2588    }
2589
2590    /// Update an existing memory in place, preserving its ULID,
2591    /// `created_at` and access stats. Returns [`WriteOutcome::Updated`]
2592    /// on a real change, [`WriteOutcome::Unchanged`] when every field in
2593    /// the patch already matches the live record.
2594    ///
2595    /// Reserved `_mnemo.*` metadata keys on the existing record are
2596    /// preserved across a `metadata` replacement; a `metadata` of
2597    /// `Some(map)` replaces every **user-visible** key with `map`'s
2598    /// entries. A `content` change requires a matching `vector`
2599    /// (use [`Mnemo::update_text`] to re-embed) — a bare `content` patch
2600    /// is rejected to stop text and embedding drifting apart silently.
2601    pub fn update(&mut self, id: &Ulid, patch: MemoryPatch) -> Result<WriteOutcome> {
2602        if self.read_only {
2603            return Err(MnemoError::ReadOnly);
2604        }
2605        if patch.content.is_some() && patch.vector.is_none() {
2606            return Err(MnemoError::Invalid(
2607                "update: a content change needs a matching vector (or call update_text \
2608                 to re-embed through the registered embedder)"
2609                    .into(),
2610            ));
2611        }
2612        let idx = self
2613            .index
2614            .get(id.0, &self.catalog)
2615            .ok_or_else(|| MnemoError::NotFound(id.to_string()))?;
2616        if self.catalog[idx].deleted {
2617            return Err(MnemoError::NotFound(id.to_string()));
2618        }
2619        let entry = self.catalog[idx].clone();
2620        let existing = self.read_memory(&entry)?;
2621
2622        // Apply patch into a candidate, detect no-op after the fact.
2623        let mut cand = existing.clone();
2624        if let Some(c) = &patch.content {
2625            cand.content = c.clone();
2626        }
2627        if let Some(v) = &patch.vector {
2628            if v.len() != self.dimensions {
2629                return Err(MnemoError::DimensionMismatch {
2630                    expected: self.dimensions,
2631                    got: v.len(),
2632                });
2633            }
2634            cand.vector = v.clone();
2635        }
2636        if let Some(i) = patch.importance {
2637            cand.importance = i.clamp(0.0, 1.0);
2638        }
2639        if let Some(mt) = patch.memory_type {
2640            cand.memory_type = mt;
2641        }
2642        if let Some(new_meta) = patch.metadata {
2643            // Keep reserved `_mnemo.*` keys from the existing record.
2644            let reserved: Vec<(String, Value)> = existing
2645                .metadata
2646                .iter()
2647                .filter(|(k, _)| k.starts_with(crate::memory::MNEMO_META_NS))
2648                .map(|(k, v)| (k.clone(), v.clone()))
2649                .collect();
2650            cand.metadata = new_meta;
2651            for (k, v) in reserved {
2652                cand.metadata.insert(k, v);
2653            }
2654        }
2655        if let Some(ttl) = patch.ttl_secs {
2656            cand.ttl_secs = ttl;
2657        }
2658
2659        // No-op detection: compare all user-visible fields.
2660        if cand.content == existing.content
2661            && cand.vector == existing.vector
2662            && (cand.importance - existing.importance).abs() <= f32::EPSILON
2663            && cand.memory_type == existing.memory_type
2664            && cand.metadata == existing.metadata
2665            && cand.ttl_secs == existing.ttl_secs
2666        {
2667            return Ok(WriteOutcome::Unchanged);
2668        }
2669
2670        // Stamp `_mnemo.updated_at` and preserve id + created_at.
2671        let now = memory::now_secs();
2672        cand.metadata
2673            .insert(MNEMO_META_UPDATED_AT.into(), Value::Number(now.into()));
2674        cand.id = existing.id;
2675        cand.created_at = existing.created_at;
2676        // put() preserves accessed_at/access_count via the index lookup.
2677        self.put(cand)?;
2678        // `put` kept the key slot and the other handle caches current.
2679        Ok(WriteOutcome::Updated)
2680    }
2681
2682    /// In-place update that re-embeds `new_text` through the registered
2683    /// embedder. Shorthand for `update(id, patch.with_content_and_vector)`.
2684    pub fn update_text(
2685        &mut self,
2686        id: &Ulid,
2687        new_text: &str,
2688        mut patch: MemoryPatch,
2689    ) -> Result<WriteOutcome> {
2690        if self.read_only {
2691            return Err(MnemoError::ReadOnly);
2692        }
2693        let embedder = self.embedder.as_ref().ok_or_else(|| {
2694            MnemoError::Invalid(
2695                "no embedder registered on this handle; call set_embedder first".into(),
2696            )
2697        })?;
2698        let mut vectors = embedder.embed(&[new_text])?;
2699        let vector = vectors.pop().ok_or_else(|| {
2700            MnemoError::Invalid("embedder returned no vector for the input text".into())
2701        })?;
2702        if vector.len() != self.dimensions {
2703            return Err(MnemoError::DimensionMismatch {
2704                expected: self.dimensions,
2705                got: vector.len(),
2706            });
2707        }
2708        patch.content = Some(new_text.to_string());
2709        patch.vector = Some(vector);
2710        self.update(id, patch)
2711    }
2712
2713    /// Walk the full chain for a keyed slot: every record this key has
2714    /// ever held, newest first. Entry 0 is the **live** record (unless
2715    /// the slot is empty, in which case the vector is empty); subsequent
2716    /// entries are the superseded history in reverse-supersession order.
2717    ///
2718    /// **Slot resolution (v0.8.1 bugfix, lane A).** The lookup slot
2719    /// matches a keyed `remember` with the same arguments, so the
2720    /// obvious `history(key, None)` walks the private slot of the
2721    /// default agent — the same slot
2722    /// `remember_keyed(Memory::new(...), key)` writes to when the
2723    /// caller omits an explicit `agent_id` / `scope`. Pre-v0.8.1
2724    /// `agent_id = None` incorrectly looked up the shared slot and
2725    /// returned an empty chain for the common case; the old behavior
2726    /// is still reachable via `history_in(key, KeyScope::Shared)`.
2727    ///
2728    /// * `agent_id = Some(a)` → private slot for that agent.
2729    /// * `agent_id = None` → private slot for the default agent (same
2730    ///   as [`KeyScope::private_default`]).
2731    ///
2732    /// Use [`Mnemo::history_in`] with [`KeyScope::Shared`] for the
2733    /// shared slot.
2734    pub fn history(&mut self, key: &str, agent_id: Option<&str>) -> Result<Vec<Memory>> {
2735        let scope = match agent_id {
2736            Some(a) => KeyScope::Private(a.to_string()),
2737            None => KeyScope::private_default(),
2738        };
2739        self.history_in(key, scope)
2740    }
2741
2742    /// Walk the full chain for a keyed slot identified by an explicit
2743    /// [`KeyScope`]. Returns the same shape as [`Mnemo::history`] —
2744    /// live record first, then the superseded chain newest-first.
2745    ///
2746    /// Use this when you need the **shared** slot (`KeyScope::Shared`),
2747    /// which `history(key, None)` no longer reaches as of v0.8.1.
2748    pub fn history_in(&mut self, key: &str, scope: KeyScope) -> Result<Vec<Memory>> {
2749        self.ensure_key_caches()?;
2750        let slot = crate::store::KeySlot {
2751            scope_owner: match scope {
2752                KeyScope::Private(a) => Some(a),
2753                KeyScope::Shared => None,
2754            },
2755            key: key.to_string(),
2756        };
2757        let live_id = self.key_index.as_ref().and_then(|k| k.get(&slot).copied());
2758        let mut out = Vec::new();
2759        let mut cursor = live_id;
2760        while let Some(id_u) = cursor {
2761            let idx = self.index.get(id_u, &self.catalog).ok_or_else(|| {
2762                MnemoError::Invalid("history: superseded chain references missing id".into())
2763            })?;
2764            let entry = self.catalog[idx].clone();
2765            if entry.deleted {
2766                break;
2767            }
2768            let m = self.read_memory(&entry)?;
2769            let next = m
2770                .metadata
2771                .get(MNEMO_META_SUPERSEDES)
2772                .and_then(|v| v.as_str())
2773                .and_then(|s| Ulid::from_string(s).ok())
2774                .map(|u| u.0);
2775            out.push(m);
2776            cursor = next;
2777        }
2778        Ok(out)
2779    }
2780
2781    // --- Phase 12.7a: thread archive --------------------------------
2782
2783    /// Ingest a batch of turns into the archive layer of `thread_id`.
2784    ///
2785    /// Each [`ThreadTurn`] becomes one or more `Episodic` records
2786    /// tagged with the usual thread-archive reserved-metadata set
2787    /// (`_mnemo.kind = "turn"`, `_mnemo.thread`, `_mnemo.turn_id`,
2788    /// `_mnemo.seq`, `_mnemo.role`). The ingest path is
2789    /// **idempotent** per `turn_id`: a turn whose `turn_id` already
2790    /// appears on a record in this thread is skipped, so re-running
2791    /// `ingest_thread` with an extended transcript appends only
2792    /// what's new.
2793    ///
2794    /// **Chunking.** A non-tool turn whose UTF-8 content exceeds
2795    /// [`THREAD_TURN_CHUNK_BYTES`] (8 KB) is split into chunks at
2796    /// char boundaries, each stored as its own record with
2797    /// `_mnemo.part = 0, 1, …` under the same `turn_id`.
2798    ///
2799    /// **Truncation.** A `role = "tool"` turn whose content exceeds
2800    /// [`THREAD_TOOL_RESULT_TRUNCATE_BYTES`] (16 KB) is stored
2801    /// truncated — the first 16 KB of UTF-8 goes into the record,
2802    /// and `_mnemo.sha256` + `_mnemo.length` capture the original
2803    /// hash and byte length so the full content can be recovered
2804    /// (or verified) out-of-band.
2805    ///
2806    /// **Embedding.** An embedder must be registered
2807    /// ([`Mnemo::set_embedder`]); the content strings (one per
2808    /// resulting chunk / truncated record) are batched into a single
2809    /// `embed()` call.
2810    ///
2811    /// Call `self.flush()` after a successful ingest to persist.
2812    pub fn ingest_thread(&mut self, thread_id: &str, turns: &[ThreadTurn]) -> Result<IngestReport> {
2813        if self.read_only {
2814            return Err(MnemoError::ReadOnly);
2815        }
2816        let plan = self.plan_ingest(thread_id, turns)?;
2817        if plan.pending.is_empty() {
2818            return Ok(plan.report);
2819        }
2820        // Batched embed: one round trip to the embedder for the whole
2821        // ingest, whatever the turn count.
2822        let vectors = {
2823            let embedder = self.embedder.as_ref().ok_or_else(|| {
2824                MnemoError::Invalid(
2825                    "ingest_thread: no embedder registered on this handle; call set_embedder first"
2826                        .into(),
2827                )
2828            })?;
2829            embedder.embed(&plan.texts())?
2830        };
2831        self.check_vectors("ingest_thread", &vectors, plan.text_count())?;
2832        self.apply_ingest(thread_id, plan, vectors, None)
2833    }
2834
2835    /// Check a batch of embedder output before anything is written:
2836    /// one vector per input, the file's width, finite and non-zero.
2837    fn check_vectors(&self, ctx: &str, vectors: &[Vec<f32>], expected: usize) -> Result<()> {
2838        if vectors.len() != expected {
2839            return Err(MnemoError::Invalid(format!(
2840                "{ctx}: embedder returned {} vectors for {expected} texts",
2841                vectors.len()
2842            )));
2843        }
2844        for v in vectors {
2845            self.check_vector(ctx, v)?;
2846        }
2847        Ok(())
2848    }
2849
2850    /// The checks `put` applies to one vector, run up front so a batch
2851    /// can't fail half-way through its writes.
2852    fn check_vector(&self, ctx: &str, v: &[f32]) -> Result<()> {
2853        if v.len() != self.dimensions {
2854            return Err(MnemoError::DimensionMismatch {
2855                expected: self.dimensions,
2856                got: v.len(),
2857            });
2858        }
2859        if embed::is_zero_vector(v) {
2860            return Err(MnemoError::Invalid(format!(
2861                "{ctx}: all-zero vector refused"
2862            )));
2863        }
2864        if v.iter().any(|x| !x.is_finite()) {
2865            return Err(MnemoError::Invalid(format!(
2866                "{ctx}: vector contains a non-finite value"
2867            )));
2868        }
2869        Ok(())
2870    }
2871
2872    /// Planning half of [`Mnemo::ingest_thread`]: validate the turns, skip
2873    /// the ones already archived, and chunk or truncate the rest. Reads
2874    /// only; nothing is written.
2875    fn plan_ingest(&mut self, thread_id: &str, turns: &[ThreadTurn]) -> Result<IngestPlan> {
2876        if thread_id.is_empty() {
2877            return Err(MnemoError::Invalid(
2878                "ingest_thread: 'thread_id' must not be empty".into(),
2879            ));
2880        }
2881        let mut report = IngestReport::default();
2882        if turns.is_empty() {
2883            return Ok(IngestPlan {
2884                pending: Vec::new(),
2885                next_seq: 0,
2886                report,
2887            });
2888        }
2889
2890        // Build the set of turn_ids already in this thread so the
2891        // idempotence check is O(1) per incoming turn. First-pass
2892        // cost: one decrypt per existing thread record. A future v9
2893        // format bump can move `_mnemo.turn_id` into the catalog row
2894        // and drop this scan.
2895        // R-004.2: the turn ids already archived and the next `_mnemo.seq`
2896        // come from the handle caches: one lazy scan per handle, then O(1)
2897        // per ingest instead of two full decrypt scans each time.
2898        self.ensure_key_caches()?;
2899        let cached = self.layers.as_ref().and_then(|l| l.threads.get(thread_id));
2900        let existing_turn_ids: std::collections::HashSet<&str> = cached
2901            .map(|tc| tc.turns.keys().map(String::as_str).collect())
2902            .unwrap_or_default();
2903        // A repeat ingest continues the sequence instead of restarting at 0.
2904        let next_seq = cached.and_then(|tc| tc.max_seq).map_or(0, |n| n + 1);
2905
2906        // Determine each incoming turn's final text pieces (post-
2907        // chunking / truncation) before any writes, so the caller can
2908        // embed the whole batch in one `embed()` call.
2909        let mut pending: Vec<PendingTurn> = Vec::new();
2910        let now = memory::now_secs();
2911        for t in turns {
2912            if t.turn_id.is_empty() {
2913                return Err(MnemoError::Invalid(
2914                    "ingest_thread: turn_id must not be empty".into(),
2915                ));
2916            }
2917            if existing_turn_ids.contains(t.turn_id.as_str()) {
2918                report.skipped_existing_turns += 1;
2919                continue;
2920            }
2921            let content_bytes = t.content.len();
2922            let is_tool = t.role == "tool";
2923            let (parts, truncation) =
2924                if is_tool && content_bytes > THREAD_TOOL_RESULT_TRUNCATE_BYTES {
2925                    report.truncated_turns += 1;
2926                    let hash = {
2927                        use sha2::{Digest, Sha256};
2928                        let h = Sha256::digest(t.content.as_bytes());
2929                        hex_encode(&h)
2930                    };
2931                    let truncated = take_utf8_prefix(&t.content, THREAD_TOOL_RESULT_TRUNCATE_BYTES);
2932                    (vec![truncated], Some((hash, content_bytes as u64)))
2933                } else if content_bytes > THREAD_TURN_CHUNK_BYTES {
2934                    report.chunked_turns += 1;
2935                    (chunk_utf8(&t.content, THREAD_TURN_CHUNK_BYTES), None)
2936                } else {
2937                    (vec![t.content.clone()], None)
2938                };
2939            pending.push(PendingTurn {
2940                turn_id: t.turn_id.clone(),
2941                role: t.role.clone(),
2942                created_at: t.timestamp.unwrap_or(now),
2943                parts,
2944                truncation,
2945            });
2946        }
2947        Ok(IngestPlan {
2948            pending,
2949            next_seq,
2950            report,
2951        })
2952    }
2953
2954    /// Write half of [`Mnemo::ingest_thread`]. `vectors` holds one vector
2955    /// per text of `plan.texts()`, already checked. `ttl` stamps every
2956    /// new record (the `commit_deltas` `ExpireAfter` policy).
2957    fn apply_ingest(
2958        &mut self,
2959        thread_id: &str,
2960        plan: IngestPlan,
2961        vectors: Vec<Vec<f32>>,
2962        ttl: Option<i64>,
2963    ) -> Result<IngestReport> {
2964        let IngestPlan {
2965            pending,
2966            mut next_seq,
2967            mut report,
2968        } = plan;
2969        // Order: each pending turn's chunks in part order; each turn
2970        // gets a single `_mnemo.seq` slot, so `_mnemo.part`
2971        // disambiguates chunks of the same turn.
2972        let mut vectors = vectors.into_iter();
2973        for p in &pending {
2974            let this_seq = next_seq;
2975            next_seq += 1;
2976            for (part_idx, part_text) in p.parts.iter().enumerate() {
2977                let vec = vectors.next().ok_or_else(|| {
2978                    MnemoError::Invalid("ingest_thread: ran out of vectors".into())
2979                })?;
2980                let mut m =
2981                    Memory::new(part_text.clone(), MemoryType::Episodic, vec).with_importance(0.5);
2982                if let Some(t) = ttl {
2983                    m = m.with_ttl(t);
2984                }
2985                m.created_at = p.created_at;
2986                m.accessed_at = p.created_at;
2987                m.metadata.insert(
2988                    MNEMO_META_KIND.into(),
2989                    Value::String(MNEMO_META_KIND_TURN.into()),
2990                );
2991                m.metadata.insert(
2992                    MNEMO_META_THREAD.into(),
2993                    Value::String(thread_id.to_string()),
2994                );
2995                m.metadata
2996                    .insert(MNEMO_META_TURN_ID.into(), Value::String(p.turn_id.clone()));
2997                m.metadata
2998                    .insert(MNEMO_META_SEQ.into(), Value::Number(this_seq.into()));
2999                m.metadata
3000                    .insert(MNEMO_META_ROLE.into(), Value::String(p.role.clone()));
3001                if p.parts.len() > 1 {
3002                    m.metadata.insert(
3003                        MNEMO_META_PART.into(),
3004                        Value::Number((part_idx as u64).into()),
3005                    );
3006                }
3007                if let Some((sha, len)) = &p.truncation {
3008                    m.metadata
3009                        .insert(MNEMO_META_SHA256.into(), Value::String(sha.clone()));
3010                    m.metadata
3011                        .insert(MNEMO_META_LENGTH.into(), Value::Number((*len).into()));
3012                }
3013                self.put(m)?;
3014                report.added_records += 1;
3015            }
3016        }
3017        Ok(report)
3018    }
3019
3020    /// Export the archive layer for `thread_id` in `_mnemo.seq`
3021    /// order. Each entry is the Memory as stored; chunks of a
3022    /// multi-part turn come out in `_mnemo.part` order within their
3023    /// `_mnemo.seq`. The record `.content` is the text that was
3024    /// stored (post-chunking / post-truncation); callers that need
3025    /// the pre-truncation original should round-trip via
3026    /// `_mnemo.sha256` + `_mnemo.length`.
3027    pub fn thread_export(&mut self, thread_id: &str) -> Result<Vec<Memory>> {
3028        // R-004.2: the thread's own records, from the handle caches.
3029        let entries = self.thread_entries(thread_id)?;
3030        let mut out: Vec<Memory> = Vec::new();
3031        for e in &entries {
3032            let m = self.read_memory(e)?;
3033            if m.metadata.get(MNEMO_META_THREAD).and_then(|v| v.as_str()) != Some(thread_id) {
3034                continue;
3035            }
3036            if m.metadata.get(MNEMO_META_KIND).and_then(|v| v.as_str())
3037                != Some(MNEMO_META_KIND_TURN)
3038            {
3039                continue;
3040            }
3041            out.push(m);
3042        }
3043        out.sort_by(|a, b| {
3044            let seq_a = a
3045                .metadata
3046                .get(MNEMO_META_SEQ)
3047                .and_then(|v| v.as_u64())
3048                .unwrap_or(0);
3049            let seq_b = b
3050                .metadata
3051                .get(MNEMO_META_SEQ)
3052                .and_then(|v| v.as_u64())
3053                .unwrap_or(0);
3054            let part_a = a
3055                .metadata
3056                .get(MNEMO_META_PART)
3057                .and_then(|v| v.as_u64())
3058                .unwrap_or(0);
3059            let part_b = b
3060                .metadata
3061                .get(MNEMO_META_PART)
3062                .and_then(|v| v.as_u64())
3063                .unwrap_or(0);
3064            seq_a.cmp(&seq_b).then_with(|| part_a.cmp(&part_b))
3065        });
3066        Ok(out)
3067    }
3068
3069    /// Export a thread's archive in the **ingest shape** (Phase 12.7d):
3070    /// one [`ExportedTurn`] per source turn, in `_mnemo.seq` order, with
3071    /// a chunked turn's parts joined back into one `content`. Passing
3072    /// the result back to [`Mnemo::ingest_thread`] for the same thread
3073    /// skips every turn; for a new thread it re-creates the same records.
3074    /// This is what Python `db.thread_export`, MCP `thread_export` and
3075    /// `mnemo thread export` return; [`Mnemo::thread_export`] still
3076    /// returns the raw records.
3077    pub fn thread_export_turns(&mut self, thread_id: &str) -> Result<Vec<ExportedTurn>> {
3078        let records = self.thread_export(thread_id)?;
3079        let mut out: Vec<ExportedTurn> = Vec::new();
3080        let mut last_seq: Option<u64> = None;
3081        for m in records {
3082            let meta_str = |k: &str| {
3083                m.metadata
3084                    .get(k)
3085                    .and_then(|v| v.as_str())
3086                    .unwrap_or("")
3087                    .to_string()
3088            };
3089            let seq = m.metadata.get(MNEMO_META_SEQ).and_then(|v| v.as_u64());
3090            // Records are sorted by (seq, part), so the parts of one turn
3091            // are adjacent and share a seq.
3092            if seq.is_some() && seq == last_seq {
3093                if let Some(prev) = out.last_mut() {
3094                    prev.content.push_str(&m.content);
3095                    continue;
3096                }
3097            }
3098            last_seq = seq;
3099            out.push(ExportedTurn {
3100                turn_id: meta_str(MNEMO_META_TURN_ID),
3101                role: meta_str(MNEMO_META_ROLE),
3102                timestamp: m.created_at,
3103                sha256: m
3104                    .metadata
3105                    .get(MNEMO_META_SHA256)
3106                    .and_then(|v| v.as_str())
3107                    .map(String::from),
3108                original_length: m.metadata.get(MNEMO_META_LENGTH).and_then(|v| v.as_u64()),
3109                content: m.content,
3110            });
3111        }
3112        Ok(out)
3113    }
3114
3115    /// Soft-delete every record in the archive layer of `thread_id`.
3116    /// Returns the number of records marked deleted. Permanent
3117    /// removal still requires a subsequent `compact_file`, same as
3118    /// any `Mnemo::delete`.
3119    ///
3120    /// Does **not** touch `_mnemo.kind = "essence"` records or other
3121    /// non-turn metadata the thread may carry — only `_mnemo.kind =
3122    /// "turn"` records matching the thread id.
3123    pub fn thread_forget(&mut self, thread_id: &str) -> Result<usize> {
3124        if self.read_only {
3125            return Err(MnemoError::ReadOnly);
3126        }
3127        // Collect first, then delete — can't hold catalog borrows
3128        // across `self.delete`.
3129        let entries = self.thread_entries(thread_id)?;
3130        let mut victims: Vec<Ulid> = Vec::new();
3131        for e in &entries {
3132            let m = self.read_memory(e)?;
3133            if m.metadata.get(MNEMO_META_THREAD).and_then(|v| v.as_str()) != Some(thread_id) {
3134                continue;
3135            }
3136            if m.metadata.get(MNEMO_META_KIND).and_then(|v| v.as_str())
3137                != Some(MNEMO_META_KIND_TURN)
3138            {
3139                continue;
3140            }
3141            victims.push(m.id);
3142        }
3143        let n = victims.len();
3144        for id in &victims {
3145            self.delete(id)?;
3146        }
3147        Ok(n)
3148    }
3149
3150    // --- Phase 12.7b: essence (timeline + sections) -----------------
3151
3152    /// Phase 12.7b: add a new essence item to a section (any section
3153    /// except [`EssenceSection::Timeline`]). Content ≤
3154    /// [`memory::ESSENCE_SECTION_MAX_BYTES`] (280 bytes); items over
3155    /// that are rejected. Starting status is `"open"`.
3156    ///
3157    /// Use [`Mnemo::essence_append_timeline`] for timeline entries —
3158    /// they have a different cap (600 bytes), are append-only, and
3159    /// carry no status lifecycle.
3160    pub fn essence_add(
3161        &mut self,
3162        scope: EssenceScope,
3163        section: EssenceSection,
3164        text: &str,
3165        refs: Option<Vec<String>>,
3166    ) -> Result<Ulid> {
3167        if self.read_only {
3168            return Err(MnemoError::ReadOnly);
3169        }
3170        Self::check_essence_add(section, text)?;
3171        self.essence_add_with(scope, section, text, refs, None)
3172    }
3173
3174    /// Validation shared by [`Mnemo::essence_add`] and `commit_deltas`.
3175    fn check_essence_add(section: EssenceSection, text: &str) -> Result<()> {
3176        if section == EssenceSection::Timeline {
3177            return Err(MnemoError::Invalid(
3178                "essence_add: use essence_append_timeline for the timeline section".into(),
3179            ));
3180        }
3181        if text.is_empty() {
3182            return Err(MnemoError::Invalid(
3183                "essence_add: text must not be empty".into(),
3184            ));
3185        }
3186        if text.len() > section.item_max_bytes() {
3187            return Err(MnemoError::Invalid(format!(
3188                "essence_add: '{}' items must be <= {} bytes; got {}",
3189                section.as_str(),
3190                section.item_max_bytes(),
3191                text.len()
3192            )));
3193        }
3194        Ok(())
3195    }
3196
3197    /// Write half of `essence_add`, with an optional precomputed vector.
3198    fn essence_add_with(
3199        &mut self,
3200        scope: EssenceScope,
3201        section: EssenceSection,
3202        text: &str,
3203        refs: Option<Vec<String>>,
3204        vector: Option<Vec<f32>>,
3205    ) -> Result<Ulid> {
3206        let now = memory::now_secs();
3207        let mem = self.build_essence_record(scope, section, text, refs, Some(now), vector)?;
3208        self.put(mem)
3209    }
3210
3211    /// Phase 12.7b: append a new timeline entry (append-only
3212    /// milestone). Content ≤ [`memory::ESSENCE_TIMELINE_MAX_BYTES`]
3213    /// (600 bytes); entries over that are rejected (not truncated —
3214    /// truncating a milestone silently loses the narrative).
3215    pub fn essence_append_timeline(
3216        &mut self,
3217        scope: EssenceScope,
3218        text: &str,
3219        refs: Option<Vec<String>>,
3220    ) -> Result<Ulid> {
3221        if self.read_only {
3222            return Err(MnemoError::ReadOnly);
3223        }
3224        Self::check_essence_append(text)?;
3225        self.essence_append_with(scope, text, refs, None)
3226    }
3227
3228    /// Validation shared by [`Mnemo::essence_append_timeline`] and
3229    /// `commit_deltas`.
3230    fn check_essence_append(text: &str) -> Result<()> {
3231        if text.is_empty() {
3232            return Err(MnemoError::Invalid(
3233                "essence_append_timeline: text must not be empty".into(),
3234            ));
3235        }
3236        if text.len() > EssenceSection::Timeline.item_max_bytes() {
3237            return Err(MnemoError::Invalid(format!(
3238                "essence_append_timeline: entries must be <= {} bytes; got {}",
3239                EssenceSection::Timeline.item_max_bytes(),
3240                text.len()
3241            )));
3242        }
3243        Ok(())
3244    }
3245
3246    /// Write half of `essence_append_timeline`, with an optional
3247    /// precomputed vector.
3248    fn essence_append_with(
3249        &mut self,
3250        scope: EssenceScope,
3251        text: &str,
3252        refs: Option<Vec<String>>,
3253        vector: Option<Vec<f32>>,
3254    ) -> Result<Ulid> {
3255        let mem =
3256            self.build_essence_record(scope, EssenceSection::Timeline, text, refs, None, vector)?;
3257        self.put(mem)
3258    }
3259
3260    /// Phase 12.7b: update an existing essence section item in place.
3261    /// Returns [`WriteOutcome::Updated`] / [`WriteOutcome::Unchanged`].
3262    /// Timeline entries are append-only — calling this on a timeline
3263    /// record is an error.
3264    pub fn essence_update(
3265        &mut self,
3266        id: &Ulid,
3267        text: &str,
3268        refs: Option<Vec<String>>,
3269    ) -> Result<WriteOutcome> {
3270        if self.read_only {
3271            return Err(MnemoError::ReadOnly);
3272        }
3273        let existing = self.load_essence_record(id)?;
3274        let section = self.section_of(&existing)?;
3275        if section == EssenceSection::Timeline {
3276            return Err(MnemoError::Invalid(
3277                "essence_update: timeline entries are append-only; append a new one instead".into(),
3278            ));
3279        }
3280        if text.is_empty() {
3281            return Err(MnemoError::Invalid(
3282                "essence_update: text must not be empty".into(),
3283            ));
3284        }
3285        if text.len() > section.item_max_bytes() {
3286            return Err(MnemoError::Invalid(format!(
3287                "essence_update: '{}' items must be <= {} bytes; got {}",
3288                section.as_str(),
3289                section.item_max_bytes(),
3290                text.len()
3291            )));
3292        }
3293        let mut patch = MemoryPatch {
3294            content: Some(text.to_string()),
3295            ..Default::default()
3296        };
3297        // We don't have a usable embedder-free path for an in-place
3298        // content update, so provide the same vector as the existing
3299        // record (the text-vector drift guard in `update` will accept
3300        // this).
3301        patch.vector = Some(existing.vector.clone());
3302        if let Some(r) = refs {
3303            let mut meta = Map::new();
3304            // Preserve the user-visible metadata (none for essence —
3305            // everything lives under `_mnemo.*`) and let `update`'s
3306            // metadata-replacement path keep the reserved keys.
3307            meta.insert(
3308                MNEMO_META_REFS.into(),
3309                Value::Array(r.into_iter().map(Value::String).collect()),
3310            );
3311            patch.metadata = Some(meta);
3312        }
3313        self.update(id, patch)
3314    }
3315
3316    /// Phase 12.7b: mark a section item as resolved (open → resolved
3317    /// transition). A new history entry lands on
3318    /// `_mnemo.status_history`. Idempotent — resolving a record that
3319    /// is already resolved is a no-op.
3320    pub fn essence_resolve(&mut self, id: &Ulid) -> Result<WriteOutcome> {
3321        self.essence_transition_status(id, essence_status::RESOLVED)
3322    }
3323
3324    /// Phase 12.7b: soft-delete a section item. Same semantics as
3325    /// [`Mnemo::delete`] on any record — space reclaimed by a later
3326    /// compact.
3327    pub fn essence_remove(&mut self, id: &Ulid) -> Result<()> {
3328        if self.read_only {
3329            return Err(MnemoError::ReadOnly);
3330        }
3331        self.delete(id)
3332    }
3333
3334    /// Phase 12.7b: deterministic, budgeted markdown render of the
3335    /// essence layer for a thread (or the project layer when
3336    /// `scope = EssenceScope::Project`). `max_bytes` caps the output;
3337    /// sections are emitted in the fixed [`EssenceSection::render_order`],
3338    /// and each section's own budget is proportional to its position
3339    /// in the order. Oldest timeline entries fold into a
3340    /// `+N earlier milestones` tail when the budget runs out.
3341    ///
3342    /// The render always ends with an archive pointer so the reader
3343    /// knows how to drill in deeper (`mnemo thread export …` or
3344    /// `recall --thread …`).
3345    pub fn essence(&mut self, scope: EssenceScope, max_bytes: usize) -> Result<String> {
3346        let max_bytes = if max_bytes == 0 {
3347            ESSENCE_DEFAULT_RENDER_BYTES
3348        } else {
3349            max_bytes
3350        };
3351        // Collect every live essence record for this scope.
3352        // R-004.2: only the essence items, from the handle caches.
3353        let entries = self.essence_entries()?;
3354        let mut by_section: std::collections::BTreeMap<&'static str, Vec<Memory>> =
3355            std::collections::BTreeMap::new();
3356        for section in EssenceSection::render_order() {
3357            by_section.insert(section.as_str(), Vec::new());
3358        }
3359        for e in &entries {
3360            let m = self.read_memory(e)?;
3361            if m.metadata.get(MNEMO_META_KIND).and_then(|v| v.as_str())
3362                != Some(MNEMO_META_KIND_ESSENCE)
3363            {
3364                continue;
3365            }
3366            // Scope filter. Project-scope records are shown in every
3367            // thread's render (and in the project render); thread-scope
3368            // records only match when scope == same thread.
3369            let rec_thread = m.metadata.get(MNEMO_META_THREAD).and_then(|v| v.as_str());
3370            let matches_scope = match (&scope, rec_thread) {
3371                (EssenceScope::Project, None) => true,
3372                (EssenceScope::Thread(_), None) => true, // project-scope always shows
3373                (EssenceScope::Thread(t), Some(rt)) => rt == t,
3374                (EssenceScope::Project, Some(_)) => false,
3375            };
3376            if !matches_scope {
3377                continue;
3378            }
3379            let section_str = m
3380                .metadata
3381                .get(MNEMO_META_SECTION)
3382                .and_then(|v| v.as_str())
3383                .unwrap_or("timeline");
3384            by_section
3385                .entry(
3386                    EssenceSection::parse(section_str)
3387                        .map(|s| s.as_str())
3388                        .unwrap_or("timeline"),
3389                )
3390                .or_default()
3391                .push(m);
3392        }
3393
3394        // Sort each section's records:
3395        // - Decisions + Timeline: newest-first (by created_at desc).
3396        // - Everything else: sort open items before resolved, then
3397        //   newest-first. Cheap, deterministic.
3398        for (name, items) in by_section.iter_mut() {
3399            let newest_first = matches!(*name, "decisions" | "timeline");
3400            items.sort_by(|a, b| {
3401                let a_resolved = status_of(a) == Some(essence_status::RESOLVED);
3402                let b_resolved = status_of(b) == Some(essence_status::RESOLVED);
3403                match (a_resolved, b_resolved) {
3404                    (false, true) => std::cmp::Ordering::Less,
3405                    (true, false) => std::cmp::Ordering::Greater,
3406                    _ => {
3407                        if newest_first {
3408                            b.created_at.cmp(&a.created_at)
3409                        } else {
3410                            a.created_at.cmp(&b.created_at)
3411                        }
3412                    }
3413                }
3414            });
3415        }
3416
3417        // Per-section budget: give intent / conventions / decisions a
3418        // larger share (they frame the thread), timeline a smaller
3419        // share (it can fold). This is deterministic because it
3420        // depends only on `max_bytes`.
3421        let header_reserve = 256usize; // header + archive pointer
3422        let body_budget = max_bytes.saturating_sub(header_reserve);
3423        let per_section_budget = |section: EssenceSection| -> usize {
3424            match section {
3425                EssenceSection::Intent => body_budget / 8,
3426                EssenceSection::Conventions => body_budget / 6,
3427                EssenceSection::Decisions => body_budget / 4,
3428                EssenceSection::State => body_budget / 8,
3429                EssenceSection::Open => body_budget / 8,
3430                EssenceSection::Next => body_budget / 12,
3431                EssenceSection::Lessons => body_budget / 12,
3432                EssenceSection::Timeline => body_budget / 4,
3433            }
3434        };
3435
3436        let mut out = String::with_capacity(max_bytes);
3437        let scope_label = match &scope {
3438            EssenceScope::Thread(t) => format!("thread `{t}`"),
3439            EssenceScope::Project => "project".to_string(),
3440        };
3441        out.push_str(&format!("# essence — {scope_label}\n\n"));
3442
3443        // Phase 12.7c: continuation. If this is a thread render AND
3444        // the thread has a `_mnemo.continues` link, pull the prior
3445        // thread's open section items, its `next`, and its last-N
3446        // timeline entries in ahead of the new thread's own essence.
3447        if let EssenceScope::Thread(new_id) = &scope {
3448            if let Some(prev_id) = self.predecessor_thread(new_id)? {
3449                let continuation = self
3450                    .render_continuation_prelude(&prev_id, max_bytes.saturating_sub(out.len()))?;
3451                if !continuation.is_empty() && out.len() + continuation.len() <= max_bytes {
3452                    out.push_str(&continuation);
3453                }
3454            }
3455        }
3456
3457        for section in EssenceSection::render_order() {
3458            let items = match by_section.get(section.as_str()) {
3459                Some(v) if !v.is_empty() => v,
3460                _ => continue,
3461            };
3462            let budget = per_section_budget(*section);
3463            let section_header = format!("## {}\n", section.as_str());
3464            if out.len() + section_header.len() > max_bytes {
3465                break;
3466            }
3467            out.push_str(&section_header);
3468            let mut used = 0usize;
3469            let mut rendered = 0usize;
3470            for item in items {
3471                let line = render_essence_item(*section, item);
3472                if used + line.len() > budget {
3473                    break;
3474                }
3475                if out.len() + line.len() > max_bytes {
3476                    break;
3477                }
3478                out.push_str(&line);
3479                used += line.len();
3480                rendered += 1;
3481            }
3482            // Fold the remainder for timeline.
3483            if *section == EssenceSection::Timeline && rendered < items.len() {
3484                let fold = format!("- +{} earlier milestones\n", items.len() - rendered);
3485                if out.len() + fold.len() <= max_bytes {
3486                    out.push_str(&fold);
3487                }
3488            }
3489            if out.len() < max_bytes {
3490                out.push('\n');
3491            }
3492        }
3493
3494        // Archive pointer.
3495        let pointer = match &scope {
3496            EssenceScope::Thread(t) => format!(
3497                "> full history: `mnemo thread export <file> --thread {t}` or `recall --thread {t}`\n"
3498            ),
3499            EssenceScope::Project => String::from(
3500                "> full history: `mnemo thread export <file> --thread <id>` or `recall --thread <id>`\n",
3501            ),
3502        };
3503        if out.len() + pointer.len() <= max_bytes {
3504            out.push_str(&pointer);
3505        }
3506        Ok(out)
3507    }
3508
3509    // --- essence internal helpers ----------------------------------
3510
3511    /// Build an essence record. `vector` is a precomputed embedding (the
3512    /// 12.7d `commit_deltas` batch path); `None` embeds `text` with the
3513    /// registered embedder, or falls back to an axis-neutral unit vector
3514    /// when there is none. Essence records are read by `_mnemo.kind` and
3515    /// scope, not by similarity, so the fallback costs no retrieval
3516    /// quality; it only has to pass `put`'s non-zero guard.
3517    fn build_essence_record(
3518        &self,
3519        scope: EssenceScope,
3520        section: EssenceSection,
3521        text: &str,
3522        refs: Option<Vec<String>>,
3523        status_at: Option<i64>,
3524        vector: Option<Vec<f32>>,
3525    ) -> Result<Memory> {
3526        let v = match (vector, self.embedder.as_ref()) {
3527            (Some(v), _) => v,
3528            (None, Some(e)) => e.embed(&[text])?.pop().ok_or_else(|| {
3529                MnemoError::Invalid("essence: embedder returned no vector".into())
3530            })?,
3531            (None, None) => self.sentinel_vector(),
3532        };
3533        let mut m = Memory::new(text.to_string(), MemoryType::Semantic, v).with_importance(0.5);
3534        m.metadata.insert(
3535            MNEMO_META_KIND.into(),
3536            Value::String(MNEMO_META_KIND_ESSENCE.into()),
3537        );
3538        m.metadata.insert(
3539            MNEMO_META_SECTION.into(),
3540            Value::String(section.as_str().into()),
3541        );
3542        if let EssenceScope::Thread(t) = &scope {
3543            m.metadata
3544                .insert(MNEMO_META_THREAD.into(), Value::String(t.clone()));
3545        }
3546        if let Some(r) = refs {
3547            m.metadata.insert(
3548                MNEMO_META_REFS.into(),
3549                Value::Array(r.into_iter().map(Value::String).collect()),
3550            );
3551        }
3552        if let Some(at) = status_at {
3553            m.metadata.insert(
3554                MNEMO_META_STATUS.into(),
3555                Value::String(essence_status::OPEN.into()),
3556            );
3557            m.metadata.insert(
3558                MNEMO_META_STATUS_HISTORY.into(),
3559                Value::Array(vec![serde_json::json!({
3560                    "status": essence_status::OPEN,
3561                    "at": at,
3562                })]),
3563            );
3564        }
3565        Ok(m)
3566    }
3567
3568    /// Axis-neutral unit vector (every component `1/sqrt(dims)`), used by
3569    /// essence records without an embedder and by thread-link markers.
3570    fn sentinel_vector(&self) -> Vec<f32> {
3571        let dims = self.dimensions.max(1);
3572        vec![1.0 / (dims as f32).sqrt(); dims]
3573    }
3574
3575    fn load_essence_record(&mut self, id: &Ulid) -> Result<Memory> {
3576        let m = self.get(id)?;
3577        if m.metadata.get(MNEMO_META_KIND).and_then(|v| v.as_str()) != Some(MNEMO_META_KIND_ESSENCE)
3578        {
3579            return Err(MnemoError::Invalid(format!(
3580                "essence: record {id} is not an essence record (missing `_mnemo.kind = essence`)"
3581            )));
3582        }
3583        Ok(m)
3584    }
3585
3586    fn section_of(&self, m: &Memory) -> Result<EssenceSection> {
3587        let s = m
3588            .metadata
3589            .get(MNEMO_META_SECTION)
3590            .and_then(|v| v.as_str())
3591            .ok_or_else(|| {
3592                MnemoError::Invalid("essence: record missing `_mnemo.section`".into())
3593            })?;
3594        EssenceSection::parse(s)
3595            .ok_or_else(|| MnemoError::Invalid(format!("essence: unknown section '{s}' on record")))
3596    }
3597
3598    fn essence_transition_status(&mut self, id: &Ulid, to: &str) -> Result<WriteOutcome> {
3599        if self.read_only {
3600            return Err(MnemoError::ReadOnly);
3601        }
3602        let existing = self.load_essence_record(id)?;
3603        let section = self.section_of(&existing)?;
3604        if !section.has_status_lifecycle() {
3605            return Err(MnemoError::Invalid(format!(
3606                "essence_transition: '{}' section has no status lifecycle",
3607                section.as_str()
3608            )));
3609        }
3610        let from = existing
3611            .metadata
3612            .get(MNEMO_META_STATUS)
3613            .and_then(|v| v.as_str())
3614            .unwrap_or(essence_status::OPEN);
3615        if from == to {
3616            return Ok(WriteOutcome::Unchanged);
3617        }
3618        let allowed = ESSENCE_TRANSITIONS
3619            .iter()
3620            .any(|(f, t)| *f == from && *t == to);
3621        if !allowed {
3622            return Err(MnemoError::Invalid(format!(
3623                "essence_transition: '{from}' → '{to}' is not a valid transition for essence"
3624            )));
3625        }
3626        // Mutate metadata in place and `put` back under the same id
3627        // so `_mnemo.updated_at` gets stamped and access stats are
3628        // preserved.
3629        let mut new = existing.clone();
3630        new.metadata
3631            .insert(MNEMO_META_STATUS.into(), Value::String(to.to_string()));
3632        let now = memory::now_secs();
3633        let history_entry = serde_json::json!({ "status": to, "at": now });
3634        let history = new
3635            .metadata
3636            .entry(MNEMO_META_STATUS_HISTORY.to_string())
3637            .or_insert_with(|| Value::Array(Vec::new()));
3638        if let Value::Array(arr) = history {
3639            arr.push(history_entry);
3640        }
3641        new.metadata
3642            .insert(MNEMO_META_UPDATED_AT.into(), Value::Number(now.into()));
3643        new.id = existing.id;
3644        new.created_at = existing.created_at;
3645        self.put(new)?;
3646        Ok(WriteOutcome::Updated)
3647    }
3648
3649    // --- Phase 12.7c: continuation + commit_deltas ------------------
3650
3651    /// Start a new thread that continues from a previous one
3652    /// (Phase 12.7c). Writes a thread-link marker record with
3653    /// `_mnemo.kind = "thread_link"` + `_mnemo.thread = <new_id>` +
3654    /// `_mnemo.continues = <from_id>`. The marker holds no user
3655    /// content; its purpose is to let `essence(Thread(new_id))`
3656    /// discover the continuation chain.
3657    ///
3658    /// The render for the new thread pulls the previous thread's
3659    /// open section items, its `next`, and its last
3660    /// [`CONTINUATION_TIMELINE_ENTRIES`] timeline entries ahead of
3661    /// the new thread's own essence — the "new thread, same
3662    /// experience" path.
3663    pub fn thread_start(&mut self, new_thread_id: &str, from_thread_id: &str) -> Result<Ulid> {
3664        if self.read_only {
3665            return Err(MnemoError::ReadOnly);
3666        }
3667        if new_thread_id.is_empty() || from_thread_id.is_empty() {
3668            return Err(MnemoError::Invalid(
3669                "thread_start: 'new_thread_id' and 'from_thread_id' must not be empty".into(),
3670            ));
3671        }
3672        if new_thread_id == from_thread_id {
3673            return Err(MnemoError::Invalid(
3674                "thread_start: 'new_thread_id' must differ from 'from_thread_id'".into(),
3675            ));
3676        }
3677        // Store a tiny marker record so the continuation chain is
3678        // queryable from a fresh handle. The sentinel vector passes
3679        // `put`'s zero-vector guard; default recall skips link records.
3680        let v = self.sentinel_vector();
3681        let content = format!("thread link: {new_thread_id} continues from {from_thread_id}");
3682        let mut m = Memory::new(content, MemoryType::Semantic, v).with_importance(0.1);
3683        m.metadata.insert(
3684            MNEMO_META_KIND.into(),
3685            Value::String(MNEMO_META_KIND_THREAD_LINK.into()),
3686        );
3687        m.metadata.insert(
3688            MNEMO_META_THREAD.into(),
3689            Value::String(new_thread_id.to_string()),
3690        );
3691        m.metadata.insert(
3692            MNEMO_META_CONTINUES.into(),
3693            Value::String(from_thread_id.to_string()),
3694        );
3695        let id = self.put(m)?;
3696        Ok(id)
3697    }
3698
3699    /// Phase 12.7c helper: render the predecessor thread's open
3700    /// items + `next` + last-N timeline entries as a "continued from"
3701    /// prelude. Used by [`Mnemo::essence`] to carry forward the
3702    /// prior thread's live state into a fresh thread.
3703    fn render_continuation_prelude(
3704        &mut self,
3705        prev_thread: &str,
3706        max_bytes: usize,
3707    ) -> Result<String> {
3708        if max_bytes == 0 {
3709            return Ok(String::new());
3710        }
3711        let entries = self.essence_entries()?;
3712        let mut open_items: Vec<Memory> = Vec::new();
3713        let mut next_items: Vec<Memory> = Vec::new();
3714        let mut timeline_items: Vec<Memory> = Vec::new();
3715        for e in &entries {
3716            let m = self.read_memory(e)?;
3717            if m.metadata.get(MNEMO_META_KIND).and_then(|v| v.as_str())
3718                != Some(MNEMO_META_KIND_ESSENCE)
3719            {
3720                continue;
3721            }
3722            if m.metadata.get(MNEMO_META_THREAD).and_then(|v| v.as_str()) != Some(prev_thread) {
3723                continue;
3724            }
3725            let section = match m
3726                .metadata
3727                .get(MNEMO_META_SECTION)
3728                .and_then(|v| v.as_str())
3729                .and_then(EssenceSection::parse)
3730            {
3731                Some(s) => s,
3732                None => continue,
3733            };
3734            let status = m
3735                .metadata
3736                .get(MNEMO_META_STATUS)
3737                .and_then(|v| v.as_str())
3738                .unwrap_or(essence_status::OPEN);
3739            match section {
3740                EssenceSection::Open if status == essence_status::OPEN => {
3741                    open_items.push(m);
3742                }
3743                EssenceSection::Next if status == essence_status::OPEN => {
3744                    next_items.push(m);
3745                }
3746                EssenceSection::Timeline => timeline_items.push(m),
3747                _ => {}
3748            }
3749        }
3750        // Newest-first timeline, cap to last N.
3751        timeline_items.sort_by_key(|m| std::cmp::Reverse(m.created_at));
3752        timeline_items.truncate(CONTINUATION_TIMELINE_ENTRIES);
3753        // Oldest-first for stable chronological read in the render.
3754        timeline_items.reverse();
3755
3756        let mut out = String::new();
3757        let header = format!("## continued from `{prev_thread}`\n");
3758        if header.len() > max_bytes {
3759            return Ok(String::new());
3760        }
3761        out.push_str(&header);
3762        if !open_items.is_empty() {
3763            let sub = "### open\n";
3764            if out.len() + sub.len() <= max_bytes {
3765                out.push_str(sub);
3766            }
3767            for m in &open_items {
3768                let line = format!("- {}\n", m.content);
3769                if out.len() + line.len() > max_bytes {
3770                    break;
3771                }
3772                out.push_str(&line);
3773            }
3774        }
3775        if !next_items.is_empty() {
3776            let sub = "### next\n";
3777            if out.len() + sub.len() <= max_bytes {
3778                out.push_str(sub);
3779            }
3780            for m in &next_items {
3781                let line = format!("- {}\n", m.content);
3782                if out.len() + line.len() > max_bytes {
3783                    break;
3784                }
3785                out.push_str(&line);
3786            }
3787        }
3788        if !timeline_items.is_empty() {
3789            let sub = "### recent milestones\n";
3790            if out.len() + sub.len() <= max_bytes {
3791                out.push_str(sub);
3792            }
3793            for m in &timeline_items {
3794                let line = format!("- {}\n", m.content);
3795                if out.len() + line.len() > max_bytes {
3796                    break;
3797                }
3798                out.push_str(&line);
3799            }
3800        }
3801        if out.len() < max_bytes {
3802            out.push('\n');
3803        }
3804        Ok(out)
3805    }
3806
3807    /// Phase 12.7c helper: find the predecessor thread id for
3808    /// `thread_id` by reading its `_mnemo.kind = "thread_link"`
3809    /// marker. Returns `None` if the thread has no predecessor
3810    /// (fresh thread, no `thread_start` call).
3811    fn predecessor_thread(&mut self, thread_id: &str) -> Result<Option<String>> {
3812        self.ensure_key_caches()?;
3813        Ok(self
3814            .layers
3815            .as_ref()
3816            .and_then(|l| l.links.get(thread_id))
3817            .map(|(from, _)| from.clone()))
3818    }
3819
3820    /// Atomic batch write combining 12.1 facts, 12.7b essence ops,
3821    /// and 12.7a archive turns in one flush (Phase 12.7c; batch
3822    /// embedding added in 12.7d).
3823    ///
3824    /// The batch runs in four steps:
3825    ///
3826    /// 1. **Structural checks** on every item: required fields, size
3827    ///    caps, empty keys, `thread_id` for turns, and that every
3828    ///    `Update` / `Resolve` / `Remove` targets a live essence record
3829    ///    the op can apply to.
3830    /// 2. **One `Embedder::embed` call** for every text that needs a
3831    ///    vector: facts whose `vector` is empty, essence `Add` /
3832    ///    `AppendTimeline` texts (when an embedder is registered;
3833    ///    without one, essence keeps its sentinel vector), and the
3834    ///    archive turns' chunks.
3835    /// 3. **Vector checks** (width, non-zero, finite) on supplied and
3836    ///    embedded vectors alike.
3837    /// 4. **Writes**: facts, then essence ops, then turns per
3838    ///    `turn_policy`, then exactly one `flush()`.
3839    ///
3840    /// A failure in steps 1–3 returns an error with nothing staged. A
3841    /// text-only fact, or any turn to archive, with no embedder
3842    /// registered is a step-1 error. `put` has no rollback, so an I/O
3843    /// error during step 4 can leave the batch's earlier writes staged
3844    /// on the handle; steps 1–3 exist so nothing else can fail that
3845    /// late.
3846    pub fn commit_deltas(&mut self, batch: CommitBatch) -> Result<CommitReport> {
3847        if self.read_only {
3848            return Err(MnemoError::ReadOnly);
3849        }
3850        let CommitBatch {
3851            mut facts,
3852            essence_ops,
3853            turns,
3854            thread_id,
3855            turn_policy,
3856        } = batch;
3857        let has_embedder = self.embedder.is_some();
3858        let in_ctx = |ctx: &str, e: MnemoError| MnemoError::Invalid(format!("{ctx}: {e}"));
3859
3860        // --- 1. Structural checks (reads only) ----------------------
3861        for (i, f) in facts.iter().enumerate() {
3862            if let Some(k) = &f.key {
3863                if k.is_empty() {
3864                    return Err(MnemoError::Invalid(format!(
3865                        "commit_deltas: facts[{i}] key is empty"
3866                    )));
3867                }
3868            }
3869            if f.memory.vector.is_empty() {
3870                if f.memory.content.is_empty() {
3871                    return Err(MnemoError::Invalid(format!(
3872                        "commit_deltas: facts[{i}] has neither a vector nor content to embed"
3873                    )));
3874                }
3875                if !has_embedder {
3876                    return Err(MnemoError::Invalid(format!(
3877                        "commit_deltas: facts[{i}] has no vector and no embedder is registered \
3878                         on this handle; pass a vector or call set_embedder first"
3879                    )));
3880                }
3881            }
3882        }
3883        let mut removed_in_batch: std::collections::HashSet<u128> =
3884            std::collections::HashSet::new();
3885        for (i, op) in essence_ops.iter().enumerate() {
3886            let ctx = format!("commit_deltas: essence_ops[{i}]");
3887            match op.kind {
3888                EssenceOpKind::Add => {
3889                    if op.scope.is_none() {
3890                        return Err(MnemoError::Invalid(format!("{ctx} Add missing 'scope'")));
3891                    }
3892                    let section = op.section.ok_or_else(|| {
3893                        MnemoError::Invalid(format!("{ctx} Add missing 'section'"))
3894                    })?;
3895                    let text = op
3896                        .text
3897                        .as_deref()
3898                        .ok_or_else(|| MnemoError::Invalid(format!("{ctx} Add missing 'text'")))?;
3899                    Self::check_essence_add(section, text).map_err(|e| in_ctx(&ctx, e))?;
3900                }
3901                EssenceOpKind::AppendTimeline => {
3902                    if op.scope.is_none() {
3903                        return Err(MnemoError::Invalid(format!(
3904                            "{ctx} AppendTimeline missing 'scope'"
3905                        )));
3906                    }
3907                    let text = op.text.as_deref().ok_or_else(|| {
3908                        MnemoError::Invalid(format!("{ctx} AppendTimeline missing 'text'"))
3909                    })?;
3910                    Self::check_essence_append(text).map_err(|e| in_ctx(&ctx, e))?;
3911                }
3912                EssenceOpKind::Update | EssenceOpKind::Resolve | EssenceOpKind::Remove => {
3913                    let id = op
3914                        .id
3915                        .ok_or_else(|| MnemoError::Invalid(format!("{ctx} needs 'id'")))?;
3916                    if removed_in_batch.contains(&id.0) {
3917                        return Err(MnemoError::Invalid(format!(
3918                            "{ctx}: record {id} is removed earlier in this batch"
3919                        )));
3920                    }
3921                    let existing = self.load_essence_record(&id).map_err(|e| in_ctx(&ctx, e))?;
3922                    let section = self.section_of(&existing).map_err(|e| in_ctx(&ctx, e))?;
3923                    match op.kind {
3924                        EssenceOpKind::Update => {
3925                            let text = op.text.as_deref().ok_or_else(|| {
3926                                MnemoError::Invalid(format!("{ctx} Update needs 'text'"))
3927                            })?;
3928                            if section == EssenceSection::Timeline {
3929                                return Err(MnemoError::Invalid(format!(
3930                                    "{ctx}: timeline entries are append-only"
3931                                )));
3932                            }
3933                            if text.is_empty() || text.len() > section.item_max_bytes() {
3934                                return Err(MnemoError::Invalid(format!(
3935                                    "{ctx}: '{}' items must be 1..={} bytes; got {}",
3936                                    section.as_str(),
3937                                    section.item_max_bytes(),
3938                                    text.len()
3939                                )));
3940                            }
3941                        }
3942                        EssenceOpKind::Resolve => {
3943                            if !section.has_status_lifecycle() {
3944                                return Err(MnemoError::Invalid(format!(
3945                                    "{ctx}: '{}' section has no status lifecycle",
3946                                    section.as_str()
3947                                )));
3948                            }
3949                        }
3950                        _ => {
3951                            removed_in_batch.insert(id.0);
3952                        }
3953                    }
3954                }
3955            }
3956        }
3957        if !turns.is_empty() && thread_id.is_none() {
3958            return Err(MnemoError::Invalid(
3959                "commit_deltas: 'turns' non-empty but 'thread_id' missing".into(),
3960            ));
3961        }
3962        let archive_turns = !turns.is_empty() && !matches!(turn_policy, TurnPolicy::Drop);
3963        if archive_turns && !has_embedder {
3964            return Err(MnemoError::Invalid(
3965                "commit_deltas: archiving 'turns' needs an embedder registered on this handle; \
3966                 call set_embedder first or use turn_policy Drop"
3967                    .into(),
3968            ));
3969        }
3970        let plan = match (&thread_id, archive_turns) {
3971            (Some(t), true) => Some(self.plan_ingest(t, &turns)?),
3972            _ => None,
3973        };
3974
3975        // --- 2. One batched embed -----------------------------------
3976        let embeds_essence = |op: &EssenceOp| {
3977            has_embedder && matches!(op.kind, EssenceOpKind::Add | EssenceOpKind::AppendTimeline)
3978        };
3979        let (vectors, n_texts) = {
3980            let mut texts: Vec<&str> = Vec::new();
3981            for f in &facts {
3982                if f.memory.vector.is_empty() {
3983                    texts.push(f.memory.content.as_str());
3984                }
3985            }
3986            for op in essence_ops.iter().filter(|op| embeds_essence(op)) {
3987                texts.push(op.text.as_deref().expect("checked in step 1"));
3988            }
3989            if let Some(p) = &plan {
3990                texts.extend(p.texts());
3991            }
3992            let vectors = match (texts.is_empty(), self.embedder.as_ref()) {
3993                (true, _) => Vec::new(),
3994                (false, Some(e)) => e.embed(&texts)?,
3995                (false, None) => unreachable!("texts need an embedder; checked in step 1"),
3996            };
3997            (vectors, texts.len())
3998        };
3999
4000        // --- 3. Vector checks ---------------------------------------
4001        self.check_vectors("commit_deltas", &vectors, n_texts)?;
4002        for (i, f) in facts.iter().enumerate() {
4003            let v = &f.memory.vector;
4004            if v.is_empty() {
4005                continue;
4006            }
4007            if v.len() != self.dimensions {
4008                return Err(MnemoError::Invalid(format!(
4009                    "commit_deltas: facts[{i}] vector has {} dims; this file is {}",
4010                    v.len(),
4011                    self.dimensions
4012                )));
4013            }
4014            let ctx = format!("commit_deltas: facts[{i}]");
4015            self.check_vector(&ctx, v)?;
4016        }
4017
4018        // Hand the embedded vectors back out, in the order they were
4019        // requested: facts, then essence ops, then turn chunks.
4020        let mut vectors = vectors.into_iter();
4021        for f in facts.iter_mut() {
4022            if f.memory.vector.is_empty() {
4023                f.memory.vector = vectors.next().expect("one vector per text");
4024            }
4025        }
4026        let essence_vectors: Vec<Option<Vec<f32>>> = essence_ops
4027            .iter()
4028            .map(|op| embeds_essence(op).then(|| vectors.next().expect("one vector per text")))
4029            .collect();
4030        let turn_vectors: Vec<Vec<f32>> = vectors.collect();
4031
4032        // --- 4. Writes ----------------------------------------------
4033        let mut report = CommitReport::default();
4034
4035        for f in facts {
4036            match f.key {
4037                Some(k) => {
4038                    let (_id, outcome) = self.remember_keyed(f.memory, &k)?;
4039                    match outcome {
4040                        WriteOutcome::Added => report.facts_added += 1,
4041                        WriteOutcome::Superseded | WriteOutcome::Updated => {
4042                            report.facts_superseded += 1
4043                        }
4044                        WriteOutcome::Unchanged => report.facts_unchanged += 1,
4045                    }
4046                }
4047                None => {
4048                    self.remember(f.memory)?;
4049                    report.facts_added += 1;
4050                }
4051            }
4052        }
4053
4054        for (op, vector) in essence_ops.into_iter().zip(essence_vectors) {
4055            let outcome = match op.kind {
4056                EssenceOpKind::Add => {
4057                    self.essence_add_with(
4058                        op.scope.expect("checked in step 1"),
4059                        op.section.expect("checked in step 1"),
4060                        op.text.as_deref().expect("checked in step 1"),
4061                        op.refs,
4062                        vector,
4063                    )?;
4064                    WriteOutcome::Added
4065                }
4066                EssenceOpKind::AppendTimeline => {
4067                    self.essence_append_with(
4068                        op.scope.expect("checked in step 1"),
4069                        op.text.as_deref().expect("checked in step 1"),
4070                        op.refs,
4071                        vector,
4072                    )?;
4073                    WriteOutcome::Added
4074                }
4075                EssenceOpKind::Update => self.essence_update(
4076                    op.id.as_ref().expect("checked in step 1"),
4077                    op.text.as_deref().expect("checked in step 1"),
4078                    op.refs,
4079                )?,
4080                EssenceOpKind::Resolve => {
4081                    self.essence_resolve(op.id.as_ref().expect("checked in step 1"))?
4082                }
4083                EssenceOpKind::Remove => {
4084                    self.essence_remove(op.id.as_ref().expect("checked in step 1"))?;
4085                    WriteOutcome::Updated
4086                }
4087            };
4088            match outcome {
4089                WriteOutcome::Unchanged => report.essence_ops_unchanged += 1,
4090                _ => report.essence_ops_applied += 1,
4091            }
4092        }
4093
4094        if let (Some(plan), Some(t)) = (plan, thread_id.as_deref()) {
4095            let ttl = match turn_policy {
4096                TurnPolicy::ExpireAfter(secs) => Some(secs),
4097                _ => None,
4098            };
4099            let r = self.apply_ingest(t, plan, turn_vectors, ttl)?;
4100            report.turns_added += r.added_records;
4101            report.turns_skipped += r.skipped_existing_turns;
4102            report.turns_truncated += r.truncated_turns;
4103            report.turns_chunked += r.chunked_turns;
4104        }
4105
4106        self.flush()?;
4107        Ok(report)
4108    }
4109
4110    /// P-0 device budget breakdown of this file and handle. **Unstable
4111    /// diagnostic**; see [`BudgetReport`]. Reads the header, the catalog
4112    /// and file metadata only; nothing is decrypted.
4113    #[doc(hidden)]
4114    pub fn __budget_report(&self) -> Result<BudgetReport> {
4115        let h = &self.header;
4116        let ps = h.page_size as u64;
4117        let mut r = BudgetReport {
4118            file_bytes: std::fs::metadata(&self.path)?.len(),
4119            page_size: ps,
4120            header_bytes: ps,
4121            catalog_run_bytes: h.catalog_pages * ps,
4122            catalog_payload_bytes: h.catalog_len,
4123            index_run_bytes: h.index_pages * ps,
4124            index_payload_bytes: h.index_len,
4125            index_start_page: h.index_start,
4126            wal_bytes: h.wal_pages * ps,
4127            snapshot_manifest_bytes: h.manifest_pages * ps,
4128            cache_dir_run_bytes: h.cache_pages * ps,
4129            ..Default::default()
4130        };
4131        for e in &self.catalog {
4132            let pages = e.page_count as u64 * ps;
4133            if e.deleted {
4134                r.deleted_record_page_bytes += pages;
4135            } else {
4136                r.record_page_bytes += pages;
4137                r.record_payload_bytes += e.len as u64;
4138                r.live_records += 1;
4139            }
4140        }
4141        let accounted = r.header_bytes
4142            + r.record_page_bytes
4143            + r.deleted_record_page_bytes
4144            + r.catalog_run_bytes
4145            + r.index_run_bytes
4146            + r.wal_bytes
4147            + r.snapshot_manifest_bytes
4148            + r.cache_dir_run_bytes;
4149        r.unreferenced_bytes = r.file_bytes.saturating_sub(accounted);
4150
4151        r.ram_catalog = (self.catalog.capacity() * std::mem::size_of::<CatalogEntry>()) as u64;
4152        r.ram_id_map = self.index.heap_bytes() as u64;
4153        if let Some(ann) = &self.ann {
4154            let (c, cb, ids, codes, assign) = ann.heap_bytes();
4155            r.ram_ann_centroids = c as u64;
4156            r.ram_ann_codebook = cb as u64;
4157            r.ram_ann_posting_ids = ids as u64;
4158            r.ram_ann_posting_codes = codes as u64;
4159            r.ram_ann_assignment = assign as u64;
4160        }
4161        r.ram_cache_dir =
4162            (self.cache_dir.capacity() * std::mem::size_of::<CacheDirectoryEntry>()) as u64;
4163        let key_index = self.key_index.as_ref().map_or(0, |k| {
4164            crate::index::hash_map_bytes::<KeySlot, u128>(k.capacity())
4165                + k.keys()
4166                    .map(|s| s.key.capacity() + s.scope_owner.as_ref().map_or(0, |o| o.capacity()))
4167                    .sum::<usize>()
4168        });
4169        let superseded = self.superseded.as_ref().map_or(0, |s| {
4170            crate::index::hash_map_bytes::<u128, ()>(s.capacity())
4171        });
4172        r.ram_key_caches = (key_index + superseded) as u64;
4173        r.ram_total = r.ram_catalog
4174            + r.ram_id_map
4175            + r.ram_ann_centroids
4176            + r.ram_ann_codebook
4177            + r.ram_ann_posting_ids
4178            + r.ram_ann_posting_codes
4179            + r.ram_ann_assignment
4180            + r.ram_cache_dir
4181            + r.ram_key_caches;
4182        Ok(r)
4183    }
4184
4185    /// Return every live memory (used by tooling and compaction).
4186    pub fn memories(&mut self) -> Result<Vec<Memory>> {
4187        let entries: Vec<CatalogEntry> = self
4188            .catalog
4189            .iter()
4190            .filter(|e| !e.deleted)
4191            .cloned()
4192            .collect();
4193        let mut out = Vec::with_capacity(entries.len());
4194        for e in &entries {
4195            out.push(self.read_memory(e)?);
4196        }
4197        Ok(out)
4198    }
4199
4200    /// Return the database's **self-describing onboarding memories** — the
4201    /// ones tagged `metadata.area = "onboarding"`. This is the engine-level
4202    /// surface for the *single-file philosophy*: an agent who receives a
4203    /// `.mnemo` file (and its passphrase) can call this to learn what the
4204    /// file is, which embedder it expects, the recommended agent id, and any
4205    /// other conventions the file's author chose to record — all without
4206    /// needing any external documentation.
4207    ///
4208    /// Ordering: the canonical manifest (tag `metadata.topic = "manifest"`)
4209    /// always comes first; everything else follows in `importance` descending,
4210    /// then `created_at` ascending for deterministic results.
4211    pub fn about(&mut self) -> Result<Vec<Memory>> {
4212        // Phase 12.1: drop records that have been superseded by a newer
4213        // keyed write. They still live in snapshot history and
4214        // `history(key)` — just not here.
4215        // R-004.2: read only the onboarding records, from the handle caches.
4216        self.ensure_key_caches()?;
4217        let ids: Vec<u128> = self
4218            .layers
4219            .as_ref()
4220            .map(|l| l.onboarding.iter().copied().collect())
4221            .unwrap_or_default();
4222        let mut out: Vec<Memory> = Vec::new();
4223        for e in self.live_entries(ids) {
4224            let m = self.read_memory(&e)?;
4225            if !m.metadata.contains_key(MNEMO_META_SUPERSEDED_BY) && is_onboarding(&m) {
4226                out.push(m);
4227            }
4228        }
4229        out.sort_by(|a, b| {
4230            let a_manifest = is_manifest(a);
4231            let b_manifest = is_manifest(b);
4232            // Manifest topic wins first; then importance desc; then created_at asc.
4233            b_manifest
4234                .cmp(&a_manifest)
4235                .then_with(|| b.importance.total_cmp(&a.importance))
4236                .then_with(|| a.created_at.cmp(&b.created_at))
4237        });
4238        Ok(out)
4239    }
4240
4241    /// The project essence for the `about` briefing (Phase 12.7d): the
4242    /// [`EssenceScope::Project`] render at the default budget
4243    /// ([`ESSENCE_DEFAULT_RENDER_BYTES`]), or `None` when the file has no
4244    /// open project-scope item or timeline entry. The CLI and MCP `about`
4245    /// append it after the manifest and onboarding memories; [`Mnemo::about`]
4246    /// itself keeps returning records.
4247    pub fn about_essence(&mut self) -> Result<Option<String>> {
4248        let entries = self.essence_entries()?;
4249        let mut any = false;
4250        for e in &entries {
4251            let m = self.read_memory(e)?;
4252            let is_project_essence = m.metadata.get(MNEMO_META_KIND).and_then(|v| v.as_str())
4253                == Some(MNEMO_META_KIND_ESSENCE)
4254                && !m.metadata.contains_key(MNEMO_META_THREAD);
4255            if is_project_essence && status_of(&m) != Some(essence_status::RESOLVED) {
4256                any = true;
4257                break;
4258            }
4259        }
4260        if !any {
4261            return Ok(None);
4262        }
4263        self.essence(EssenceScope::Project, ESSENCE_DEFAULT_RENDER_BYTES)
4264            .map(Some)
4265    }
4266
4267    // --- retrieval --------------------------------------------------------
4268
4269    /// Phase-1 brute-force search: rank live memories by raw similarity only.
4270    /// Read-only — does not update access statistics.
4271    pub fn search(
4272        &mut self,
4273        query: &[f32],
4274        top_k: usize,
4275        metric: Metric,
4276    ) -> Result<Vec<(Memory, f32)>> {
4277        if query.len() != self.dimensions {
4278            return Err(MnemoError::DimensionMismatch {
4279                expected: self.dimensions,
4280                got: query.len(),
4281            });
4282        }
4283        let now = memory::now_secs();
4284        let entries: Vec<CatalogEntry> = self
4285            .catalog
4286            .iter()
4287            .filter(|e| !e.deleted)
4288            .cloned()
4289            .collect();
4290
4291        let mut scored: Vec<(Memory, f32)> = Vec::new();
4292        for e in &entries {
4293            let m = self.read_memory(e)?;
4294            if m.is_expired(now) {
4295                continue;
4296            }
4297            let sim = memory::similarity(metric, query, &m.vector);
4298            scored.push((m, sim));
4299        }
4300        scored.sort_by(|a, b| b.1.total_cmp(&a.1));
4301        scored.truncate(top_k);
4302        Ok(scored)
4303    }
4304
4305    /// Phase-5 multi-signal recall: rank by
4306    /// `α·similarity + β·recency + γ·importance + δ·ln(freq)`, with type and
4307    /// agent filtering. Updates `accessed_at` / `access_count` on the
4308    /// returned memories (persisted on the next `flush`).
4309    ///
4310    /// When an ANN index has been built ([`Mnemo::build_index`]), recall runs
4311    /// the tiered IVF→PQ→rerank pipeline: it scores only the similarity-nearest
4312    /// `n_rerank` candidates rather than every memory. Without an index it
4313    /// scores every live memory exactly.
4314    /// One pass of filter + score for [`Mnemo::recall`]. Factored out of
4315    /// `recall` so the outer path can retry with a wider shortlist when
4316    /// the type/agent/supersession/TTL filters starve the ANN-narrowed
4317    /// result below `top_k` (Phase 12.1 short-results-on-indexed-path
4318    /// fix).
4319    ///
4320    /// * `force_exact = false` → use the ANN index when one exists; fall
4321    ///   back to a full scan only if none is built.
4322    /// * `force_exact = true` → skip the ANN index and scan every live
4323    ///   non-superseded record. Used by the outer path as the last-resort
4324    ///   attempt when widening the shortlist would already touch half or
4325    ///   more of the live non-superseded set.
4326    ///
4327    /// Returns an unsorted, un-truncated `Vec<RecallResult>`; the outer
4328    /// `recall` sorts and truncates once.
4329    fn score_once(
4330        &mut self,
4331        req: &RecallRequest,
4332        now: i64,
4333        force_exact: bool,
4334    ) -> Result<Vec<RecallResult>> {
4335        // Candidate set: ANN-narrowed when an index exists and the
4336        // caller hasn't forced an exact scan, else everything live.
4337        // R-004.2: a filter pinning `_mnemo.thread` scans just that
4338        // thread's records (from the handle caches); one that can only
4339        // match non-fact kinds skips the ANN, which doesn't hold them.
4340        if let Some(thread) = filter_thread_pin(&req.filter) {
4341            let thread = thread.to_string();
4342            let entries = self.thread_entries(&thread)?;
4343            return self.score_entries(req, now, &entries);
4344        }
4345        let force_exact = force_exact || filter_bypasses_ann(&req.filter);
4346        let entries: Vec<CatalogEntry> = match (&self.ann, force_exact) {
4347            (Some(ann), false) => {
4348                let ids = ann.query(&req.query, req.n_probe, req.n_rerank);
4349                ids.iter()
4350                    .filter_map(|&id| self.index.get(id, &self.catalog))
4351                    .map(|i| self.catalog[i].clone())
4352                    .filter(|e| !e.deleted)
4353                    .collect()
4354            }
4355            _ => self
4356                .catalog
4357                .iter()
4358                .filter(|e| !e.deleted)
4359                .cloned()
4360                .collect(),
4361        };
4362
4363        self.score_entries(req, now, &entries)
4364    }
4365
4366    /// Score one candidate set for [`Mnemo::score_once`].
4367    fn score_entries(
4368        &mut self,
4369        req: &RecallRequest,
4370        now: i64,
4371        entries: &[CatalogEntry],
4372    ) -> Result<Vec<RecallResult>> {
4373        let mut scored: Vec<RecallResult> = Vec::with_capacity(entries.len());
4374        for e in entries {
4375            // Phase 12.1: drop superseded records.
4376            //
4377            // Fast path: if the lazy `superseded` set has already been
4378            // built by a prior keyed op on this handle, check it before
4379            // decrypt — O(1), no AES work. On a fresh read-only handle
4380            // it is typically `None`, so we fall through to the slow
4381            // path (one `_mnemo.superseded_by` lookup per surviving
4382            // record, O(k) over user metadata), and never trigger the
4383            // full-file scan that `ensure_key_caches()` would do.
4384            if self.superseded.as_ref().is_some_and(|s| s.contains(&e.id)) {
4385                continue;
4386            }
4387            let m = self.read_memory(e)?;
4388            if m.metadata.contains_key(MNEMO_META_SUPERSEDED_BY) {
4389                continue;
4390            }
4391            if m.is_expired(now) {
4392                continue;
4393            }
4394            if let Some(types) = &req.memory_types {
4395                if !types.contains(&m.memory_type) {
4396                    continue;
4397                }
4398            }
4399            if let Some(agent) = &req.agent_id {
4400                let visible = m.agent_id == *agent || m.scope == Scope::Shared;
4401                if !visible {
4402                    continue;
4403                }
4404            }
4405            // Phase 12.6: pre-scoring filter (session_id + metadata
4406            // conditions). Cheap when `req.filter.is_empty()`; the
4407            // filter struct's `matches` is a short-circuit AND.
4408            if !req.filter.is_empty() && !req.filter.matches(&m) {
4409                continue;
4410            }
4411            // Phase 12.7: default-exclude the thread layers from fact
4412            // recall: archive turns (12.7a), essence items (12.7b; read
4413            // by scope through the render, not by similarity) and
4414            // thread-link markers (12.7c; they carry an all-equal
4415            // sentinel vector and would surface as junk hits). A caller
4416            // that wants them opts in by placing any `_mnemo.*` term in
4417            // the filter (equality, `$in`, `$exists`, or a range on
4418            // `_mnemo.thread` / `_mnemo.kind` / `_mnemo.seq`).
4419            if !req.filter.mentions_mnemo_meta()
4420                && matches!(
4421                    m.metadata.get(MNEMO_META_KIND).and_then(|v| v.as_str()),
4422                    Some(MNEMO_META_KIND_TURN)
4423                        | Some(MNEMO_META_KIND_ESSENCE)
4424                        | Some(MNEMO_META_KIND_THREAD_LINK)
4425                )
4426            {
4427                continue;
4428            }
4429            let sim = memory::similarity(req.metric, &req.query, &m.vector);
4430            let age_secs = (now - m.created_at) as f32;
4431            let score = req
4432                .weights
4433                .score(sim, age_secs, m.importance, m.access_count);
4434            scored.push(RecallResult {
4435                memory: m,
4436                score,
4437                similarity: sim,
4438            });
4439        }
4440        Ok(scored)
4441    }
4442
4443    /// Phase-5 multi-signal recall under the v0.7.0 scoring-v2 formula
4444    /// (see [`crate::ScoreWeights`]). Thin wrapper over
4445    /// [`Mnemo::recall_budgeted`] that drops the `truncated` flag;
4446    /// use the budgeted form if you need to know whether
4447    /// [`RecallRequest::max_bytes`] cut anything.
4448    pub fn recall(&mut self, req: &RecallRequest) -> Result<Vec<RecallResult>> {
4449        Ok(self.recall_budgeted(req)?.hits)
4450    }
4451
4452    /// Phase 12.8 budgeted recall: scoring v2 (see [`crate::ScoreWeights`])
4453    /// with an optional UTF-8-content byte cap. Ranks by
4454    /// `sim + max(sim, 0) · (β·recency + γ·(importance − 0.5) + δ·tanh(uses/5))`,
4455    /// with type / agent / supersession / TTL filters and the 12.6
4456    /// `Filter` applied before the scorer. Updates `accessed_at` /
4457    /// `access_count` on the returned memories (persisted on the next
4458    /// `flush`).
4459    ///
4460    /// When an ANN index has been built ([`Mnemo::build_index`]),
4461    /// recall runs the tiered IVF→PQ→rerank pipeline. If the filters
4462    /// (type / agent / supersession / TTL / 12.6) reduce the ANN
4463    /// shortlist below `top_k`, Phase 12.1's short-results-on-indexed-
4464    /// path fix kicks in: the query is retried with `n_probe` and
4465    /// `n_rerank` doubled, and if that would already touch half or
4466    /// more of the live non-superseded records, the ANN pass is
4467    /// skipped for an exact scan. Without an index, recall scores
4468    /// every live non-superseded memory.
4469    ///
4470    /// Phase 12.8: after `top_k` is applied, if
4471    /// [`RecallRequest::max_bytes`] is set, hits are dropped in rank
4472    /// order once the running sum of `content` UTF-8 bytes would
4473    /// exceed the budget. The top-1 hit is always kept even if it
4474    /// alone exceeds the budget. [`RecallResponse::truncated`] is
4475    /// `true` iff at least one hit was dropped by the budget.
4476    pub fn recall_budgeted(&mut self, req: &RecallRequest) -> Result<RecallResponse> {
4477        if req.query.len() != self.dimensions {
4478            return Err(MnemoError::DimensionMismatch {
4479                expected: self.dimensions,
4480                got: req.query.len(),
4481            });
4482        }
4483        // PERF-1 step 3: the ANN index loads on the first recall.
4484        self.ensure_ann()?;
4485        // Access-tracking bumps `accessed_at` / `access_count` in the
4486        // catalog, which is a mutation. On a read-only handle the caller
4487        // must pass `track_access(false)` (via CLI `--no-track`).
4488        if self.read_only && req.track_access {
4489            return Err(MnemoError::ReadOnly);
4490        }
4491        // Phase 12.1 **fix** (lane A v0.8.0 REVIEW): do NOT call
4492        // `ensure_key_caches()` here. A fresh read-only handle that
4493        // only ever recalls would otherwise pay a full-file decrypt
4494        // on its first query to build the lazy superseded set. Instead
4495        // `score_once` checks `_mnemo.superseded_by` per-record after
4496        // decrypt (slow path) and consults the set only when it has
4497        // already been built by a prior keyed op on this handle (fast
4498        // path). `live_non_superseded` is not knowable without a
4499        // decrypt pass either, so we approximate by `live_count`; the
4500        // exact-scan gate uses this as a conservative upper bound,
4501        // which errs toward widening (cheaper) rather than skipping
4502        // straight to a full scan.
4503        let live_count = self.catalog.iter().filter(|e| !e.deleted).count();
4504        let live_non_superseded = live_count;
4505        let now = memory::now_secs();
4506
4507        // Scoring v2 (shipped v0.7.0). Single-pass filter + score:
4508        //   * `age_secs = now - m.created_at`, so a recall doesn't bump
4509        //     its own results' recency on later queries.
4510        //   * `ScoreWeights::score` applies the similarity-gated,
4511        //     importance-centered, saturating-usage formula documented
4512        //     on `ScoreWeights`.
4513        //
4514        // Phase 12.1 + 12.6 short-results-on-indexed-path fix: when the
4515        // ANN shortlist is narrower than the live count, the filters
4516        // (type / agent / supersession / TTL / 12.6) can reduce the
4517        // result below `top_k`. If that happens, re-query the index
4518        // with `n_probe` / `n_rerank` doubled from their *effective*
4519        // base; if that would already cover ≥50% of the live set,
4520        // skip the index and do an exact scan.
4521        //
4522        // Lane A v0.8.0 REVIEW fix: widen from
4523        // `IvfPqIndex::effective_defaults()` when the caller didn't
4524        // pin the knobs. The pre-fix code only doubled when `req.n_*`
4525        // was already `Some(_)`, which skipped the widen step for the
4526        // default (unpinned) recall path and jumped straight to an
4527        // exact scan on moderately selective filters.
4528        let mut scored = self.score_once(req, now, false)?;
4529        let mut used_exact_scan = self.ann.is_none() || filter_bypasses_ann(&req.filter);
4530        if !used_exact_scan && scored.len() < req.top_k && live_non_superseded > req.top_k {
4531            if let Some(ann) = &self.ann {
4532                let (default_probe, default_rerank) = ann.effective_defaults();
4533                let base_probe = req.n_probe.unwrap_or(default_probe);
4534                let base_rerank = req.n_rerank.unwrap_or(default_rerank);
4535                let widen_probe = base_probe.saturating_mul(2);
4536                let widen_rerank = base_rerank.saturating_mul(2);
4537                // Attempt 2: widen once, 2× from the effective base.
4538                // Only worth another ANN pass if either knob actually
4539                // grew (saturating_mul on usize::MAX would cap).
4540                if widen_probe > base_probe || widen_rerank > base_rerank {
4541                    let widened = RecallRequest {
4542                        n_probe: Some(widen_probe),
4543                        n_rerank: Some(widen_rerank),
4544                        ..req.clone()
4545                    };
4546                    scored = self.score_once(&widened, now, false)?;
4547                }
4548                // Attempt 3: if still short after widening, pay the
4549                // exact scan — it's the only path that guarantees
4550                // `top_k` correct hits when the filter is more
4551                // selective than `widen_rerank / live_count`. Before
4552                // the fix this was gated on `widened_rerank * 2 >=
4553                // live_non_superseded`, which silently returned fewer
4554                // than `top_k` hits on selective filters that fell in
4555                // the gap. Exact scan is bounded (one decrypt per
4556                // live record); the previous widen is an optimization.
4557                if scored.len() < req.top_k {
4558                    scored = self.score_once(req, now, true)?;
4559                    used_exact_scan = true;
4560                }
4561            }
4562        }
4563        scored.sort_by(|a, b| b.score.total_cmp(&a.score));
4564        scored.truncate(req.top_k);
4565
4566        // Update access statistics for everything we surfaced — but only
4567        // mutate the catalog entry, not the full record on disk. Pre-v5
4568        // this loop called `self.put(r.memory.clone())`, which serialized
4569        // and rewrote every result (vector included) to fresh pages,
4570        // turning a top-K recall into K full record rewrites. Now the
4571        // catalog is the source of truth for these two fields and the
4572        // record body stays untouched until the next real edit.
4573        if req.track_access {
4574            for r in &mut scored {
4575                if let Some(idx) = self.index.get(r.memory.id.0, &self.catalog) {
4576                    let entry = &mut self.catalog[idx];
4577                    entry.accessed_at = now;
4578                    entry.access_count = entry.access_count.saturating_add(1);
4579                    // Propagate the just-updated values onto the returned
4580                    // Memory so the caller sees the post-recall state.
4581                    r.memory.accessed_at = entry.accessed_at;
4582                    r.memory.access_count = entry.access_count;
4583                }
4584            }
4585            // Mark the catalog dirty so the next flush persists these
4586            // bumps. The records themselves are not dirty.
4587            if !scored.is_empty() {
4588                self.dirty_catalog = true;
4589            }
4590        }
4591
4592        // Phase 12.8: apply the `max_bytes` budget *after* tracking has
4593        // seen the full top-k list. Access stats record what the scorer
4594        // chose; the byte budget shouldn't suppress the signal.
4595        let truncated = if let Some(budget) = req.max_bytes {
4596            let orig_len = scored.len();
4597            let mut running = 0usize;
4598            let mut cut_at: Option<usize> = None;
4599            for (i, r) in scored.iter().enumerate() {
4600                let w = r.memory.content.len();
4601                if i == 0 {
4602                    // Always keep the top hit, even if it's over budget
4603                    // on its own. Avoid "I asked for top_k results and
4604                    // got zero" when a single memory exceeds the cap.
4605                    running += w;
4606                    continue;
4607                }
4608                if running.saturating_add(w) > budget {
4609                    cut_at = Some(i);
4610                    break;
4611                }
4612                running += w;
4613            }
4614            if let Some(idx) = cut_at {
4615                scored.truncate(idx);
4616                scored.len() < orig_len
4617            } else {
4618                false
4619            }
4620        } else {
4621            false
4622        };
4623
4624        Ok(RecallResponse {
4625            hits: scored,
4626            truncated,
4627            used_exact_scan,
4628        })
4629    }
4630
4631    // --- approximate index ------------------------------------------------
4632
4633    /// Build an IVF+PQ approximate-nearest-neighbour index over every live
4634    /// memory, using default tuning. After this, [`Mnemo::recall`] runs the
4635    /// tiered pipeline instead of an exact scan. Persisted on the next
4636    /// `flush`. Returns a snapshot of the index shape.
4637    pub fn build_index(&mut self) -> Result<IndexInfo> {
4638        self.build_index_with(IndexConfig::default())
4639    }
4640
4641    /// Build the ANN index with explicit tuning.
4642    pub fn build_index_with(&mut self, cfg: IndexConfig) -> Result<IndexInfo> {
4643        if self.read_only {
4644            return Err(MnemoError::ReadOnly);
4645        }
4646        // R-004.2: facts only; turns, essence items and thread links stay
4647        // out of the ANN.
4648        let mems: Vec<Memory> = self.memories()?.into_iter().filter(is_ann_kind).collect();
4649        if mems.is_empty() {
4650            return Err(MnemoError::Invalid(
4651                "cannot build an index over an empty database".into(),
4652            ));
4653        }
4654        let items: Vec<(u128, &[f32])> =
4655            mems.iter().map(|m| (m.id.0, m.vector.as_slice())).collect();
4656        let idx = IvfPqIndex::build(self.dimensions, &items, cfg)?;
4657        let info = idx.info();
4658        self.ann = Some(idx);
4659        self.ann_unloaded = false;
4660        self.dirty_index = true;
4661        self.index_persist.forced = true;
4662        Ok(info)
4663    }
4664
4665    /// Rebuild the ANN index from scratch — re-clusters centroids and retrains
4666    /// the PQ codebook, undoing the cluster drift that accumulates as memories
4667    /// are inserted against fixed centroids.
4668    pub fn rebuild_index(&mut self) -> Result<IndexInfo> {
4669        // ANN-1.1: re-derive every parameter (partitions, subspaces,
4670        // n_probe, n_rerank) for the current corpus size by passing
4671        // `IndexConfig::default()`. Previously this freezed the
4672        // stored `(n_probe, n_rerank)` from the current index,
4673        // defeating the point of rebuilding on a grown corpus. If
4674        // you really want to pin specific values, call
4675        // `build_index_with(IndexConfig { … })` directly.
4676        self.build_index_with(IndexConfig::default())
4677    }
4678
4679    /// Drop the ANN index; recall reverts to exact scans. Persisted on flush.
4680    /// No-op on a read-only handle (signature is infallible — callers of
4681    /// `build_index` / `flush` on a read-only handle already get a typed
4682    /// `ReadOnly` error, which is the enforceable boundary).
4683    pub fn drop_index(&mut self) {
4684        if self.read_only {
4685            return;
4686        }
4687        if self.ann.is_some() || self.ann_unloaded {
4688            self.ann = None;
4689            self.ann_unloaded = false;
4690            self.dirty_index = true;
4691            self.index_persist.forced = true;
4692        }
4693    }
4694
4695    /// PERF-1 (R-004.1): bring a just-loaded ANN index up to date with the
4696    /// catalog. Called wherever an index run is loaded: on first use after
4697    /// `open` / `open_read_only` (`ensure_ann`, PERF-1 step 3) and by
4698    /// `restore_to`, with the first page of that run.
4699    ///
4700    /// `flush` persists the index only every so often, so that run can lag
4701    /// the catalog. It does two things:
4702    ///
4703    /// 1. **Prune** every indexed id the catalog no longer has live
4704    ///    (missing or deleted).
4705    /// 2. **Re-add** every live record written after the index run. Pages
4706    ///    are allocated monotonically in v8, and every write, including an
4707    ///    in-place `update` or `essence_update` (they rewrite the record to
4708    ///    fresh pages), lands at a page `> index_start`. Those records are
4709    ///    read and re-assigned. A record of a non-fact kind (turn, essence,
4710    ///    thread link) is removed instead: those never belong in the ANN.
4711    ///
4712    /// The work is bounded by [`INDEX_PERSIST_EVERY_CHANGES`] in the common
4713    /// case, since a writer persists the index at least that often.
4714    fn reconcile_ann_with_catalog(&mut self, index_start: u64) -> Result<()> {
4715        if self.ann.is_none() {
4716            return Ok(());
4717        }
4718        let mut changed = 0usize;
4719        // 1. Prune ids that are no longer live.
4720        let stale: Vec<u128> = match &self.ann {
4721            Some(ann) => ann
4722                .ids()
4723                .filter(|&id| !matches!(self.index.get(id, &self.catalog), Some(i) if !self.catalog[i].deleted))
4724                .collect(),
4725            None => Vec::new(),
4726        };
4727        if let Some(ann) = &mut self.ann {
4728            for id in &stale {
4729                ann.remove(*id);
4730                changed += 1;
4731            }
4732        }
4733        // 2. Re-add records written after the index run.
4734        let pending: Vec<CatalogEntry> = self
4735            .catalog
4736            .iter()
4737            .filter(|e| !e.deleted && e.start_page > index_start)
4738            .cloned()
4739            .collect();
4740        for e in &pending {
4741            let m = self.read_memory(e)?;
4742            if let Some(ann) = &mut self.ann {
4743                if is_ann_kind(&m) {
4744                    ann.add(e.id, &m.vector);
4745                } else {
4746                    ann.remove(e.id);
4747                }
4748                changed += 1;
4749            }
4750        }
4751        if changed > 0 {
4752            self.dirty_index = true;
4753            self.index_persist.changes += changed;
4754        }
4755        Ok(())
4756    }
4757
4758    /// Whether the database has an ANN index (loaded, or persisted and
4759    /// loading on first use).
4760    pub fn has_index(&self) -> bool {
4761        self.ann.is_some() || self.ann_unloaded
4762    }
4763
4764    /// PERF-1 step 3: load the persisted ANN index on first use, then bring
4765    /// it up to date with the catalog (`reconcile_ann_with_catalog`) and,
4766    /// when the handle caches are already built, prune non-fact ids an
4767    /// older persisted index may hold.
4768    fn ensure_ann(&mut self) -> Result<()> {
4769        if !self.ann_unloaded {
4770            return Ok(());
4771        }
4772        self.ann_unloaded = false;
4773        self.ann = load_index(
4774            &mut self.pager,
4775            self.header.index_start,
4776            self.header.index_pages,
4777            self.header.index_len,
4778            self.dimensions,
4779        )?;
4780        let index_start = self.header.index_start;
4781        self.reconcile_ann_with_catalog(index_start)?;
4782        let non_facts: Vec<u128> = match &self.layers {
4783            Some(l) => l
4784                .essence
4785                .iter()
4786                .copied()
4787                .chain(l.threads.values().flat_map(|t| t.records.iter().copied()))
4788                .collect(),
4789            None => Vec::new(),
4790        };
4791        if let Some(ann) = &mut self.ann {
4792            let mut pruned = 0usize;
4793            for id in non_facts {
4794                if ann.contains(id) {
4795                    ann.remove(id);
4796                    pruned += 1;
4797                }
4798            }
4799            if pruned > 0 {
4800                self.dirty_index = true;
4801                self.index_persist.changes += pruned;
4802            }
4803        }
4804        Ok(())
4805    }
4806
4807    // --- durability & maintenance ----------------------------------------
4808
4809    /// Seal a serialized buffer into a fresh run of encrypted home pages and
4810    /// append a WAL frame for each. Returns the run's `(start_page, pages)`.
4811    /// The pages are *not* written to their home locations here — they go to
4812    /// the WAL and are folded in later by [`Mnemo::checkpoint`].
4813    fn seal_run(&mut self, bytes: &[u8], frames: &mut Vec<wal::Frame>) -> Result<(u64, u32)> {
4814        let pc = bytes.len().div_ceil(PAYLOAD).max(1);
4815        let start = self.header.next_page;
4816        self.header.next_page += pc as u64;
4817        for i in 0..pc {
4818            let lo = i * PAYLOAD;
4819            let hi = ((i + 1) * PAYLOAD).min(bytes.len());
4820            let page_no = start + i as u64;
4821            let sealed = self.pager.seal_page(page_no, &bytes[lo..hi])?;
4822            frames.push((page_no, sealed.to_vec()));
4823        }
4824        Ok((start, pc as u32))
4825    }
4826
4827    /// Grow the WAL region if `frame_pages` worth of frames would not fit.
4828    /// Called at the top of `flush`, where the WAL is always spent (every
4829    /// flush ends with a checkpoint), so relocating it cannot strand a
4830    /// committed transaction.
4831    fn ensure_wal_capacity(&mut self, frame_pages: usize) -> Result<()> {
4832        let need = wal::txn_byte_len(frame_pages, PAGE_SIZE);
4833        if need <= self.header.wal_pages * PAGE_SIZE as u64 {
4834            return Ok(());
4835        }
4836        // Allocate a new, larger region at the file tail (~1.5x headroom).
4837        let req = need.div_ceil(PAGE_SIZE as u64);
4838        let new_pages = (req + req / 2 + 4).max(DEFAULT_WAL_PAGES);
4839        let old_wal_start = self.header.wal_start;
4840        let new_start = self.header.next_page;
4841        self.header.next_page += new_pages;
4842        self.header.wal_start = new_start;
4843        self.header.wal_pages = new_pages;
4844        self.header.wal_seq += 1;
4845        // DUR-1 (R-002.1): WAL-log the header change into the *new* region
4846        // first, checkpoint page 0, then invalidate the old region. A crash
4847        // anywhere here either keeps the old state (if page 0 wasn't yet
4848        // overwritten) or recovers to the new state via the torn-header
4849        // scan in `Mnemo::open`, which prefers the highest-txn-id valid
4850        // committed header frame.
4851        self.relocate_header_through_wal(old_wal_start)
4852    }
4853
4854    /// Shared helper for every post-`create` page-0 commit: WAL-log the
4855    /// new header into the current WAL region, checkpoint page 0, and
4856    /// invalidate `old_wal_start` (the previous region, or the current
4857    /// region when it hasn't moved). The caller is expected to have
4858    /// mutated `self.header` to its target values *including a bumped
4859    /// `wal_seq`* (one more than the last on-disk `wal_seq`); this
4860    /// routine applies the seal, writes the frame + commit, folds page 0,
4861    /// then zeroes the old region's first frame header.
4862    ///
4863    /// If `old_wal_start == self.header.wal_start`, the invalidation is
4864    /// skipped — the region didn't move, so overwriting its first frame
4865    /// would destroy the just-committed one.
4866    fn relocate_header_through_wal(&mut self, old_wal_start: u64) -> Result<()> {
4867        self.header.write_counter = self.pager.write_counter;
4868        let mut page = self.header.to_page();
4869        self.header.apply_seal(&mut page, self.pager.dek())?;
4870        let txn_id = self.header.wal_seq;
4871        let frames = vec![(0u64, page.to_vec())];
4872        wal::commit(
4873            self.pager.file_mut(),
4874            self.header.wal_start,
4875            self.header.wal_pages,
4876            txn_id,
4877            &frames,
4878        )?;
4879        self.pager.write_raw(0, &page)?;
4880        self.pager.sync()?;
4881        if old_wal_start != self.header.wal_start {
4882            wal::invalidate_region(self.pager.file_mut(), old_wal_start)?;
4883        }
4884        Ok(())
4885    }
4886
4887    /// Fold a committed transaction's WAL frames into their home pages.
4888    fn checkpoint(&mut self, frames: &[wal::Frame]) -> Result<()> {
4889        for (page_no, bytes) in frames {
4890            let mut img = [0u8; PAGE_SIZE];
4891            img.copy_from_slice(bytes);
4892            self.pager.write_sealed(*page_no, &img)?;
4893        }
4894        self.pager.sync()?;
4895        Ok(())
4896    }
4897
4898    /// Persist all pending changes as one **write-ahead-logged transaction**.
4899    ///
4900    /// Sequence:
4901    ///
4902    /// 1. **Prepare.** Serialize the control plane (catalog, ANN index),
4903    ///    grow the WAL if needed, and **lease** counter and page slots —
4904    ///    bump `header.write_counter` and `header.next_page` by upper
4905    ///    bounds on this transaction's consumption and persist that
4906    ///    *leased* header to disk before any encrypted page is written.
4907    /// 2. **Data pages.** [`Pager::flush`] writes dirty record pages
4908    ///    copy-on-write under fresh nonces and fsyncs them. The orphan-page
4909    ///    nonce-reuse window (see [Phase 1.1 of the improvement plan])
4910    ///    is closed because the leased header on disk already records a
4911    ///    `write_counter` past anything this step can produce.
4912    /// 3. **Control plane.** Seal the new catalog / ANN index / snapshot
4913    ///    manifest into fresh home page runs.
4914    /// 4. **Commit.** Log all of step 3's frames plus the final header
4915    ///    frame into the WAL and fsync — the durability point.
4916    /// 5. **Checkpoint.** Fold the WAL into the home pages.
4917    ///
4918    /// A crash before step 4 leaves the previous state intact. A crash
4919    /// after step 4 is repaired by [`Mnemo::open`] replaying the WAL.
4920    /// Safe to call repeatedly.
4921    pub fn flush(&mut self) -> Result<()> {
4922        if self.read_only {
4923            return Err(MnemoError::ReadOnly);
4924        }
4925        // Snapshot the thread-local fsync counter. Every `sync_all` the
4926        // pager / wal / checkpoint path makes between here and the end
4927        // of this function bumps the same counter; `delta` lands on
4928        // `self.fsyncs_last_flush` so `Stats::fsyncs_last_flush` can
4929        // report it. The planner wants this to decide whether cutting
4930        // fsync count is a viable write-latency win before we touch
4931        // the write path's durability model.
4932        let fsync_before = crate::fsync_count();
4933        // PERF-1 step 3: while the index is unloaded, records written since
4934        // its run pile up for the next load to re-assign. Once there are
4935        // INDEX_PERSIST_EVERY_CHANGES of them, load it now so this flush
4936        // persists a current index and keeps that work bounded.
4937        if self.ann_unloaded && self.dirty_catalog {
4938            let index_start = self.header.index_start;
4939            let pending = self
4940                .catalog
4941                .iter()
4942                .filter(|e| !e.deleted && e.start_page > index_start)
4943                .count();
4944            if pending >= INDEX_PERSIST_EVERY_CHANGES {
4945                self.ensure_ann()?;
4946            }
4947        }
4948        // PERF-1: persist the (derived) ANN index only when forced or due.
4949        // Otherwise the header keeps pointing at the previous index run, and
4950        // records written since then are re-assigned when it's next loaded
4951        // (`reconcile_ann_with_catalog`). `ann == None` with `dirty_index` set is a
4952        // dropped index, which is persisted at once.
4953        let persist_index = self.dirty_index
4954            && (self.index_persist.forced
4955                || self.ann.is_none()
4956                || self.index_persist.commits + 1 >= INDEX_PERSIST_EVERY_COMMITS
4957                || self.index_persist.changes >= INDEX_PERSIST_EVERY_CHANGES);
4958        if !self.dirty_catalog && !persist_index && !self.dirty_cache {
4959            // No control-plane changes pending. Because every `put` that
4960            // dirties a data page also dirties the catalog, and every
4961            // `cache_put` that dirties a data page also dirties
4962            // `dirty_cache`, there can't be dirty data pages either —
4963            // nothing to do.
4964            self.fsyncs_last_flush = 0;
4965            return Ok(());
4966        }
4967
4968        // FIX-1 R-001.8 part 3: snapshot the memory/index-dirty flags
4969        // before step 3b decides whether to push a restore point.
4970        // Only flushes that change the memory catalog or ANN index
4971        // need a snapshot — a cache-only flush (`dirty_cache` set,
4972        // `dirty_catalog` + `dirty_index` both false) must not push,
4973        // because snapshots don't capture cache state and every push
4974        // counts against `max_snapshots` (default 256), so a cache-
4975        // heavy workload would otherwise evict every real memory
4976        // restore point within ~256 cache writes.
4977        let is_memory_or_index_flush = self.dirty_catalog || persist_index;
4978
4979        // 1. Serialize control plane, ensure WAL capacity, lease counter +
4980        //    page slots, and persist the leased header *before* any data
4981        //    page hits the disk.
4982        let ctx = self.prepare_for_flush(persist_index)?;
4983
4984        // 2. Record (vector) data pages: copy-on-write to fresh pages, fsynced.
4985        //    Safe: the lease guarantees that even if we crash here, reopen
4986        //    will see a `write_counter` past every nonce this step uses.
4987        self.pager.flush()?;
4988
4989        // 3. Seal the catalog / index control pages into fresh home runs.
4990        let mut frames: Vec<wal::Frame> = Vec::new();
4991        if let Some(bytes) = &ctx.cat_bytes {
4992            let (start, pages) = self.seal_run(bytes, &mut frames)?;
4993            self.header.catalog_start = start;
4994            self.header.catalog_pages = pages as u64;
4995            self.header.catalog_len = bytes.len() as u64;
4996        }
4997        if persist_index {
4998            match &ctx.idx_bytes {
4999                Some(bytes) => {
5000                    let (start, pages) = self.seal_run(bytes, &mut frames)?;
5001                    self.header.index_start = start;
5002                    self.header.index_pages = pages as u64;
5003                    self.header.index_len = bytes.len() as u64;
5004                }
5005                None => {
5006                    self.header.index_start = 0;
5007                    self.header.index_pages = 0;
5008                    self.header.index_len = 0;
5009                }
5010            }
5011        }
5012        // v8+ result-cache directory (Phase 10.1). Seal the serialized
5013        // directory into a fresh page run and update the pointer in the
5014        // header. The seal AAD already covers cache_start/pages/len for
5015        // v8+ (see `header_seal_aad`), so tampering with the pointer
5016        // trips the header seal on the next open.
5017        if self.dirty_cache {
5018            match &ctx.cache_bytes {
5019                Some(bytes) if !self.cache_dir.is_empty() => {
5020                    let (start, pages) = self.seal_run(bytes, &mut frames)?;
5021                    self.header.cache_start = start;
5022                    self.header.cache_pages = pages as u64;
5023                    self.header.cache_len = bytes.len() as u64;
5024                }
5025                _ => {
5026                    // Either the directory serialized to nothing (all
5027                    // entries deleted after tombstone-only migration) or
5028                    // it's empty — collapse the pointer to zero.
5029                    self.header.cache_start = 0;
5030                    self.header.cache_pages = 0;
5031                    self.header.cache_len = 0;
5032                }
5033            }
5034        }
5035
5036        // 3b. Record this transaction as a restorable snapshot. The manifest
5037        //     update is staged in a local and adopted only once the commit
5038        //     below succeeds, so a failed flush leaves no phantom entry.
5039        //
5040        //     The cap (`self.max_snapshots`) is applied to the staged local
5041        //     before the new entry lands, so after appending the manifest
5042        //     holds at most `max_snapshots` entries — the most-recent N.
5043        //     `max_snapshots == 0` disables the cap. Pages referenced only
5044        //     by pruned snapshots stay on disk until `compact_file`.
5045        //
5046        //     FIX-1 R-001.8 part 3: skip the snapshot push on a
5047        //     cache-only flush. Snapshots don't capture the cache
5048        //     anyway, and every push counts against `max_snapshots`
5049        //     (default 256) + rewrites the manifest run — so a
5050        //     cache-heavy workload would otherwise evict every real
5051        //     memory restore point within ~256 cache writes. On a
5052        //     cache-only flush we reuse the current manifest
5053        //     verbatim; the manifest run already lives at
5054        //     `self.header.manifest_{start,pages,len}` (set by the
5055        //     last memory-touching flush, or at `create` time).
5056        let live = self.len() as u64;
5057        let txn_id = self.header.wal_seq + 1;
5058        let mut manifest = self.manifest.clone();
5059        if is_memory_or_index_flush {
5060            if self.max_snapshots > 0 {
5061                // Drop the oldest entries so there's room for the new one.
5062                while manifest.len() >= self.max_snapshots {
5063                    manifest.remove(0);
5064                }
5065            }
5066            manifest.push(Snapshot {
5067                txn_id,
5068                created_at: memory::now_secs(),
5069                catalog_start: self.header.catalog_start,
5070                catalog_pages: self.header.catalog_pages,
5071                catalog_len: self.header.catalog_len,
5072                index_start: self.header.index_start,
5073                index_pages: self.header.index_pages,
5074                index_len: self.header.index_len,
5075                memory_count: live,
5076            });
5077            let man_bytes =
5078                rmp_serde::to_vec(&manifest).map_err(|e| MnemoError::Serialize(e.to_string()))?;
5079            let (m_start, m_pages) = self.seal_run(&man_bytes, &mut frames)?;
5080            self.header.manifest_start = m_start;
5081            self.header.manifest_pages = m_pages as u64;
5082            self.header.manifest_len = man_bytes.len() as u64;
5083        }
5084
5085        // 3c. The header is the transaction's final frame; stamp the new id
5086        //     and the *actual* (post-flush) write counter, which is <= the
5087        //     leased value persisted in `prepare_for_flush`. This is the
5088        //     value that ultimately replaces the leased header on checkpoint.
5089        self.header.vector_count = live;
5090        self.header.write_counter = self.pager.write_counter;
5091        self.header.wal_seq = txn_id;
5092        let mut hpage = self.header.to_page();
5093        self.header.apply_seal(&mut hpage, self.pager.dek())?;
5094        frames.push((0, hpage.to_vec()));
5095
5096        // 4. COMMIT — log the transaction and fsync. Nothing at a home page
5097        //    other than the leased header has changed yet; this single
5098        //    fsync is what makes the rest durable.
5099        let (wal_start, wal_pages) = (self.header.wal_start, self.header.wal_pages);
5100        wal::commit(self.pager.file_mut(), wal_start, wal_pages, txn_id, &frames)?;
5101
5102        // 5. Checkpoint — fold the WAL into the home pages.
5103        self.checkpoint(&frames)?;
5104
5105        self.manifest = manifest;
5106        self.dirty_catalog = false;
5107        if persist_index {
5108            self.dirty_index = false;
5109            self.index_persist = IndexPersist::default();
5110        } else if self.dirty_index {
5111            self.index_persist.commits += 1;
5112        }
5113        self.dirty_cache = false;
5114        // Batched policy: successful flush consumes the pending batch.
5115        self.cache_batch.reset();
5116        // Record how many fsyncs this flush issued, for Stats reporting.
5117        // Thread-local counter guarantees we see only this handle's own
5118        // sync_all calls between the snapshot above and this read.
5119        self.fsyncs_last_flush =
5120            (crate::fsync_count().wrapping_sub(fsync_before)).min(u32::MAX as u64) as u32;
5121        Ok(())
5122    }
5123
5124    /// Pre-flush prelude shared by [`Mnemo::flush`] and the
5125    /// `__crash_partial_flush_for_testing` hook.
5126    ///
5127    /// Serializes the control plane (catalog, ANN index, cache directory)
5128    /// and grows the WAL if needed. That's it: this used to persist a
5129    /// leased header too (SEC-1 obsoleted it; see DUR-1 R-002.1 note in
5130    /// the body for the full rationale), but after DUR-1 the only page-0
5131    /// writes outside `create` go through the WAL via
5132    /// [`Mnemo::relocate_header_through_wal`].
5133    ///
5134    /// The historical pre-stamping used to close the nonce-reuse window
5135    /// flagged in Phase 1.1 of the improvement plan: a crash anywhere
5136    /// between here and the WAL commit left a file whose on-disk header
5137    /// recorded a `write_counter` past every nonce the rest of this
5138    /// transaction could produce, so the next reopen started from a
5139    /// counter that guaranteed uniqueness against any orphan data pages
5140    /// still on disk. SEC-1's per-session random nonce prefix gives the
5141    /// same guarantee without needing a persisted counter, which is why
5142    /// this routine no longer writes page 0.
5143    fn prepare_for_flush(&mut self, persist_index: bool) -> Result<FlushPrelude> {
5144        // Serialize.
5145        let cat_bytes = if self.dirty_catalog {
5146            Some(
5147                rmp_serde::to_vec(&self.catalog)
5148                    .map_err(|e| MnemoError::Serialize(e.to_string()))?,
5149            )
5150        } else {
5151            None
5152        };
5153        let idx_bytes: Option<Vec<u8>> = if persist_index {
5154            match &self.ann {
5155                Some(ann) => {
5156                    Some(rmp_serde::to_vec(ann).map_err(|e| MnemoError::Serialize(e.to_string()))?)
5157                }
5158                None => None,
5159            }
5160        } else {
5161            None
5162        };
5163        // Serialize the result-cache directory when it's dirty. An
5164        // empty directory serializes to a small header + 0-element
5165        // array; still cheap enough that we don't special-case it.
5166        let cache_bytes: Option<Vec<u8>> = if self.dirty_cache {
5167            Some(
5168                rmp_serde::to_vec(&self.cache_dir)
5169                    .map_err(|e| MnemoError::Serialize(e.to_string()))?,
5170            )
5171        } else {
5172            None
5173        };
5174
5175        let cat_pc = cat_bytes
5176            .as_ref()
5177            .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
5178        let idx_pc = idx_bytes
5179            .as_ref()
5180            .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
5181        let cache_pc = cache_bytes
5182            .as_ref()
5183            .map_or(0, |b| b.len().div_ceil(PAYLOAD).max(1));
5184        // Upper bound on the manifest run: one extra entry, <=82 bytes each
5185        // (rmp_serde encodes a Snapshot as a 9-element fixarray of i64/u64,
5186        // worst case 81 bytes plus a 1-byte array header).
5187        let man_upper = 9 + (self.manifest.len() + 1) * 82;
5188        let man_pc_est = man_upper.div_ceil(PAYLOAD).max(1);
5189
5190        // Grow WAL if needed (DUR-1: relocation is WAL-logged and does
5191        // its own fsyncs). The cache directory rides in the same
5192        // transaction so its page count contributes to the reservation.
5193        self.ensure_wal_capacity(cat_pc + idx_pc + cache_pc + man_pc_est + 1)?;
5194
5195        // DUR-1 (R-002.1): the pre-flush "leased header" write that used
5196        // to live here is gone. It existed to close the Phase 1.1 nonce
5197        // reuse window (reopen must see a `write_counter` past any nonce
5198        // this flush could produce before data pages hit disk). SEC-1
5199        // superseded that: nonces now carry a per-session random prefix
5200        // and `write_counter` is no longer a nonce input, so the leased
5201        // write bought us nothing but an extra fsync per flush. Orphan
5202        // data pages past `header.next_page` on a crashed flush are
5203        // still safe — the next session's random prefix makes any later
5204        // write at the same page_no uniquely nonced regardless of
5205        // `write_counter`.
5206
5207        Ok(FlushPrelude {
5208            cat_bytes,
5209            idx_bytes,
5210            cache_bytes,
5211        })
5212    }
5213
5214    /// Flush and close. Equivalent to `flush()` on a read-write handle; on
5215    /// a read-only handle it's a no-op (no dirty state to persist). The
5216    /// file is released on drop, which also drops the OS advisory lock.
5217    pub fn close(&mut self) -> Result<()> {
5218        if self.read_only {
5219            return Ok(());
5220        }
5221        // PERF-1: closing persists a dirty ANN index too, so the next open
5222        // has nothing to re-assign.
5223        self.index_persist.forced = true;
5224        self.flush()
5225    }
5226
5227    /// PERF-1: flush, persisting the ANN index even if the
5228    /// [`INDEX_PERSIST_EVERY_COMMITS`] / [`INDEX_PERSIST_EVERY_CHANGES`]
5229    /// policy wouldn't yet. Plain [`Mnemo::flush`] writes only the catalog
5230    /// between persists; the index is derived, so a crash costs nothing but
5231    /// re-assigning the newer records on the next load.
5232    pub fn flush_index(&mut self) -> Result<()> {
5233        if self.read_only {
5234            return Err(MnemoError::ReadOnly);
5235        }
5236        self.index_persist.forced = true;
5237        self.flush()
5238    }
5239
5240    /// **TEST ONLY — do not use.** Reproduces a crash inside
5241    /// [`Mnemo::flush`] *after* the prelude has leased counter+page slots
5242    /// and persisted the leased header, but *before* the WAL is committed.
5243    /// Runs `prepare_for_flush` + `pager.flush` and returns — leaving
5244    /// data pages and a leased header on disk, with no commit frame.
5245    ///
5246    /// Used by `tests/integration.rs::nonce_unique_after_crashed_data_flush`
5247    /// to verify that this window (Phase 1.1 of the improvement plan) does
5248    /// not enable AES-GCM nonce reuse. Calling this in production code
5249    /// strands data pages; never expose it from a binding.
5250    #[doc(hidden)]
5251    pub fn __crash_partial_flush_for_testing(&mut self) -> Result<()> {
5252        if self.read_only {
5253            return Err(MnemoError::ReadOnly);
5254        }
5255        if !self.dirty_catalog && !self.dirty_index {
5256            return Ok(());
5257        }
5258        let _prelude = self.prepare_for_flush(self.dirty_index)?;
5259        self.pager.flush()
5260    }
5261
5262    /// Bound the in-memory page cache to `pages` decrypted pages.
5263    ///
5264    /// The cache holds decrypted page payloads to speed repeated reads. By
5265    /// default it is capped at 1024 pages (~8 MiB; 8192 before v0.10.0).
5266    /// Lower the cap to trade hit rate for a smaller footprint, or raise
5267    /// it for a hotter cache. The
5268    /// cap governs *clean* pages — pages with un-flushed writes are always
5269    /// retained until [`Mnemo::flush`], regardless of the cap.
5270    pub fn set_cache_capacity(&mut self, pages: usize) {
5271        self.pager.set_cache_capacity(pages);
5272    }
5273
5274    /// Page-cache occupancy: `(pages_cached, capacity)`. Renamed from
5275    /// `cache_stats` in v0.4.0: the result-cache API (Phase 10.1) took
5276    /// that name for its per-namespace [`CacheStats`] summary, and the
5277    /// page-cache pair is the older/less-user-facing of the two, so it
5278    /// gets the more-specific name.
5279    pub fn page_cache_stats(&self) -> (usize, usize) {
5280        self.pager.cache_stats()
5281    }
5282
5283    // === Result cache (Phase 10.1 + 10.3) ============================
5284
5285    /// Insert or overwrite a cache entry at `(namespace, key)`.
5286    ///
5287    /// The engine SHA-256-hashes `key` — callers pass the raw input
5288    /// string. Persists to disk on the next `flush()` under the current
5289    /// flush policy (see [`Mnemo::set_cache_flush_policy`]); in
5290    /// [`CacheFlushPolicy::Batched`] the engine may call `flush()`
5291    /// internally when a threshold trips.
5292    ///
5293    /// On overflow of the namespace's [`CacheBudget`] (default 10,000
5294    /// entries / 64 MiB) the LRU entry is tombstoned before the new
5295    /// insert. Tombstones are reclaimed by
5296    /// [`Mnemo::compact_file`], not the runtime flush.
5297    ///
5298    /// Returns [`MnemoError::ReadOnly`] on a read-only handle.
5299    pub fn cache_put(
5300        &mut self,
5301        namespace: &str,
5302        key: &str,
5303        value: &[u8],
5304        opts: CachePutOpts,
5305    ) -> Result<()> {
5306        if self.read_only {
5307            return Err(MnemoError::ReadOnly);
5308        }
5309        let key_hash = result_cache::hash_key(key);
5310        let key_preview = result_cache::key_preview(key);
5311        let now = memory::now_secs();
5312        let size_bytes = value.len() as u64;
5313
5314        let entry = CacheEntry {
5315            namespace: namespace.to_string(),
5316            key_hash,
5317            key_preview: key_preview.clone(),
5318            value: value.to_vec(),
5319            content_type: opts.content_type.clone(),
5320            created_at: now,
5321            accessed_at: now,
5322            access_count: 0,
5323            ttl_secs: opts.ttl_secs,
5324            cost_hint_ms: opts.cost_hint_ms,
5325            size_bytes,
5326            vector: None,
5327            model: None,
5328        };
5329        let bytes = rmp_serde::to_vec(&entry).map_err(|e| MnemoError::Serialize(e.to_string()))?;
5330        let (start_page, page_count) = self.write_record(&bytes)?;
5331
5332        // Tombstone any prior entry for this key. Its pages remain on
5333        // disk until `compact_file`, exactly like memory-record
5334        // overwrites — the tombstone just hides it from lookups.
5335        if let Some(prev_pos) = self.cache_idx.get(namespace, &key_hash) {
5336            self.cache_dir[prev_pos].deleted = true;
5337            self.cache_idx.remove(namespace, &key_hash);
5338        }
5339
5340        let dir_entry = CacheDirectoryEntry {
5341            namespace: namespace.to_string(),
5342            key_hash,
5343            start_page,
5344            page_count,
5345            len: bytes.len() as u32,
5346            deleted: false,
5347            accessed_at: now,
5348            access_count: 0,
5349            size_bytes,
5350            ttl_secs: opts.ttl_secs,
5351            created_at: now,
5352            vector: None,
5353            model: None,
5354        };
5355        let pos = self.cache_dir.len();
5356        self.cache_dir.push(dir_entry);
5357        self.cache_idx.insert(namespace, key_hash, pos);
5358
5359        // Enforce budget — evict LRU tombstones until the namespace
5360        // fits under both caps.
5361        self.evict_to_budget(namespace)?;
5362
5363        self.dirty_cache = true;
5364        self.cache_batch.record_dirty();
5365
5366        // Batched policy: auto-flush if a threshold tripped. Successful
5367        // `flush()` calls `cache_batch.reset()` for us.
5368        if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
5369            self.flush()?;
5370        }
5371        Ok(())
5372    }
5373
5374    /// Look up `(namespace, key)`. Returns `None` on miss (either the
5375    /// key was never inserted or the entry is TTL-expired). On hit
5376    /// updates catalog-side access stats (mirrors the recall v5-trick);
5377    /// on a read-only handle the stats update is silently skipped.
5378    ///
5379    /// A miss is recorded as a miss in this handle's counters even if
5380    /// the key was TTL-expired — from the caller's perspective the two
5381    /// are indistinguishable.
5382    pub fn cache_get(&mut self, namespace: &str, key: &str) -> Result<Option<CachedValue>> {
5383        // FIX-1 R-001.8 part 4: under `Batched`, honor `max_age` on
5384        // reads too. Pre-FIX-1 the age check only fired on `cache_put`,
5385        // so a workload that wrote a few entries and then switched to
5386        // pure reads could sit with its pending batch past the
5387        // nominal max_age indefinitely. `max_dirty` is already
5388        // maintained per-write; this adds the symmetric time-bound
5389        // check on the read path. Only runs on read-write handles —
5390        // flush() returns ReadOnly for shared-lock readers.
5391        if !self.read_only && self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
5392            self.flush()?;
5393        }
5394        let key_hash = result_cache::hash_key(key);
5395        let pos = match self.cache_idx.get(namespace, &key_hash) {
5396            Some(p) => p,
5397            None => {
5398                self.cache_counters.misses += 1;
5399                return Ok(None);
5400            }
5401        };
5402        let now = memory::now_secs();
5403        // Snapshot the directory entry fields we need before any
5404        // mutation — the borrow checker gets grumpy otherwise.
5405        let (start_page, page_count, len, ttl_secs, expired, created_at) = {
5406            let e = &self.cache_dir[pos];
5407            (
5408                e.start_page,
5409                e.page_count,
5410                e.len,
5411                e.ttl_secs,
5412                e.is_expired(now),
5413                e.created_at,
5414            )
5415        };
5416        if expired {
5417            // Treat expired as a miss for the caller. Don't tombstone
5418            // here — cache_purge does that. Just skip.
5419            self.cache_counters.misses += 1;
5420            return Ok(None);
5421        }
5422        // Decrypt and decode the record body.
5423        let mut buf = Vec::with_capacity(len as usize);
5424        for i in 0..page_count as u64 {
5425            buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
5426        }
5427        buf.truncate(len as usize);
5428        let entry: CacheEntry =
5429            rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
5430
5431        // Bump access stats catalog-only (v5 trick — no full record
5432        // rewrite on a hit). Skipped on read-only handles so a shared-lock
5433        // reader can't dirty the catalog.
5434        if !self.read_only {
5435            self.cache_dir[pos].accessed_at = now;
5436            self.cache_dir[pos].access_count = self.cache_dir[pos].access_count.saturating_add(1);
5437            self.dirty_cache = true;
5438        }
5439        self.cache_counters.hits += 1;
5440        // On a mutating handle: access_count reflects the just-bumped
5441        // value. On a read-only handle: the on-disk directory value is
5442        // unchanged, so returning it verbatim is honest (the caller's
5443        // reads did not, and cannot, alter the persisted counter).
5444        //
5445        // FIX-1 R-002.2 part 1: `accessed_at` must also come from the
5446        // directory on a read-only handle, not from `entry.accessed_at`
5447        // in the record body. The record body is written once at insert
5448        // time and never updated on a hit — every live `accessed_at`
5449        // update lands in the directory (see the `!self.read_only`
5450        // branch above). The pre-FIX-1 code returned
5451        // `entry.accessed_at`, which is really the entry's **creation
5452        // time** by then, so a read-only reader saw "last accessed =
5453        // when it was inserted" even after many successful hits by
5454        // concurrent writers.
5455        let access_count = self.cache_dir[pos].access_count;
5456        let accessed_at = if self.read_only {
5457            self.cache_dir[pos].accessed_at
5458        } else {
5459            now
5460        };
5461        Ok(Some(CachedValue {
5462            value: entry.value,
5463            content_type: entry.content_type,
5464            created_at,
5465            accessed_at,
5466            access_count,
5467            ttl_secs,
5468        }))
5469    }
5470
5471    /// Tombstone the entry at `(namespace, key)`. Silently succeeds if
5472    /// no such entry exists. Space is reclaimed by `compact_file`.
5473    pub fn cache_delete(&mut self, namespace: &str, key: &str) -> Result<()> {
5474        if self.read_only {
5475            return Err(MnemoError::ReadOnly);
5476        }
5477        let key_hash = result_cache::hash_key(key);
5478        if let Some(pos) = self.cache_idx.get(namespace, &key_hash) {
5479            self.cache_dir[pos].deleted = true;
5480            self.cache_idx.remove(namespace, &key_hash);
5481            self.dirty_cache = true;
5482            self.cache_batch.record_dirty();
5483            if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
5484                self.flush()?;
5485            }
5486        }
5487        Ok(())
5488    }
5489
5490    /// Purge cache entries. If `namespace` is `Some`, restrict to that
5491    /// namespace; otherwise purge across the whole cache. If
5492    /// `expired_only` is `true`, only TTL-expired entries are removed;
5493    /// otherwise every live entry in the target scope is tombstoned.
5494    /// Returns the number of entries newly tombstoned.
5495    pub fn cache_purge(&mut self, namespace: Option<&str>, expired_only: bool) -> Result<usize> {
5496        if self.read_only {
5497            return Err(MnemoError::ReadOnly);
5498        }
5499        let now = memory::now_secs();
5500        let mut purged = 0usize;
5501        // Iterate the directory by index; mark matching entries as
5502        // deleted and pull them out of the in-memory index.
5503        for pos in 0..self.cache_dir.len() {
5504            if self.cache_dir[pos].deleted {
5505                continue;
5506            }
5507            if let Some(ns) = namespace {
5508                if self.cache_dir[pos].namespace != ns {
5509                    continue;
5510                }
5511            }
5512            if expired_only && !self.cache_dir[pos].is_expired(now) {
5513                continue;
5514            }
5515            let key_hash = self.cache_dir[pos].key_hash;
5516            let ns = self.cache_dir[pos].namespace.clone();
5517            self.cache_dir[pos].deleted = true;
5518            self.cache_idx.remove(&ns, &key_hash);
5519            purged += 1;
5520        }
5521        if purged > 0 {
5522            self.dirty_cache = true;
5523            self.cache_batch.record_dirty();
5524            if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
5525                self.flush()?;
5526            }
5527        }
5528        Ok(purged)
5529    }
5530
5531    /// Statistics for `namespace` (or the whole cache if `None`).
5532    /// `hits`/`misses`/`evictions` are per-handle in-memory counters
5533    /// that reset on reopen; `entries`/`bytes` reflect the durable
5534    /// on-disk state (live entries only, tombstones excluded).
5535    pub fn cache_stats(&self, namespace: Option<&str>) -> CacheStats {
5536        let mut entries = 0usize;
5537        let mut bytes = 0u64;
5538        for e in &self.cache_dir {
5539            if e.deleted {
5540                continue;
5541            }
5542            if let Some(ns) = namespace {
5543                if e.namespace != ns {
5544                    continue;
5545                }
5546            }
5547            entries += 1;
5548            bytes += e.size_bytes;
5549        }
5550        let hits = self.cache_counters.hits;
5551        let misses = self.cache_counters.misses;
5552        let total = hits + misses;
5553        let hit_rate = if total > 0 {
5554            hits as f64 / total as f64
5555        } else {
5556            0.0
5557        };
5558        CacheStats {
5559            entries,
5560            bytes,
5561            hits,
5562            misses,
5563            hit_rate,
5564            evictions: self.cache_counters.evictions,
5565        }
5566    }
5567
5568    /// Set (or update) the eviction budget for `namespace`. Applies to
5569    /// subsequent `cache_put` calls; does not retroactively evict
5570    /// entries that already sit under a looser cap.
5571    pub fn set_cache_budget(&mut self, namespace: &str, budget: CacheBudget) {
5572        self.cache_budgets.insert(namespace.to_string(), budget);
5573    }
5574
5575    /// Set the per-handle cache flush policy. Defaults to
5576    /// [`CacheFlushPolicy::Strict`]. Memory writes are never batched;
5577    /// this affects only cache mutations.
5578    pub fn set_cache_flush_policy(&mut self, policy: CacheFlushPolicy) {
5579        self.cache_flush_policy = policy;
5580        // Switching mid-run: if we're moving to Strict and the batch is
5581        // non-empty, the pending count still counts toward dirty_cache
5582        // which the next flush() will consume.
5583    }
5584
5585    /// Evict LRU live entries in `namespace` until it fits under both
5586    /// budget caps. Called from `cache_put` after the insert; the new
5587    /// entry is included in the count so we always leave room.
5588    fn evict_to_budget(&mut self, namespace: &str) -> Result<()> {
5589        let budget = self
5590            .cache_budgets
5591            .get(namespace)
5592            .copied()
5593            .unwrap_or_default();
5594        loop {
5595            // Sum current live count + bytes in the namespace.
5596            let mut live_count = 0usize;
5597            let mut live_bytes = 0u64;
5598            for e in &self.cache_dir {
5599                if !e.deleted && e.namespace == namespace {
5600                    live_count += 1;
5601                    live_bytes += e.size_bytes;
5602                }
5603            }
5604            if live_count <= budget.max_entries && live_bytes <= budget.max_bytes {
5605                return Ok(());
5606            }
5607            // Find the LRU (oldest accessed_at) live entry.
5608            let victim_pos = self
5609                .cache_dir
5610                .iter()
5611                .enumerate()
5612                .filter(|(_, e)| !e.deleted && e.namespace == namespace)
5613                .min_by_key(|(_, e)| e.accessed_at)
5614                .map(|(i, _)| i);
5615            let Some(pos) = victim_pos else {
5616                // Nothing to evict but still over budget — the single
5617                // just-inserted entry itself is bigger than
5618                // max_bytes. Nothing we can safely do here without
5619                // rejecting the put; leave it and return.
5620                return Ok(());
5621            };
5622            let key_hash = self.cache_dir[pos].key_hash;
5623            let ns = self.cache_dir[pos].namespace.clone();
5624            self.cache_dir[pos].deleted = true;
5625            self.cache_idx.remove(&ns, &key_hash);
5626            self.cache_counters.evictions += 1;
5627        }
5628    }
5629
5630    // === Semantic cache (Phase 10.2) ==================================
5631
5632    /// Insert or overwrite a **semantic** cache entry — same as
5633    /// [`Mnemo::cache_put`] but with a required embedding vector and
5634    /// model identifier. The vector is stored on the directory entry
5635    /// so [`Mnemo::cache_get_semantic`] can score without decrypting
5636    /// record bodies.
5637    ///
5638    /// Vector dimensionality is validated against the database's
5639    /// configured `dimensions` — a mismatch returns
5640    /// [`MnemoError::DimensionMismatch`], the same signal `remember`
5641    /// gives for the memory catalog.
5642    ///
5643    /// The `model` string is matched **exactly** on lookup — a
5644    /// different model's cache is invisible to a given query. Pick
5645    /// canonical identifiers (`"text-embedding-3-small"`,
5646    /// `"bge-large-en-v1.5"`, ...) and stick to them.
5647    pub fn cache_put_semantic(
5648        &mut self,
5649        namespace: &str,
5650        key: &str,
5651        vector: Vec<f32>,
5652        value: &[u8],
5653        opts: SemanticCachePutOpts,
5654    ) -> Result<()> {
5655        if self.read_only {
5656            return Err(MnemoError::ReadOnly);
5657        }
5658        if vector.len() != self.dimensions {
5659            return Err(MnemoError::DimensionMismatch {
5660                expected: self.dimensions,
5661                got: vector.len(),
5662            });
5663        }
5664        let key_hash = result_cache::hash_key(key);
5665        let key_preview = result_cache::key_preview(key);
5666        let now = memory::now_secs();
5667        let size_bytes = value.len() as u64;
5668
5669        let entry = CacheEntry {
5670            namespace: namespace.to_string(),
5671            key_hash,
5672            key_preview: key_preview.clone(),
5673            value: value.to_vec(),
5674            content_type: opts.content_type.clone(),
5675            created_at: now,
5676            accessed_at: now,
5677            access_count: 0,
5678            ttl_secs: opts.ttl_secs,
5679            cost_hint_ms: opts.cost_hint_ms,
5680            size_bytes,
5681            vector: Some(vector.clone()),
5682            model: Some(opts.model.clone()),
5683        };
5684        let bytes = rmp_serde::to_vec(&entry).map_err(|e| MnemoError::Serialize(e.to_string()))?;
5685        let (start_page, page_count) = self.write_record(&bytes)?;
5686
5687        if let Some(prev_pos) = self.cache_idx.get(namespace, &key_hash) {
5688            self.cache_dir[prev_pos].deleted = true;
5689            self.cache_idx.remove(namespace, &key_hash);
5690        }
5691
5692        let dir_entry = CacheDirectoryEntry {
5693            namespace: namespace.to_string(),
5694            key_hash,
5695            start_page,
5696            page_count,
5697            len: bytes.len() as u32,
5698            deleted: false,
5699            accessed_at: now,
5700            access_count: 0,
5701            size_bytes,
5702            ttl_secs: opts.ttl_secs,
5703            created_at: now,
5704            vector: Some(vector),
5705            model: Some(opts.model),
5706        };
5707        let pos = self.cache_dir.len();
5708        self.cache_dir.push(dir_entry);
5709        self.cache_idx.insert(namespace, key_hash, pos);
5710
5711        self.evict_to_budget(namespace)?;
5712
5713        self.dirty_cache = true;
5714        self.cache_batch.record_dirty();
5715        if self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
5716            self.flush()?;
5717        }
5718        Ok(())
5719    }
5720
5721    /// Semantic recall over the cache: score every live entry in
5722    /// `namespace` whose `model` matches against `query_vector` using
5723    /// cosine similarity; return the top-1 hit and its similarity iff
5724    /// `sim >= threshold`. `None` otherwise.
5725    ///
5726    /// Uses [`crate::result_cache::DEFAULT_SEMANTIC_THRESHOLD`] if you
5727    /// want a sensible default — pass it explicitly (`0.97` today).
5728    ///
5729    /// Exact-key entries (those without a vector) are transparently
5730    /// skipped, as are entries whose model doesn't match — a
5731    /// different-model hit would be a semantic bug, not a win.
5732    /// TTL-expired entries are also skipped.
5733    ///
5734    /// Access-stat updates on hit follow the same policy as
5735    /// [`Mnemo::cache_get`]: bumped on read-write handles, silently
5736    /// skipped on read-only. Hits and misses are recorded in the
5737    /// per-handle counters just like exact-key lookups.
5738    pub fn cache_get_semantic(
5739        &mut self,
5740        namespace: &str,
5741        query_vector: &[f32],
5742        threshold: f32,
5743        model: &str,
5744    ) -> Result<Option<(CachedValue, f32)>> {
5745        if query_vector.len() != self.dimensions {
5746            return Err(MnemoError::DimensionMismatch {
5747                expected: self.dimensions,
5748                got: query_vector.len(),
5749            });
5750        }
5751        // FIX-1 R-001.8 part 4: same Batched max_age check as
5752        // `cache_get`. See the comment there.
5753        if !self.read_only && self.cache_batch.should_auto_flush(&self.cache_flush_policy) {
5754            self.flush()?;
5755        }
5756        let now = memory::now_secs();
5757
5758        // Scan namespace live entries, filter by model, compute
5759        // cosine, keep the best. Exact-key entries have `vector: None`
5760        // and are skipped by the filter.
5761        let mut best: Option<(usize, f32)> = None;
5762        for (i, e) in self.cache_dir.iter().enumerate() {
5763            if e.deleted || e.namespace != namespace || e.is_expired(now) {
5764                continue;
5765            }
5766            let Some(ref v) = e.vector else { continue };
5767            let Some(ref m) = e.model else { continue };
5768            if m != model {
5769                continue;
5770            }
5771            let sim = memory::cosine(query_vector, v);
5772            if best.map(|(_, s)| sim > s).unwrap_or(true) {
5773                best = Some((i, sim));
5774            }
5775        }
5776
5777        let Some((pos, sim)) = best else {
5778            self.cache_counters.misses += 1;
5779            return Ok(None);
5780        };
5781        if sim < threshold {
5782            self.cache_counters.misses += 1;
5783            return Ok(None);
5784        }
5785
5786        // Extract fields before any mutation to keep the borrow checker
5787        // happy.
5788        let (start_page, page_count, len, ttl_secs, created_at) = {
5789            let e = &self.cache_dir[pos];
5790            (e.start_page, e.page_count, e.len, e.ttl_secs, e.created_at)
5791        };
5792
5793        // Decrypt and decode the record body.
5794        let mut buf = Vec::with_capacity(len as usize);
5795        for i in 0..page_count as u64 {
5796            buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
5797        }
5798        buf.truncate(len as usize);
5799        let entry: CacheEntry =
5800            rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
5801
5802        if !self.read_only {
5803            self.cache_dir[pos].accessed_at = now;
5804            self.cache_dir[pos].access_count = self.cache_dir[pos].access_count.saturating_add(1);
5805            self.dirty_cache = true;
5806        }
5807        self.cache_counters.hits += 1;
5808        let access_count = self.cache_dir[pos].access_count;
5809        // FIX-1 R-002.2 part 1: same read-only `accessed_at` fix as
5810        // `cache_get`. The directory carries the live value; the record
5811        // body only holds the insert-time snapshot.
5812        let accessed_at = if self.read_only {
5813            self.cache_dir[pos].accessed_at
5814        } else {
5815            now
5816        };
5817        Ok(Some((
5818            CachedValue {
5819                value: entry.value,
5820                content_type: entry.content_type,
5821                created_at,
5822                accessed_at,
5823                access_count,
5824                ttl_secs,
5825            },
5826            sim,
5827        )))
5828    }
5829
5830    /// Begin a conversation [`Session`](crate::Session) for `agent_id`.
5831    ///
5832    /// The session borrows the database for its lifetime, records turns as
5833    /// working memory, and consolidates them into episodic memory when closed.
5834    pub fn session(&mut self, agent_id: impl Into<String>) -> crate::session::Session<'_> {
5835        crate::session::Session::new(self, agent_id.into())
5836    }
5837
5838    // --- snapshots & point-in-time recovery ------------------------------
5839
5840    /// Every committed transaction, oldest first — the restore points
5841    /// available to [`Mnemo::restore_to`] and [`Mnemo::restore_to_time`].
5842    ///
5843    /// Each `flush` appends one snapshot. Because the storage engine is
5844    /// append-only, the pages a past flush wrote are still on disk, so any
5845    /// listed snapshot can be reinstated exactly. The history reaches back to
5846    /// the last [`Mnemo::compact_file`], which reclaims space by collapsing
5847    /// it.
5848    pub fn snapshots(&self) -> Vec<SnapshotInfo> {
5849        let mut v: Vec<SnapshotInfo> = self
5850            .manifest
5851            .iter()
5852            .map(|s| SnapshotInfo {
5853                txn_id: s.txn_id,
5854                created_at: s.created_at,
5855                memory_count: s.memory_count,
5856            })
5857            .collect();
5858        v.sort_by_key(|s| s.txn_id);
5859        v
5860    }
5861
5862    /// Load a past snapshot's state and commit it as a new transaction.
5863    fn apply_snapshot(&mut self, snap: &Snapshot) -> Result<()> {
5864        let catalog = load_catalog(
5865            &mut self.pager,
5866            snap.catalog_start,
5867            snap.catalog_pages,
5868            snap.catalog_len,
5869        )?;
5870        let ann = load_index(
5871            &mut self.pager,
5872            snap.index_start,
5873            snap.index_pages,
5874            snap.index_len,
5875            self.dimensions,
5876        )?;
5877        self.index = build_id_index(&catalog);
5878        self.catalog = catalog;
5879        self.ann = ann;
5880        self.ann_unloaded = false;
5881        // The handle caches describe the catalog being replaced.
5882        self.key_index = None;
5883        self.superseded = None;
5884        self.layers = None;
5885        self.agent_set = None;
5886        // PERF-1: the snapshot's index may predate its catalog; bring it up
5887        // to date before it's persisted again.
5888        self.reconcile_ann_with_catalog(snap.index_start)?;
5889        // Re-commit as a fresh transaction: crash-safe, and itself recorded
5890        // as a new snapshot so a restore can always be undone.
5891        self.dirty_catalog = true;
5892        self.dirty_index = true;
5893        self.index_persist.forced = true;
5894        self.flush()
5895    }
5896
5897    /// Restore the database to the snapshot produced by transaction `txn_id`.
5898    ///
5899    /// The restore is itself a new committed transaction (and a new
5900    /// snapshot), so it is crash-safe and reversible — restoring forward to a
5901    /// later snapshot afterwards works just as well.
5902    pub fn restore_to(&mut self, txn_id: u64) -> Result<SnapshotInfo> {
5903        if self.read_only {
5904            return Err(MnemoError::ReadOnly);
5905        }
5906        let snap = self
5907            .manifest
5908            .iter()
5909            .find(|s| s.txn_id == txn_id)
5910            .cloned()
5911            .ok_or_else(|| MnemoError::NotFound(format!("snapshot for transaction {txn_id}")))?;
5912        self.apply_snapshot(&snap)?;
5913        Ok(SnapshotInfo {
5914            txn_id: snap.txn_id,
5915            created_at: snap.created_at,
5916            memory_count: snap.memory_count,
5917        })
5918    }
5919
5920    /// Restore the database to the latest snapshot committed at or before
5921    /// `unix_secs`. Returns [`MnemoError::NotFound`] if no snapshot is that
5922    /// old. Like [`Mnemo::restore_to`], the restore is a new transaction.
5923    pub fn restore_to_time(&mut self, unix_secs: i64) -> Result<SnapshotInfo> {
5924        if self.read_only {
5925            return Err(MnemoError::ReadOnly);
5926        }
5927        let snap = self
5928            .manifest
5929            .iter()
5930            .filter(|s| s.created_at <= unix_secs)
5931            .max_by_key(|s| s.txn_id)
5932            .cloned()
5933            .ok_or_else(|| {
5934                MnemoError::NotFound(format!("no snapshot at or before time {unix_secs}"))
5935            })?;
5936        self.apply_snapshot(&snap)?;
5937        Ok(SnapshotInfo {
5938            txn_id: snap.txn_id,
5939            created_at: snap.created_at,
5940            memory_count: snap.memory_count,
5941        })
5942    }
5943
5944    /// Return the KDF (Argon2id) parameters this file currently uses to
5945    /// wrap its DEK. Needed by callers that want to [`rekey`] while
5946    /// keeping the file's original protection level (e.g. the Python
5947    /// binding's `rekey(fast=False)`, whose docstring promises exactly
5948    /// that). FIX-1 R-001.5.
5949    pub fn kdf_params(&self) -> KdfParams {
5950        self.kdf
5951    }
5952
5953    /// Flush any pending writes, then stream the (encrypted) file to
5954    /// `dest` atomically: writes to a sibling temp file in `dest`'s
5955    /// parent directory, `sync_all`s it, renames onto `dest`, and
5956    /// fsyncs the parent directory on Unix so a crash never leaves a
5957    /// partial export.
5958    ///
5959    /// Refuses when `dest` resolves to the file this handle has open
5960    /// — that case would otherwise `File::create`-truncate the live
5961    /// database to 0 bytes (`fs::copy` on Unix reads *after* the
5962    /// `O_TRUNC`, so a naive export-to-self destroys the source).
5963    /// Unix: compares `(dev, ino)` of both paths via `MetadataExt`,
5964    /// catching symlinks and hard links. Windows: compares
5965    /// canonicalized paths; a rename onto an open file also fails on
5966    /// Windows because of the mandatory lock the handle holds, so the
5967    /// rename step is a second line of defense.
5968    ///
5969    /// Reads the bytes from the already-open file handle, not by
5970    /// `fs::copy`. On Windows that matters: `LockFileEx` locks are
5971    /// mandatory, so a second handle opened by `fs::copy` would fail
5972    /// while this writer still holds the lock.
5973    ///
5974    /// FIX-1 R-001.5 + R-002.2 part 3.
5975    pub fn export_encrypted(&mut self, dest: impl AsRef<Path>) -> Result<()> {
5976        let dest = dest.as_ref();
5977
5978        // Step 1 — refuse export to the open file itself. We compare
5979        // by identity, not string, so symlinks and hard links to
5980        // `self.path` are caught too. The check runs before any flush
5981        // so a self-targeted export is a pure no-op.
5982        if paths_refer_to_same_file(&self.path, dest).unwrap_or(false) {
5983            return Err(MnemoError::Invalid(format!(
5984                "export_encrypted refuses to overwrite the open database at {}",
5985                dest.display()
5986            )));
5987        }
5988
5989        // Step 2 — flush, so on-disk bytes match the live state.
5990        if !self.read_only {
5991            self.flush()?;
5992        }
5993
5994        // Step 3 — stream from the open handle into a sibling temp.
5995        // Using the open handle (instead of a second open on `self.path`)
5996        // means a Windows mandatory lock can't block the copy.
5997        let parent = dest.parent().ok_or_else(|| {
5998            MnemoError::Invalid(format!(
5999                "export_encrypted: dest {} has no parent directory",
6000                dest.display()
6001            ))
6002        })?;
6003        let tmp = temp_sibling(dest);
6004        // Scope the temp handle so it closes before the rename on
6005        // Windows (which refuses to rename over an open file).
6006        {
6007            let src = self.pager.file_mut();
6008            src.seek(SeekFrom::Start(0))?;
6009            let mut out = OpenOptions::new()
6010                .write(true)
6011                .create(true)
6012                .truncate(true)
6013                .open(&tmp)?;
6014            let mut buf = vec![0u8; 1 << 20]; // 1 MiB
6015            loop {
6016                let n = src.read(&mut buf)?;
6017                if n == 0 {
6018                    break;
6019                }
6020                out.write_all(&buf[..n])?;
6021            }
6022            out.sync_all()?;
6023            crate::bump_fsync();
6024        }
6025
6026        // Step 4 — atomic rename onto `dest`.
6027        if let Err(e) = std::fs::rename(&tmp, dest) {
6028            // Best-effort cleanup; propagate the rename error.
6029            let _ = std::fs::remove_file(&tmp);
6030            return Err(MnemoError::Io(e));
6031        }
6032
6033        // Step 5 — fsync the parent directory on Unix so the rename
6034        // itself is durable. On Windows directory syncs aren't
6035        // supported; the rename above is atomic enough on NTFS.
6036        #[cfg(unix)]
6037        {
6038            if let Ok(dir) = File::open(parent) {
6039                let _ = dir.sync_all();
6040                crate::bump_fsync();
6041            }
6042        }
6043        #[cfg(not(unix))]
6044        {
6045            let _ = parent;
6046        }
6047
6048        Ok(())
6049    }
6050
6051    /// Change the passphrase. Cheap: re-derives the KEK and re-wraps the DEK;
6052    /// the encrypted pages are never rewritten.
6053    pub fn rekey(&mut self, new_passphrase: &str, kdf: KdfParams) -> Result<()> {
6054        self.rekey_secret(new_passphrase.as_bytes(), kdf)
6055    }
6056
6057    /// PERF-1 step 7: switch to (or rotate) a raw 256-bit key, recording
6058    /// [`KdfParams::raw_key`] in the header. Like [`Mnemo::rekey`], only the
6059    /// wrapped DEK changes; pages aren't rewritten.
6060    pub fn rekey_with_key(&mut self, new_key: &[u8; crypto::KEY_LEN]) -> Result<()> {
6061        self.rekey_secret(new_key, KdfParams::raw_key())
6062    }
6063
6064    fn rekey_secret(&mut self, secret: &[u8], kdf: KdfParams) -> Result<()> {
6065        if self.read_only {
6066            return Err(MnemoError::ReadOnly);
6067        }
6068        self.flush()?;
6069        let new_salt = crypto::random_salt();
6070        let new_kek = crypto::derive_kek(secret, &new_salt, kdf)?;
6071        let new_nonce = crypto::random_nonce();
6072        let wrapped = crypto::wrap_dek(&new_kek, &new_nonce, self.pager.dek())?;
6073        if wrapped.len() != WRAPPED_DEK_LEN {
6074            return Err(MnemoError::Crypto("unexpected wrapped DEK length".into()));
6075        }
6076        let mut wrapped_dek = [0u8; WRAPPED_DEK_LEN];
6077        wrapped_dek.copy_from_slice(&wrapped);
6078
6079        self.header.salt = new_salt;
6080        self.header.dek_nonce = new_nonce;
6081        self.header.wrapped_dek = wrapped_dek;
6082        self.header.m_cost = kdf.m_cost;
6083        self.header.t_cost = kdf.t_cost;
6084        self.header.p_cost = kdf.p_cost;
6085        self.kdf = kdf;
6086
6087        // DUR-1 (R-002.1): WAL-log the new page 0 before touching it.
6088        // Allocate a fresh WAL region at the tail so that a crash during
6089        // the rekey leaves either the old region (old passphrase works)
6090        // or the new region (new passphrase works) fully committed —
6091        // overwriting the existing region would risk destroying both.
6092        let old_wal_start = self.header.wal_start;
6093        let new_pages = self.header.wal_pages;
6094        let new_start = self.header.next_page;
6095        self.header.next_page += new_pages;
6096        self.header.wal_start = new_start;
6097        self.header.wal_pages = new_pages;
6098        self.header.wal_seq += 1;
6099        self.relocate_header_through_wal(old_wal_start)
6100    }
6101
6102    /// Decrypt and validate every live record. Returns the count verified
6103    /// across memories **and** cache entries (FIX-1 R-001.1: now that the
6104    /// cache survives compaction, `verify()` has to walk it too — the
6105    /// pre-FIX-1 implementation only checked `self.catalog`, so a
6106    /// corrupted cache record went unnoticed).
6107    pub fn verify(&mut self) -> Result<usize> {
6108        let entries: Vec<CatalogEntry> = self
6109            .catalog
6110            .iter()
6111            .filter(|e| !e.deleted)
6112            .cloned()
6113            .collect();
6114        let mut n = entries.len();
6115        for e in &entries {
6116            let bytes = self.read_record(e)?;
6117            let _m: Memory = rmp_serde::from_slice(&bytes)
6118                .map_err(|err| MnemoError::Serialize(err.to_string()))?;
6119        }
6120
6121        // Also decode every live cache entry. Tombstoned entries are
6122        // skipped — their pages are orphan ciphertext awaiting
6123        // `compact_file`, same as deleted memories. Expired entries
6124        // are *still* on disk until compaction, so we verify them too.
6125        let cache_positions: Vec<(u64, u32, u32)> = self
6126            .cache_dir
6127            .iter()
6128            .filter(|e| !e.deleted)
6129            .map(|e| (e.start_page, e.page_count, e.len))
6130            .collect();
6131        for (start_page, page_count, len) in cache_positions {
6132            let mut buf = Vec::with_capacity(len as usize);
6133            for i in 0..page_count as u64 {
6134                buf.extend_from_slice(&self.pager.read_page(start_page + i)?);
6135            }
6136            buf.truncate(len as usize);
6137            let _c: CacheEntry = rmp_serde::from_slice(&buf)
6138                .map_err(|err| MnemoError::Serialize(err.to_string()))?;
6139            n += 1;
6140        }
6141        Ok(n)
6142    }
6143
6144    /// Summary statistics for the open database.
6145    pub fn stats(&mut self) -> Result<Stats> {
6146        let deleted = self.catalog.iter().filter(|e| e.deleted).count();
6147        // FIX-1 R-001.8 part 2: build the agent set lazily on first
6148        // call per handle, then maintain in `put` / invalidate in
6149        // `delete`. Pre-FIX-1 every `stats()` call decrypted every
6150        // live memory (O(N) AES-GCM decrypts just to list a usually-
6151        // tiny set of distinct agent IDs). On a 1M-memory database
6152        // this was in the multi-second range on hardware AES.
6153        if self.agent_set.is_none() {
6154            let built: std::collections::BTreeSet<String> =
6155                self.memories()?.into_iter().map(|m| m.agent_id).collect();
6156            self.agent_set = Some(built);
6157        }
6158        let agents: Vec<String> = self
6159            .agent_set
6160            .as_ref()
6161            .expect("agent_set just filled above")
6162            .iter()
6163            .cloned()
6164            .collect();
6165        let file_bytes = self.header.next_page * PAGE_SIZE as u64;
6166        Ok(Stats {
6167            memories: self.len(),
6168            deleted,
6169            dimensions: self.dimensions,
6170            file_bytes,
6171            agents,
6172            encrypted: self.header.flags & FLAG_ENCRYPTED != 0,
6173            created_at: self.header.created_at,
6174            index: {
6175                self.ensure_ann()?;
6176                self.ann.as_ref().map(|a| a.info())
6177            },
6178            wal_pages: self.header.wal_pages,
6179            fsyncs_last_flush: self.fsyncs_last_flush,
6180        })
6181    }
6182
6183    /// Rewrite the file, dropping tombstoned and expired memories and
6184    /// reclaiming stale pages left by updates. Done out-of-place via a temp
6185    /// file and an atomic rename.
6186    ///
6187    /// FIX-1 R-001.1 + R-002.2 part 2: the result cache now survives
6188    /// compaction. Every live, non-expired directory entry is copied
6189    /// into the new file verbatim — `created_at`, `ttl_secs`,
6190    /// `cost_hint_ms`, `key_preview`, `size_bytes`, `content_type`,
6191    /// `vector` and `model` come from the record body; `accessed_at`
6192    /// and `access_count` come from the *directory* (the record body
6193    /// only holds the insert-time snapshot — the directory is the
6194    /// source of truth for the live counts, as of v5's recall trick).
6195    /// Tombstoned and TTL-expired entries are dropped. Pre-FIX-1
6196    /// compaction kept only the memory catalog and silently emptied
6197    /// the cache; the Planner's interim README caveat (c417d2e) called
6198    /// that out — this PR removes the surprise.
6199    pub fn compact_file(path: &str, passphrase: &str) -> Result<CompactReport> {
6200        Self::compact_file_secret(path, passphrase.as_bytes())
6201    }
6202
6203    /// PERF-1 step 7: [`Mnemo::compact_file`] for a raw-key database.
6204    pub fn compact_file_with_key(path: &str, key: &[u8; crypto::KEY_LEN]) -> Result<CompactReport> {
6205        Self::compact_file_secret(path, key)
6206    }
6207
6208    fn compact_file_secret(path: &str, secret: &[u8]) -> Result<CompactReport> {
6209        let mut old = Mnemo::open_secret(path, secret)?;
6210        let dims = old.dimensions;
6211        let kdf = old.kdf;
6212        let tmp = format!("{path}.compact-tmp");
6213
6214        let mut new = Mnemo::create_secret(
6215            &tmp,
6216            secret,
6217            MnemoConfig {
6218                dimensions: dims,
6219                kdf,
6220                ..Default::default()
6221            },
6222        )?;
6223        // ANN-1.1: compaction is "re-tune for the current size" by
6224        // default. Pass IndexConfig::default() so partitions,
6225        // subspaces, n_probe and n_rerank are all re-derived from the
6226        // compacted vector count. Previously we froze the pre-compact
6227        // (n_probe, n_rerank), which left the new index on stale
6228        // numbers for the smaller post-compact corpus.
6229        let want_index = old.has_index();
6230        let now = memory::now_secs();
6231        let all = old.memories()?;
6232        let before = all.len();
6233        let mut after = 0;
6234        for m in all {
6235            if m.is_expired(now) {
6236                continue;
6237            }
6238            new.remember(m)?;
6239            after += 1;
6240        }
6241        if want_index && after > 0 {
6242            new.build_index_with(IndexConfig::default())?;
6243        }
6244
6245        // Count + copy cache entries. We walk `old.cache_dir` by index
6246        // so each hop reads the directory's live `accessed_at` /
6247        // `access_count` directly rather than going through
6248        // `cache_get` (which would also treat read-only semantics,
6249        // tombstones, etc. — we want the raw, authoritative values).
6250        let mut cache_before = 0usize;
6251        let mut cache_after = 0usize;
6252        let positions: Vec<usize> = (0..old.cache_dir.len()).collect();
6253        for pos in positions {
6254            let keep = {
6255                let e = &old.cache_dir[pos];
6256                !e.deleted && !e.is_expired(now)
6257            };
6258            if keep {
6259                cache_before += 1;
6260                new.copy_cache_entry_from(&mut old, pos)?;
6261                cache_after += 1;
6262            }
6263        }
6264
6265        new.flush()?;
6266        drop(new);
6267        drop(old);
6268
6269        std::fs::rename(&tmp, path)?;
6270        Ok(CompactReport {
6271            before,
6272            after,
6273            cache_before,
6274            cache_after,
6275        })
6276    }
6277
6278    /// Copy one live cache entry from `old` at directory position
6279    /// `pos` into `self` (a freshly-created compaction target).
6280    /// Decrypts the old record body, substitutes the old *directory*'s
6281    /// `accessed_at` + `access_count` (the record body's copies are
6282    /// stale as of v5's directory-only hit path), re-encodes, writes
6283    /// a fresh record run, and pushes a matching directory entry.
6284    ///
6285    /// Caller is expected to have already filtered out tombstoned /
6286    /// expired entries and to call this on a read-write handle.
6287    pub(crate) fn copy_cache_entry_from(&mut self, old: &mut Mnemo, pos: usize) -> Result<()> {
6288        if self.read_only {
6289            return Err(MnemoError::ReadOnly);
6290        }
6291        // Snapshot the directory fields we need, so the borrow on
6292        // `old.cache_dir` ends before we touch `old.pager`.
6293        let (start_page, page_count, len, dir_accessed_at, dir_access_count) = {
6294            let e = &old.cache_dir[pos];
6295            (
6296                e.start_page,
6297                e.page_count,
6298                e.len,
6299                e.accessed_at,
6300                e.access_count,
6301            )
6302        };
6303
6304        // Read + decrypt the old record body.
6305        let mut buf = Vec::with_capacity(len as usize);
6306        for i in 0..page_count as u64 {
6307            buf.extend_from_slice(&old.pager.read_page(start_page + i)?);
6308        }
6309        buf.truncate(len as usize);
6310        let mut entry: CacheEntry =
6311            rmp_serde::from_slice(&buf).map_err(|e| MnemoError::Serialize(e.to_string()))?;
6312
6313        // Overwrite the stale body copies with the directory's live
6314        // values before persisting. Everything else — `created_at`,
6315        // `ttl_secs`, `cost_hint_ms`, `key_preview`, `size_bytes`,
6316        // `content_type`, `value`, `vector`, `model` — stays as the
6317        // body already recorded it.
6318        entry.accessed_at = dir_accessed_at;
6319        entry.access_count = dir_access_count;
6320
6321        let bytes =
6322            rmp_serde::to_vec(&entry).map_err(|err| MnemoError::Serialize(err.to_string()))?;
6323        let (new_start_page, new_page_count) = self.write_record(&bytes)?;
6324
6325        let dir_entry = CacheDirectoryEntry {
6326            namespace: entry.namespace.clone(),
6327            key_hash: entry.key_hash,
6328            start_page: new_start_page,
6329            page_count: new_page_count,
6330            len: bytes.len() as u32,
6331            deleted: false,
6332            accessed_at: dir_accessed_at,
6333            access_count: dir_access_count,
6334            size_bytes: entry.size_bytes,
6335            ttl_secs: entry.ttl_secs,
6336            created_at: entry.created_at,
6337            vector: entry.vector.clone(),
6338            model: entry.model.clone(),
6339        };
6340        let new_pos = self.cache_dir.len();
6341        self.cache_idx
6342            .insert(&entry.namespace, entry.key_hash, new_pos);
6343        self.cache_dir.push(dir_entry);
6344        self.dirty_cache = true;
6345        Ok(())
6346    }
6347}