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(§ion_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}