Skip to main content

wm_memory/
store.rs

1//! LMDB-backed memory store.
2//!
3//! Each galaxy is an LMDB named database (sub-DB within the same file).
4//! Reads are zero-copy (mmap'd). Writes are batched.
5
6use lmdb::{
7    Cursor, Database, DatabaseFlags, Environment, EnvironmentFlags, Transaction, WriteFlags,
8};
9use std::collections::HashMap;
10use std::path::Path;
11use std::sync::atomic::{AtomicU64, Ordering};
12use std::sync::{Arc, RwLock};
13use wm_core::{CoreError, Galaxy, Result};
14
15#[cfg(unix)]
16use std::os::unix::fs::PermissionsExt;
17
18use crate::episodic::EpisodicStore;
19use crate::indexes::IndexDbs;
20use crate::memory::{Memory, MemoryId, decode_embedding, encode_embedding};
21use crate::semantic::SemanticEncoder;
22
23/// Query filter for memories.
24#[derive(Debug, Clone, Default)]
25pub struct MemoryQuery {
26    /// Filter by tags (memory must contain ALL specified tags).
27    pub tags: Vec<String>,
28    /// Minimum importance (inclusive).
29    pub min_importance: Option<f32>,
30    /// Maximum importance (inclusive).
31    pub max_importance: Option<f32>,
32    /// Only memories created after this timestamp.
33    pub created_after: Option<chrono::DateTime<chrono::Utc>>,
34    /// Only memories created before this timestamp.
35    pub created_before: Option<chrono::DateTime<chrono::Utc>>,
36    /// Case-insensitive substring filter over content (literal match —
37    /// not tokenized or ranked; that is what the search engine is for).
38    pub content_substring: Option<String>,
39    /// Maximum number of results.
40    pub limit: usize,
41}
42
43impl MemoryQuery {
44    /// Create an empty query (matches all, limit 100).
45    #[must_use]
46    pub fn new() -> Self {
47        Self {
48            limit: 100,
49            ..Default::default()
50        }
51    }
52
53    /// Set tag filter.
54    #[must_use]
55    pub fn with_tags(mut self, tags: Vec<String>) -> Self {
56        self.tags = tags;
57        self
58    }
59
60    /// Set importance range.
61    #[must_use]
62    pub const fn with_importance_range(mut self, min: f32, max: f32) -> Self {
63        self.min_importance = Some(min);
64        self.max_importance = Some(max);
65        self
66    }
67
68    /// Set temporal range.
69    #[must_use]
70    pub const fn with_time_range(
71        mut self,
72        after: chrono::DateTime<chrono::Utc>,
73        before: chrono::DateTime<chrono::Utc>,
74    ) -> Self {
75        self.created_after = Some(after);
76        self.created_before = Some(before);
77        self
78    }
79
80    /// One-sided temporal bound: only memories created at or after this
81    /// timestamp (`created_after` API passthrough).
82    #[must_use]
83    pub const fn with_created_after(mut self, after: chrono::DateTime<chrono::Utc>) -> Self {
84        self.created_after = Some(after);
85        self
86    }
87
88    /// One-sided temporal bound: only memories created at or before this
89    /// timestamp (`created_before` API passthrough).
90    #[must_use]
91    pub const fn with_created_before(mut self, before: chrono::DateTime<chrono::Utc>) -> Self {
92        self.created_before = Some(before);
93        self
94    }
95
96    /// Set limit.
97    #[must_use]
98    pub const fn with_limit(mut self, limit: usize) -> Self {
99        self.limit = limit;
100        self
101    }
102
103    /// Set a case-insensitive substring filter over content.
104    #[must_use]
105    pub fn with_content_substring(mut self, substring: impl Into<String>) -> Self {
106        self.content_substring = Some(substring.into().to_lowercase());
107        self
108    }
109
110    /// Check if a memory matches this query.
111    #[must_use]
112    pub fn matches(&self, mem: &Memory) -> bool {
113        // Tag filter: memory must contain all specified tags
114        if !self.tags.is_empty() {
115            for tag in &self.tags {
116                if !mem.metadata.tags.iter().any(|t| t == tag) {
117                    return false;
118                }
119            }
120        }
121
122        // Importance filter
123        if let Some(min) = self.min_importance {
124            if mem.metadata.importance < min {
125                return false;
126            }
127        }
128        if let Some(max) = self.max_importance {
129            if mem.metadata.importance > max {
130                return false;
131            }
132        }
133
134        // Temporal filter
135        if let Some(after) = self.created_after {
136            if mem.metadata.created_at < after {
137                return false;
138            }
139        }
140        if let Some(before) = self.created_before {
141            if mem.metadata.created_at > before {
142                return false;
143            }
144        }
145
146        // Substring filter (literal, case-insensitive — never ranked).
147        if let Some(sub) = &self.content_substring {
148            if !mem.content.to_lowercase().contains(sub) {
149                return false;
150            }
151        }
152
153        true
154    }
155}
156
157/// The LMDB environment containing all 14 galaxy sub-databases plus 4 index DBs.
158pub struct MemoryStore {
159    /// Path to the LMDB file
160    path: std::path::PathBuf,
161    /// LMDB environment (opened once, shared across threads)
162    env: Environment,
163    /// Cached handles to the 4 secondary index sub-databases
164    index_dbs: IndexDbs,
165    /// Semantic encoder for content-derived coordinates
166    semantic_encoder: SemanticEncoder,
167    /// Optional per-galaxy entry limit (DoS prevention)
168    max_entries_per_galaxy: Option<usize>,
169    /// Monotonic counter of successful mutations since this handle was
170    /// opened. Lets the dispatch pipeline cheaply detect actual store
171    /// writes (write-audit journal) without scanning galaxies.
172    mutation_count: AtomicU64,
173    /// Dedicated database for lossless v6 episodic records.
174    episodic_db: Database,
175    /// DUP_SORT term→id postings for bounded episodic search (v2 sidecar).
176    episodic_terms_v2_db: Database,
177    /// Content-hash → vector cache (v26 "Tier 2" idea, finally wired):
178    /// warm-start for re-ingest and re-runs. Keyed by the embedder
179    /// namespace + content hash, so switching models never serves stale
180    /// vectors.
181    embedding_cache_db: Database,
182    /// Per-memory revision chains (V8 S11c): append-only content-change
183    /// history, self-verifying (seq continuity + hash linkage + head
184    /// match). See [`crate::revision`].
185    revisions_db: Database,
186    /// Per-memory creation attestations (Track F Slice A, D5): one signed
187    /// record per created memory, keyed `att:{galaxy}:{memory_id}`.
188    /// See [`crate::attestation`].
189    attestations_db: Database,
190    /// Dedicated database for compressed cold-stored memories.
191    pub(crate) cold_storage_db: Database,
192    /// Optional at-rest keyring DBI (Q39 slice A). `Some` when the store has
193    /// a keyring; read-only paths open it optionally and never create it.
194    keyring_db: Option<Database>,
195    /// Unlocked galaxy DEKs for writable at-rest stores (slice A verifies
196    /// them at open; record AEAD is slice B). `None` for plaintext stores.
197    at_rest: Option<crate::at_rest::AtRestState>,
198    /// Warm term-posting cache shared by episodic search views.
199    episodic_term_cache: std::sync::Arc<RwLock<HashMap<String, Vec<uuid::Uuid>>>>,
200    /// Optional embedder for episodic vector reranking.
201    episodic_embedder:
202        std::sync::OnceLock<Option<Arc<dyn crate::embedder::Embedder + Send + Sync>>>,
203    /// One-shot guard: rebuild the v2 episodic sidecar once per process.
204    episodic_sidecar_ensured: std::sync::OnceLock<()>,
205    /// Optional adaptive aliases for episodic key expansion.
206    episodic_aliases: std::sync::OnceLock<Option<crate::episodic_keys::AdaptiveAliases>>,
207    /// Optional vocabulary enrichment for episodic index-time term expansion.
208    episodic_enrichment: std::sync::OnceLock<Option<crate::enrichment::VocabularyEnrichment>>,
209}
210
211impl MemoryStore {
212    /// Probe whether another process holds the LMDB writer lock on this store.
213    ///
214    /// LMDB's write env-open falls back to a blocking *shared* lock when the
215    /// exclusive writer lock is held, then opens `data.mdb` for writing
216    /// anyway and wedges on its internal mutex — `wm grimoire`/`wm status`
217    /// hung forever against a live store until a SIGKILL (9.1.6). Callers
218    /// that need exclusive access probe first and fail loudly instead of
219    /// deadlocking.
220    ///
221    /// The probe is a non-blocking `fcntl(F_SETLK, F_WRLCK)` over the whole
222    /// `lock.mdb` (record locks overlap LMDB's byte-range writer lock), so
223    /// it never blocks and never mutates the store.
224    ///
225    /// Returns `Ok(())` when the writer lock is free, `Err(WouldBlock)`
226    /// when another process holds it.
227    #[cfg(unix)]
228    pub fn probe_write_lock(store_root: &Path) -> std::io::Result<()> {
229        use rustix::fs::{FlockOperation, fcntl_lock};
230        let lock_path = store_root.join("lmdb").join("lock.mdb");
231        let file = std::fs::OpenOptions::new()
232            .read(true)
233            .write(true)
234            .open(&lock_path)?;
235        match fcntl_lock(&file, FlockOperation::NonBlockingLockExclusive) {
236            Ok(()) => Ok(()),
237            Err(rustix::io::Errno::AGAIN | rustix::io::Errno::ACCESS) => Err(std::io::Error::new(
238                std::io::ErrorKind::WouldBlock,
239                "LMDB writer lock held by another process",
240            )),
241            Err(e) => Err(e.into()),
242        }
243    }
244
245    /// Non-unix: LMDB locking differs (LockFileEx on Windows); the probe is
246    /// best-effort there and reports the lock as free.
247    #[cfg(not(unix))]
248    pub fn probe_write_lock(_store_root: &Path) -> std::io::Result<()> {
249        Ok(())
250    }
251
252    /// Open or create an LMDB store at the given path.
253    ///
254    /// At-rest mode comes from the environment (`WM_AT_REST_MODE`, default
255    /// `off`) — see [`Self::open_with_at_rest`]. `off` is a provable no-op:
256    /// no keyring DBI is created and no key files are written. A
257    /// set-but-unrecognized `WM_AT_REST_MODE` value refuses the open
258    /// (fail-closed).
259    ///
260    /// On Unix, the store directory is created with mode 0o700 (owner-only
261    /// access) if it does not already exist. Existing directories are
262    /// left untouched.
263    pub fn open(path: impl AsRef<Path>, map_size: usize) -> Result<Self> {
264        Self::open_with_at_rest(path, map_size, &crate::at_rest::AtRestConfig::from_env()?)
265    }
266
267    /// Open or create an LMDB store with an explicit at-rest configuration.
268    ///
269    /// Q39 slice A: when the mode is `keyfile`/`passphrase`, the store's
270    /// keyring DBI is read (or initialized as `meta` + `rk:check` + 16
271    /// wrapped galaxy DEKs, all in one transaction) and the DEKs are
272    /// unwrapped at open. Records stay plaintext in slice A.
273    pub fn open_with_at_rest(
274        path: impl AsRef<Path>,
275        map_size: usize,
276        at_rest_config: &crate::at_rest::AtRestConfig,
277    ) -> Result<Self> {
278        let path = path.as_ref().to_path_buf();
279
280        // Ensure the directory exists with restrictive permissions.
281        std::fs::create_dir_all(&path)
282            .map_err(|e| CoreError::Memory(format!("Cannot create store dir: {e}")))?;
283        #[cfg(unix)]
284        {
285            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700))
286                .map_err(|e| CoreError::Memory(format!("Cannot set store dir permissions: {e}")))?;
287        }
288
289        let env = Environment::new()
290            .set_map_size(map_size)
291            .set_max_dbs(64)
292            .open(&path)
293            .map_err(|e| CoreError::Memory(format!("LMDB open failed: {e}")))?;
294
295        // Create all 14 galaxy sub-databases
296        for galaxy in Galaxy::all() {
297            let db = env
298                .create_db(Some(galaxy.db_name()), DatabaseFlags::default())
299                .map_err(|e| {
300                    CoreError::Memory(format!(
301                        "LMDB create_db failed for {}: {e}",
302                        galaxy.db_name()
303                    ))
304                })?;
305            let _ = db;
306        }
307
308        // Create 4 secondary index sub-databases
309        for (name, flags) in crate::indexes::INDEX_DBS {
310            let db = env
311                .create_db(Some(name), *flags)
312                .map_err(|e| CoreError::Memory(format!("LMDB create_db failed for {name}: {e}")))?;
313            let _ = db;
314        }
315
316        let index_dbs = IndexDbs::open(&env)?;
317        let episodic_db = env
318            .create_db(Some("episodic_records"), DatabaseFlags::default())
319            .map_err(|e| {
320                CoreError::Memory(format!("LMDB create_db failed for episodic_records: {e}"))
321            })?;
322        // v2 sidecar: DUP_SORT postings (term -> set of record ids). Append of
323        // a record touches only the (term, id) pairs it introduces instead of
324        // rewriting whole posting lists, so ingest cost stays O(new records)
325        // as the store grows. The v1 msgpack-Vec database is retained unused
326        // on legacy stores; v2 is rebuilt from the authoritative records when
327        // found empty.
328        let episodic_terms_v2_db = env
329            .create_db(Some("episodic_terms_v2"), DatabaseFlags::DUP_SORT)
330            .map_err(|e| {
331                CoreError::Memory(format!("LMDB create_db failed for episodic_terms_v2: {e}"))
332            })?;
333        let embedding_cache_db = env
334            .create_db(Some("embedding_cache"), DatabaseFlags::default())
335            .map_err(|e| {
336                CoreError::Memory(format!("LMDB create_db failed for embedding_cache: {e}"))
337            })?;
338        let revisions_db = env
339            .create_db(Some("revisions"), DatabaseFlags::default())
340            .map_err(|e| CoreError::Memory(format!("LMDB create_db failed for revisions: {e}")))?;
341        let attestations_db = env
342            .create_db(
343                Some(crate::attestation::ATTESTATIONS_DB),
344                DatabaseFlags::default(),
345            )
346            .map_err(|e| {
347                CoreError::Memory(format!("LMDB create_db failed for attestations: {e}"))
348            })?;
349        let cold_storage_db = env
350            .create_db(Some("cold_storage"), DatabaseFlags::default())
351            .map_err(|e| {
352                CoreError::Memory(format!("LMDB create_db failed for cold_storage: {e}"))
353            })?;
354        let (keyring_db, at_rest) = crate::at_rest::open_at_rest(&env, &path, at_rest_config)?;
355        Ok(Self {
356            path,
357            env,
358            index_dbs,
359            semantic_encoder: SemanticEncoder::new(),
360            max_entries_per_galaxy: None,
361            mutation_count: AtomicU64::new(0),
362            episodic_db,
363            episodic_terms_v2_db,
364            embedding_cache_db,
365            revisions_db,
366            attestations_db,
367            cold_storage_db,
368            keyring_db,
369            at_rest,
370            episodic_term_cache: std::sync::Arc::new(RwLock::new(HashMap::new())),
371            episodic_embedder: std::sync::OnceLock::new(),
372            episodic_sidecar_ensured: std::sync::OnceLock::new(),
373            episodic_aliases: std::sync::OnceLock::new(),
374            episodic_enrichment: std::sync::OnceLock::new(),
375        })
376    }
377
378    /// At-rest disclosure status (Q39 slice A): keyring meta only — the RK is
379    /// never resolved here, and nothing is created or written. `Absent` means
380    /// plaintext pass-through.
381    #[must_use]
382    pub fn at_rest_status(&self) -> crate::at_rest::AtRestStatus {
383        if let Some(state) = &self.at_rest {
384            return crate::at_rest::AtRestStatus::Present(state.status());
385        }
386        match &self.keyring_db {
387            None => crate::at_rest::AtRestStatus::Absent,
388            Some(db) => crate::at_rest::read_status(&self.env, *db, &self.path),
389        }
390    }
391
392    /// Unlocked keyring state for a writable at-rest store (slice B seam);
393    /// `None` for plaintext-pass-through stores and all read paths.
394    #[must_use]
395    pub const fn at_rest_state(&self) -> Option<&crate::at_rest::AtRestState> {
396        self.at_rest.as_ref()
397    }
398
399    /// Keyring DBI handle when the store has one (`None` for plaintext
400    /// pass-through stores and on read paths where the DBI is absent).
401    /// Slice-B migration reads/writes its ledger through this handle.
402    #[must_use]
403    pub const fn keyring_db(&self) -> Option<Database> {
404        self.keyring_db
405    }
406
407    /// Galaxy DEK for record AEAD, when the store runs with an unlocked
408    /// at-rest keyring (Q39 slice B).
409    pub(crate) fn record_cipher(&self, galaxy: Galaxy) -> Option<&[u8; 32]> {
410        self.at_rest
411            .as_ref()
412            .and_then(|state| state.galaxy_dek(galaxy.db_name()))
413    }
414
415    /// Encode a memory for storage: msgpack, sealed under the galaxy DEK
416    /// when the store has one. Keyring-absent stores are byte-identical to
417    /// the pre-slice-B codec.
418    pub(crate) fn encode_record_value(&self, galaxy: Galaxy, memory: &Memory) -> Result<Vec<u8>> {
419        let plaintext = rmp_serde::to_vec_named(memory)
420            .map_err(|e| CoreError::Memory(format!("serialize failed: {e}")))?;
421        let Some(key) = self.record_cipher(galaxy) else {
422            return Ok(plaintext);
423        };
424        crate::codec::seal_record(
425            &plaintext,
426            key,
427            galaxy.db_name(),
428            memory.metadata.id.as_bytes(),
429            memory.metadata.version,
430        )
431        .map_err(|e| CoreError::Memory(format!("at-rest seal failed: {e}")))
432    }
433
434    /// Decode a stored record value: sealed values open under the galaxy
435    /// DEK (failing closed without one), plaintext values follow the legacy
436    /// codec path.
437    pub(crate) fn decode_record_value(
438        &self,
439        galaxy: Galaxy,
440        key_bytes: &[u8],
441        value: &[u8],
442    ) -> Result<Memory> {
443        if crate::codec::is_sealed_record(value) {
444            let Some(dek) = self.record_cipher(galaxy) else {
445                return Err(CoreError::Memory(format!(
446                    "sealed record in {} but no at-rest key is loaded (WM_AT_REST_MODE off?)",
447                    galaxy.db_name()
448                )));
449            };
450            let record_id: [u8; 16] = key_bytes
451                .try_into()
452                .map_err(|_| CoreError::Memory("sealed record key is not a 16-byte id".into()))?;
453            let opened = crate::codec::open_record(value, dek, galaxy.db_name(), &record_id)
454                .map_err(|e| CoreError::Memory(format!("at-rest open failed: {e}")))?;
455            return crate::codec::decode(&opened)
456                .map_err(|e| CoreError::Memory(format!("deserialize failed: {e}")));
457        }
458        crate::codec::decode(value)
459            .map_err(|e| CoreError::Memory(format!("deserialize failed: {e}")))
460    }
461
462    /// Bounded env open for inspection paths (9.1.6).
463    ///
464    /// LMDB env opens can block forever: against a live writer the
465    /// exclusive-lock fallback wedges on an internal mutex, and a crashed
466    /// server can leave the lock file's in-file mutex locked so even
467    /// read-only opens hang. Inspection callers (status, doctor, grimoire)
468    /// must never hang — run the open on a worker thread and give up after
469    /// `timeout`, returning `Ok(None)` so the caller degrades loudly
470    /// instead of deadlocking.
471    pub fn open_readonly_bounded(
472        path: impl AsRef<Path>,
473        timeout: std::time::Duration,
474    ) -> Result<Option<Self>> {
475        let path = path.as_ref().to_path_buf();
476        let (tx, rx) = std::sync::mpsc::channel();
477        std::thread::spawn(move || {
478            let result = Self::open_readonly(&path);
479            let _ = tx.send(result);
480        });
481        match rx.recv_timeout(timeout) {
482            Ok(result) => result.map(Some),
483            Err(_) => Ok(None),
484        }
485    }
486
487    /// Bounded writable env open for exclusive-access paths (9.1.6).
488    /// See [`Self::open_readonly_bounded`] for the wedge rationale; write
489    /// paths bail with an actionable message on timeout instead of hanging.
490    pub fn open_default_bounded(
491        path: impl AsRef<Path>,
492        timeout: std::time::Duration,
493    ) -> Result<Option<Self>> {
494        let path = path.as_ref().to_path_buf();
495        let (tx, rx) = std::sync::mpsc::channel();
496        std::thread::spawn(move || {
497            let result = Self::open_default(&path);
498            let _ = tx.send(result);
499        });
500        match rx.recv_timeout(timeout) {
501            Ok(result) => result.map(Some),
502            Err(_) => Ok(None),
503        }
504    }
505
506    /// Default LMDB map size for this platform (`WM_DEFAULT_MAP_SIZE`
507    /// overrides). See [`Self::open_default`] for the platform rationale.
508    #[must_use]
509    pub fn default_map_size() -> usize {
510        let platform_default = if cfg!(windows) {
511            256 * 1024 * 1024
512        } else {
513            4 * 1024 * 1024 * 1024
514        };
515        std::env::var("WM_DEFAULT_MAP_SIZE")
516            .ok()
517            .and_then(|v| v.parse::<usize>().ok())
518            .filter(|&v| v > 0)
519            .unwrap_or(platform_default)
520    }
521
522    /// Open with the default map size.
523    ///
524    /// 4 GB on Unix: LMDB truncates the data file sparsely (ftruncate), so
525    /// reservation costs nothing until pages are written. On Windows NTFS
526    /// materializes the file at full map size immediately — a 4 GB default
527    /// would allocate 4 GB on disk per store the moment it opens — so the
528    /// Windows default is smaller; pass an explicit size to `open()` for
529    /// large stores. (Auto-grow on MapFull is a planned follow-up.)
530    pub fn open_default(path: impl AsRef<Path>) -> Result<Self> {
531        // Deployment knob: override the platform default explicitly (bytes).
532        // CI uses this on Windows, where NTFS materializes the map file at
533        // full size and hundreds of parallel test stores would exhaust the
534        // runner disk even at the 256MB Windows default.
535        let size = Self::default_map_size();
536        Self::open(path, size)
537    }
538
539    /// Open an existing LMDB store without creating a directory, database, or
540    /// writable LMDB environment. This is the preservation boundary used by
541    /// read-only evaluator servers: an incomplete or incompatible store must
542    /// fail closed for the caller to investigate, never be initialized or
543    /// repaired in place.
544    pub fn open_readonly(path: impl AsRef<Path>) -> Result<Self> {
545        let path = path.as_ref().to_path_buf();
546        if !path.is_dir() {
547            return Err(CoreError::Memory(format!(
548                "Read-only LMDB store directory does not exist: {}",
549                path.display()
550            )));
551        }
552        if !path.join("data.mdb").is_file() {
553            return Err(CoreError::Memory(format!(
554                "Read-only LMDB store is missing data.mdb: {}",
555                path.display()
556            )));
557        }
558
559        let env = Environment::new()
560            .set_max_dbs(32)
561            .set_flags(EnvironmentFlags::READ_ONLY)
562            .open(&path)
563            .map_err(|e| CoreError::Memory(format!("Read-only LMDB open failed: {e}")))?;
564
565        let index_dbs = IndexDbs::open(&env)?;
566        let open_named = |name: &str| {
567            env.open_db(Some(name)).map_err(|e| {
568                CoreError::Memory(format!("Read-only LMDB missing database {name}: {e}"))
569            })
570        };
571        let episodic_db = open_named("episodic_records")?;
572        let episodic_terms_v2_db = open_named("episodic_terms_v2")?;
573        let embedding_cache_db = open_named("embedding_cache")?;
574        let revisions_db = open_named("revisions")?;
575        let attestations_db = open_named(crate::attestation::ATTESTATIONS_DB)?;
576        let cold_storage_db = open_named("cold_storage")?;
577        // The at-rest keyring is optional on read paths: opened when present,
578        // never created, and its RK is never resolved here (status only).
579        let keyring_db = crate::at_rest::open_keyring_optional(&env)?;
580
581        Ok(Self {
582            path,
583            env,
584            index_dbs,
585            semantic_encoder: SemanticEncoder::new(),
586            max_entries_per_galaxy: None,
587            mutation_count: AtomicU64::new(0),
588            episodic_db,
589            episodic_terms_v2_db,
590            embedding_cache_db,
591            revisions_db,
592            attestations_db,
593            cold_storage_db,
594            keyring_db,
595            at_rest: None,
596            episodic_term_cache: std::sync::Arc::new(RwLock::new(HashMap::new())),
597            episodic_embedder: std::sync::OnceLock::new(),
598            episodic_sidecar_ensured: std::sync::OnceLock::new(),
599            episodic_aliases: std::sync::OnceLock::new(),
600            episodic_enrichment: std::sync::OnceLock::new(),
601        })
602    }
603
604    /// Named databases `open()` creates and `open_readonly()` requires,
605    /// beyond the galaxy and secondary-index sets.
606    const NAMED_DBIS: [&'static str; 6] = [
607        "episodic_records",
608        "episodic_terms_v2",
609        "embedding_cache",
610        "revisions",
611        crate::attestation::ATTESTATIONS_DB,
612        "cold_storage",
613    ];
614
615    /// Open an existing store for inspection without taking any lock
616    /// (`MDB_NOLOCK | MDB_RDONLY`), 9.1.6.
617    ///
618    /// Read-only env opens still block forever in two real situations:
619    /// a live writer holds the exclusive lock (lmdb-master falls back to a
620    /// blocking shared-lock wait), and a crashed server can leave the lock
621    /// file's in-file mutex wedged so every open hangs. Inspection paths
622    /// (status, doctor, grimoire) never need the lock file — no locks, no
623    /// reader slots, no mutex — just an mmap read of the store. The store
624    /// must exist and be schema-complete (same strict refusal as
625    /// [`Self::open_readonly`]); torn-meta-page reads are theoretically
626    /// possible mid-write and acceptable for display counts.
627    pub fn open_inspection(path: impl AsRef<Path>) -> Result<Self> {
628        let path = path.as_ref().to_path_buf();
629        if !path.is_dir() {
630            return Err(CoreError::Memory(format!(
631                "Read-only LMDB store directory does not exist: {}",
632                path.display()
633            )));
634        }
635        if !path.join("data.mdb").is_file() {
636            return Err(CoreError::Memory(format!(
637                "Read-only LMDB store is missing data.mdb: {}",
638                path.display()
639            )));
640        }
641
642        let env = Environment::new()
643            .set_max_dbs(32)
644            .set_flags(EnvironmentFlags::READ_ONLY | EnvironmentFlags::NO_LOCK)
645            .open(&path)
646            .map_err(|e| CoreError::Memory(format!("Inspection LMDB open failed: {e}")))?;
647
648        let index_dbs = IndexDbs::open(&env)?;
649        let open_named = |name: &str| {
650            env.open_db(Some(name)).map_err(|e| {
651                CoreError::Memory(format!("Inspection LMDB missing database {name}: {e}"))
652            })
653        };
654        let episodic_db = open_named("episodic_records")?;
655        let episodic_terms_v2_db = open_named("episodic_terms_v2")?;
656        let embedding_cache_db = open_named("embedding_cache")?;
657        let revisions_db = open_named("revisions")?;
658        let attestations_db = open_named(crate::attestation::ATTESTATIONS_DB)?;
659        let cold_storage_db = open_named("cold_storage")?;
660        // The at-rest keyring is optional on read paths: opened when present,
661        // never created, and its RK is never resolved here (status only).
662        let keyring_db = crate::at_rest::open_keyring_optional(&env)?;
663
664        Ok(Self {
665            path,
666            env,
667            index_dbs,
668            semantic_encoder: SemanticEncoder::new(),
669            max_entries_per_galaxy: None,
670            mutation_count: AtomicU64::new(0),
671            episodic_db,
672            episodic_terms_v2_db,
673            embedding_cache_db,
674            revisions_db,
675            attestations_db,
676            cold_storage_db,
677            keyring_db,
678            at_rest: None,
679            episodic_term_cache: std::sync::Arc::new(RwLock::new(HashMap::new())),
680            episodic_embedder: std::sync::OnceLock::new(),
681            episodic_sidecar_ensured: std::sync::OnceLock::new(),
682            episodic_aliases: std::sync::OnceLock::new(),
683            episodic_enrichment: std::sync::OnceLock::new(),
684        })
685    }
686
687    /// Complete a store's schema in place: create any galaxy, index, or named
688    /// database this build expects but an older store lacks, then let the
689    /// caller reopen normally. Returns the database names that were missing.
690    ///
691    /// Restores from older builds can be byte-exact yet not openable (found
692    /// 2026-09-14: a 9.0.0 backup lacked `cold_storage`). This is the only
693    /// in-place repair path; `open_readonly` deliberately stays strict so
694    /// preservation callers see an incomplete store instead of a silent fix.
695    ///
696    /// The at-rest `keyring` DBI is **not** required schema (Q39 slice A
697    /// decision): legacy stores and strict reads stay working, and this
698    /// function neither creates nor repairs a keyring.
699    pub fn ensure_schema(path: impl AsRef<Path>) -> Result<Vec<String>> {
700        let path = path.as_ref().to_path_buf();
701        if !path.is_dir() {
702            return Err(CoreError::Memory(format!(
703                "Store directory does not exist: {}",
704                path.display()
705            )));
706        }
707        let expected = || {
708            Galaxy::all()
709                .into_iter()
710                .map(|galaxy| galaxy.db_name().to_string())
711                .chain(
712                    crate::indexes::INDEX_DBS
713                        .iter()
714                        .map(|(name, _)| (*name).to_string()),
715                )
716                .chain(Self::NAMED_DBIS.iter().map(|name| (*name).to_string()))
717        };
718        let missing: Vec<String> = {
719            let env = Environment::new()
720                .set_max_dbs(64)
721                .set_flags(EnvironmentFlags::READ_ONLY)
722                .open(&path)
723                .map_err(|e| CoreError::Memory(format!("Read-only LMDB open failed: {e}")))?;
724            expected()
725                .filter(|name| env.open_db(Some(name.as_str())).is_err())
726                .collect()
727        };
728        if missing.is_empty() {
729            return Ok(missing);
730        }
731        // A writable open creates every missing galaxy, index, and named
732        // database. Drop it immediately; the caller reopens as usual.
733        let store = Self::open_default(&path)?;
734        drop(store);
735        Ok(missing)
736    }
737
738    /// Set a per-galaxy entry limit for DoS prevention.
739    ///
740    /// When set, `put` will reject writes that would exceed the limit.
741    /// This prevents a single galaxy from exhausting the LMDB map.
742    #[must_use]
743    pub const fn with_entry_limit(mut self, limit: usize) -> Self {
744        self.max_entries_per_galaxy = Some(limit);
745        self
746    }
747
748    /// Path to the LMDB file.
749    pub fn path(&self) -> &Path {
750        &self.path
751    }
752
753    /// Get the LMDB environment handle.
754    pub const fn env(&self) -> &Environment {
755        &self.env
756    }
757
758    /// Monotonic counter of successful mutations since this handle was
759    /// opened (puts, deletes, clears, raw writes). Used by the dispatch
760    /// pipeline's write-audit journal to detect actual store writes.
761    pub fn mutation_count(&self) -> u64 {
762        self.mutation_count.load(Ordering::Relaxed)
763    }
764
765    /// Get the cached index database handles.
766    pub const fn index_dbs(&self) -> &IndexDbs {
767        &self.index_dbs
768    }
769
770    /// Get the semantic encoder.
771    pub const fn semantic_encoder(&self) -> &SemanticEncoder {
772        &self.semantic_encoder
773    }
774
775    /// Rebuild the episodic DUP_SORT sidecar once per process when it is
776    /// empty while authoritative records exist (legacy v1 stores and
777    /// lost-sidecar recovery). Raw records are never modified; a failed
778    /// rebuild leaves search on its raw-scan fallback.
779    fn ensure_episodic_sidecar(&self) {
780        if self.episodic_sidecar_ensured.get().is_some() {
781            return;
782        }
783        let _ = self.episodic_sidecar_ensured.set(());
784        let view = EpisodicStore::new(
785            &self.env,
786            self.episodic_db,
787            self.episodic_terms_v2_db,
788            self.episodic_term_cache.clone(),
789            &self.mutation_count,
790        );
791        let needs_rebuild = matches!(
792            (view.sidecar_is_empty(), view.record_count()),
793            (Ok(true), Ok(n)) if n > 0
794        );
795        if needs_rebuild {
796            match view.rebuild_sidecar() {
797                Ok(n) => tracing::info!("episodic sidecar rebuilt from {n} records"),
798                Err(e) => {
799                    tracing::warn!("episodic sidecar rebuild failed: {e}");
800                }
801            }
802        }
803    }
804
805    /// Open the v6 lossless episodic record view.
806    #[must_use]
807    pub fn episodic(&self) -> EpisodicStore<'_> {
808        self.ensure_episodic_sidecar();
809        let mut store = EpisodicStore::new(
810            &self.env,
811            self.episodic_db,
812            self.episodic_terms_v2_db,
813            self.episodic_term_cache.clone(),
814            &self.mutation_count,
815        );
816        if let Some(Some(embedder)) = self.episodic_embedder.get() {
817            store = store.with_embedder(embedder.clone());
818        }
819        if let Some(Some(aliases)) = self.episodic_aliases.get() {
820            store = store.with_adaptive_aliases(aliases.clone());
821        }
822        if let Some(Some(enrichment)) = self.episodic_enrichment.get() {
823            store = store.with_enrichment(enrichment.clone());
824        }
825        store
826    }
827
828    /// Attach an embedder for episodic vector reranking.
829    pub fn set_episodic_embedder(
830        &self,
831        embedder: Arc<dyn crate::embedder::Embedder + Send + Sync>,
832    ) {
833        let _ = self.episodic_embedder.set(Some(embedder));
834    }
835
836    /// Attach adaptive aliases for episodic key expansion.
837    pub fn set_episodic_aliases(&self, aliases: crate::episodic_keys::AdaptiveAliases) {
838        let _ = self.episodic_aliases.set(Some(aliases));
839    }
840
841    /// Attach vocabulary enrichment for episodic index-time term expansion.
842    pub fn set_episodic_enrichment(&self, enrichment: crate::enrichment::VocabularyEnrichment) {
843        let _ = self.episodic_enrichment.set(Some(enrichment));
844    }
845
846    /// Get a named database handle for a galaxy.
847    pub fn galaxy_db(&self, galaxy: Galaxy) -> Result<Database> {
848        self.env.open_db(Some(galaxy.db_name())).map_err(|e| {
849            CoreError::Memory(format!("LMDB open_db failed for {}: {e}", galaxy.db_name()))
850        })
851    }
852
853    // ── Memory CRUD ───────────────────────────────────────────────────
854
855    /// Store a memory in the given galaxy. Keyed by memory.metadata.id.
856    /// Also updates all secondary indexes.
857    ///
858    /// Returns a clear error if the per-galaxy entry limit is exceeded
859    /// or if the LMDB map is full.
860    pub fn put(&self, galaxy: Galaxy, memory: &Memory) -> Result<()> {
861        // Check per-galaxy entry limit (DoS prevention)
862        if let Some(limit) = self.max_entries_per_galaxy {
863            let current = self.count(galaxy)?;
864            if current >= limit {
865                return Err(CoreError::Memory(format!(
866                    "galaxy {} entry limit reached ({current}/{limit}), write rejected",
867                    galaxy.db_name()
868                )));
869            }
870        }
871
872        let db = self.galaxy_db(galaxy)?;
873        let key = memory.metadata.id.as_bytes();
874        let val = self.encode_record_value(galaxy, memory)?;
875
876        let mut tx = self
877            .env
878            .begin_rw_txn()
879            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
880
881        // Overwrite semantics: capture the previous record (if any) so its
882        // index entries can be removed before the new ones are added.
883        // Otherwise stale tags, importance values, timestamps, and content
884        // hashes stay queryable after updates.
885        let existing = tx
886            .get(db, key)
887            .ok()
888            .and_then(|bytes| self.decode_record_value(galaxy, key, bytes).ok());
889
890        match tx.put(db, key, &val, lmdb::WriteFlags::default()) {
891            Ok(()) => {}
892            Err(lmdb::Error::MapFull) => {
893                tx.abort();
894                return Err(CoreError::Memory(format!(
895                    "LMDB map full: galaxy {}, consider growing map size or pruning old memories",
896                    galaxy.db_name()
897                )));
898            }
899            Err(e) => {
900                tx.abort();
901                return Err(CoreError::Memory(format!("LMDB put failed: {e}")));
902            }
903        }
904        if let Some(existing) = existing {
905            self.index_dbs.remove(&mut tx, galaxy, &existing)?;
906        }
907        self.index_dbs.add(&mut tx, galaxy, memory)?;
908        tx.commit()
909            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
910        self.mutation_count.fetch_add(1, Ordering::Relaxed);
911        Ok(())
912    }
913
914    /// Retrieve a memory by ID from the given galaxy.
915    pub fn get(&self, galaxy: Galaxy, id: uuid::Uuid) -> Result<Option<Memory>> {
916        let db = self.galaxy_db(galaxy)?;
917        let key = id.as_bytes();
918
919        let tx = self
920            .env
921            .begin_ro_txn()
922            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
923        let result = tx.get(db, key);
924        match result {
925            Ok(bytes) => {
926                let memory = self.decode_record_value(galaxy, key, bytes)?;
927                tx.commit()
928                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
929                Ok(Some(memory))
930            }
931            Err(lmdb::Error::NotFound) => {
932                // ReadOnly transactions don't strictly need commit, but it's good practice
933                tx.commit()
934                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
935                Ok(None)
936            }
937            Err(e) => Err(CoreError::Memory(format!("LMDB get failed: {e}"))),
938        }
939    }
940
941    /// Retrieve a memory by ID searching across all memory galaxies (S9 cross-galaxy traversal).
942    ///
943    /// Cross-galaxy associations and edges reference galaxy-blind UUIDs.
944    /// This searches through all memory galaxies in canonical order and returns
945    /// the first matching (Galaxy, Memory) pair, or None if not found.
946    pub fn find_across_galaxies(&self, id: uuid::Uuid) -> Result<Option<(Galaxy, Memory)>> {
947        for galaxy in Galaxy::memory_galaxies() {
948            if let Some(mem) = self.get(galaxy, id)? {
949                return Ok(Some((galaxy, mem)));
950            }
951        }
952        Ok(None)
953    }
954
955    /// Delete a memory by ID from the given galaxy. Returns true if a key was removed.
956    /// Also removes all secondary index entries for the memory.
957    pub fn delete(&self, galaxy: Galaxy, id: uuid::Uuid) -> Result<bool> {
958        let db = self.galaxy_db(galaxy)?;
959        let key = id.as_bytes();
960
961        let mut tx = self
962            .env
963            .begin_rw_txn()
964            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
965
966        // Check if key exists and deserialize for index cleanup
967        let exists = tx.get(db, key).is_ok();
968        if exists {
969            // Read memory to get index values for cleanup
970            if let Ok(bytes) = tx.get(db, key) {
971                if let Ok(memory) = self.decode_record_value(galaxy, key, bytes) {
972                    let _ = self.index_dbs.remove(&mut tx, galaxy, &memory);
973                }
974            }
975            tx.del(db, key, None)
976                .map_err(|e| CoreError::Memory(format!("LMDB del failed: {e}")))?;
977        }
978        tx.commit()
979            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
980        if exists {
981            self.mutation_count.fetch_add(1, Ordering::Relaxed);
982        }
983        Ok(exists)
984    }
985
986    /// Scan up to `limit` memories from the given galaxy (unordered by LMDB page layout).
987    pub fn scan(&self, galaxy: Galaxy, limit: usize) -> Result<Vec<Memory>> {
988        let db = self.galaxy_db(galaxy)?;
989        let tx = self
990            .env
991            .begin_ro_txn()
992            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
993
994        let mut cursor = tx
995            .open_ro_cursor(db)
996            .map_err(|e| CoreError::Memory(format!("LMDB cursor failed: {e}")))?;
997
998        let mut memories = Vec::with_capacity(limit.min(256));
999        for (i, (key, val)) in cursor.iter().enumerate() {
1000            if memories.len() >= limit {
1001                break;
1002            }
1003            match self.decode_record_value(galaxy, key, val) {
1004                Ok(memory) => memories.push(memory),
1005                Err(e) => {
1006                    tracing::warn!(
1007                        "Skipping corrupted entry at index {i} in galaxy {:?}: {e}",
1008                        galaxy
1009                    );
1010                }
1011            }
1012        }
1013
1014        drop(cursor);
1015        tx.commit()
1016            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1017        Ok(memories)
1018    }
1019
1020    /// Scan every memory in the galaxy (unordered by LMDB page layout).
1021    ///
1022    /// Used by maintenance tooling (e.g. index rebuild). The full galaxy is
1023    /// materialized in memory — prefer [`Self::scan`] for bounded reads.
1024    pub fn scan_all(&self, galaxy: Galaxy) -> Result<Vec<Memory>> {
1025        self.scan_all_impl(galaxy, false)
1026    }
1027
1028    /// Maintenance scan that refuses to omit an undecodable source record.
1029    /// Use before replacing derived indexes; a tolerant scan is not a complete
1030    /// authoritative snapshot when any record fails decoding.
1031    pub fn scan_all_strict(&self, galaxy: Galaxy) -> Result<Vec<Memory>> {
1032        self.scan_all_impl(galaxy, true)
1033    }
1034
1035    fn scan_all_impl(&self, galaxy: Galaxy, strict: bool) -> Result<Vec<Memory>> {
1036        let db = self.galaxy_db(galaxy)?;
1037        let tx = self
1038            .env
1039            .begin_ro_txn()
1040            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1041
1042        let mut cursor = tx
1043            .open_ro_cursor(db)
1044            .map_err(|e| CoreError::Memory(format!("LMDB cursor failed: {e}")))?;
1045
1046        let mut memories = Vec::new();
1047        for (i, (key, val)) in cursor.iter().enumerate() {
1048            match self.decode_record_value(galaxy, key, val) {
1049                Ok(memory) => memories.push(memory),
1050                Err(e) => {
1051                    if strict {
1052                        return Err(CoreError::Memory(format!(
1053                            "refusing incomplete scan of {}: record {i} cannot be decoded: {e}",
1054                            galaxy.db_name()
1055                        )));
1056                    }
1057                    tracing::warn!(
1058                        "Skipping corrupted entry at index {i} in galaxy {:?}: {e}",
1059                        galaxy
1060                    );
1061                }
1062            }
1063        }
1064
1065        drop(cursor);
1066        tx.commit()
1067            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1068        Ok(memories)
1069    }
1070
1071    /// Count entries in a galaxy.
1072    pub fn count(&self, galaxy: Galaxy) -> Result<usize> {
1073        let db = self.galaxy_db(galaxy)?;
1074        let tx = self
1075            .env
1076            .begin_ro_txn()
1077            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1078        let mut cursor = tx
1079            .open_ro_cursor(db)
1080            .map_err(|e| CoreError::Memory(format!("LMDB cursor failed: {e}")))?;
1081        let count = cursor.iter().count();
1082        drop(cursor);
1083        tx.commit()
1084            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1085        Ok(count)
1086    }
1087
1088    /// Count entries in a galaxy carrying a tag, using the tag index (no
1089    /// record decoding). `wm status` uses this to report logical sessions
1090    /// (records tagged `start`) instead of every turn/checkpoint record
1091    /// stored in the Sessions galaxy.
1092    pub fn count_by_tag(&self, galaxy: Galaxy, tag: &str) -> Result<usize> {
1093        let tx = self
1094            .env
1095            .begin_ro_txn()
1096            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1097        let ids = self.index_dbs.find_by_tag(&tx, galaxy, tag)?;
1098        tx.commit()
1099            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1100        Ok(ids.len())
1101    }
1102
1103    /// Clear all memories from a galaxy in a single transaction.
1104    /// Returns the number of entries cleared.
1105    /// Also removes all secondary index entries.
1106    pub fn clear_galaxy(&self, galaxy: Galaxy) -> Result<usize> {
1107        let db = self.galaxy_db(galaxy)?;
1108
1109        let mut tx = self
1110            .env
1111            .begin_rw_txn()
1112            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1113
1114        let mut cursor = tx
1115            .open_ro_cursor(db)
1116            .map_err(|e| CoreError::Memory(format!("LMDB cursor failed: {e}")))?;
1117
1118        let mut count = 0usize;
1119        let keys_to_delete: Vec<(Vec<u8>, Memory)> = cursor
1120            .iter()
1121            .filter_map(|(key, val)| {
1122                if let Ok(memory) = self.decode_record_value(galaxy, key, val) {
1123                    Some((key.to_vec(), memory))
1124                } else {
1125                    None
1126                }
1127            })
1128            .collect();
1129
1130        drop(cursor);
1131
1132        for (key, memory) in &keys_to_delete {
1133            let _ = self.index_dbs.remove(&mut tx, galaxy, memory);
1134            tx.del(db, &key, None)
1135                .map_err(|e| CoreError::Memory(format!("LMDB del failed: {e}")))?;
1136            count += 1;
1137        }
1138
1139        tx.commit()
1140            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1141        self.mutation_count
1142            .fetch_add(count as u64, Ordering::Relaxed);
1143        Ok(count)
1144    }
1145
1146    /// Put multiple memories into a galaxy in a single transaction.
1147    /// Returns the number of memories written.
1148    pub fn batch_put(&self, galaxy: Galaxy, memories: &[Memory]) -> Result<usize> {
1149        if memories.is_empty() {
1150            return Ok(0);
1151        }
1152
1153        let db = self.galaxy_db(galaxy)?;
1154
1155        let mut tx = self
1156            .env
1157            .begin_rw_txn()
1158            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1159
1160        let mut count = 0usize;
1161        for memory in memories {
1162            let key = memory.metadata.id.as_bytes();
1163            let val = self.encode_record_value(galaxy, memory)?;
1164            match tx.put(db, key, &val, WriteFlags::default()) {
1165                Ok(()) => {}
1166                Err(lmdb::Error::MapFull) => {
1167                    tx.abort();
1168                    return Err(CoreError::Memory(format!(
1169                        "LMDB map full: galaxy {}, consider growing map size",
1170                        galaxy.db_name()
1171                    )));
1172                }
1173                Err(e) => {
1174                    tx.abort();
1175                    return Err(CoreError::Memory(format!("LMDB put failed: {e}")));
1176                }
1177            }
1178            self.index_dbs.add(&mut tx, galaxy, memory)?;
1179            count += 1;
1180        }
1181
1182        tx.commit()
1183            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1184        self.mutation_count
1185            .fetch_add(count as u64, Ordering::Relaxed);
1186        Ok(count)
1187    }
1188
1189    /// Get raw key-value bytes (for advanced use cases).
1190    pub fn get_raw(&self, galaxy: Galaxy, key: &[u8]) -> Result<Option<Vec<u8>>> {
1191        let db = self.galaxy_db(galaxy)?;
1192        let tx = self
1193            .env
1194            .begin_ro_txn()
1195            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1196        match tx.get(db, &key) {
1197            Ok(bytes) => {
1198                let data = bytes.to_vec();
1199                tx.commit()
1200                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1201                Ok(Some(data))
1202            }
1203            Err(lmdb::Error::NotFound) => {
1204                tx.commit()
1205                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1206                Ok(None)
1207            }
1208            Err(e) => Err(CoreError::Memory(format!("LMDB get_raw failed: {e}"))),
1209        }
1210    }
1211
1212    /// Put raw key-value bytes (for advanced use cases).
1213    pub fn put_raw(&self, galaxy: Galaxy, key: &[u8], val: &[u8]) -> Result<()> {
1214        let db = self.galaxy_db(galaxy)?;
1215        let mut tx = self
1216            .env
1217            .begin_rw_txn()
1218            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1219        tx.put(db, &key, &val, lmdb::WriteFlags::default())
1220            .map_err(|e| CoreError::Memory(format!("LMDB put_raw failed: {e}")))?;
1221        tx.commit()
1222            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1223        self.mutation_count.fetch_add(1, Ordering::Relaxed);
1224        Ok(())
1225    }
1226
1227    /// Delete raw key-value bytes (for advanced use cases).
1228    /// Returns true if a key was removed.
1229    pub fn delete_raw(&self, galaxy: Galaxy, key: &[u8]) -> Result<bool> {
1230        let db = self.galaxy_db(galaxy)?;
1231        let mut tx = self
1232            .env
1233            .begin_rw_txn()
1234            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1235        let deleted = tx.del(db, &key, None).is_ok();
1236        tx.commit()
1237            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1238        if deleted {
1239            self.mutation_count.fetch_add(1, Ordering::Relaxed);
1240        }
1241        Ok(deleted)
1242    }
1243
1244    /// Batch multiple raw key-value writes in a single LMDB transaction.
1245    /// All writes succeed or fail atomically.
1246    pub fn put_raw_batch(&self, galaxy: Galaxy, entries: &[(&[u8], &[u8])]) -> Result<()> {
1247        self.put_raw_batch_impl(galaxy, entries)?;
1248        self.mutation_count
1249            .fetch_add(entries.len() as u64, Ordering::Relaxed);
1250        Ok(())
1251    }
1252
1253    /// Batch multiple raw key-value writes without advancing the mutation
1254    /// counter.
1255    ///
1256    /// For governance bookkeeping (karma chain, write-audit journal): these
1257    /// writes are metadata *about* dispatches, not memory mutations. Letting
1258    /// them tick the counter attributes a whole batch flush to whichever
1259    /// dispatch happens to be in flight when the threshold trips — the
1260    /// 2026-08-28 restore-drill false-positive class ("read-only" tools
1261    /// flagged with the previous batch's size as their write delta).
1262    pub fn put_raw_batch_untracked(
1263        &self,
1264        galaxy: Galaxy,
1265        entries: &[(&[u8], &[u8])],
1266    ) -> Result<()> {
1267        self.put_raw_batch_impl(galaxy, entries)
1268    }
1269
1270    fn put_raw_batch_impl(&self, galaxy: Galaxy, entries: &[(&[u8], &[u8])]) -> Result<()> {
1271        let db = self.galaxy_db(galaxy)?;
1272        let mut tx = self
1273            .env
1274            .begin_rw_txn()
1275            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1276        for (key, val) in entries {
1277            tx.put(db, key, val, WriteFlags::default())
1278                .map_err(|e| CoreError::Memory(format!("LMDB put_raw_batch failed: {e}")))?;
1279        }
1280        tx.commit()
1281            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1282        Ok(())
1283    }
1284
1285    // ── Content-hash Deduplication ────────────────────────────────────
1286
1287    /// Check if a memory with the same content hash already exists in the galaxy.
1288    /// Uses the content_hash index for O(1) lookup.
1289    /// Returns the existing memory's ID if found.
1290    pub fn find_by_content_hash(&self, galaxy: Galaxy, hash: &str) -> Result<Option<uuid::Uuid>> {
1291        let tx = self
1292            .env
1293            .begin_ro_txn()
1294            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1295        let result = self.index_dbs.find_by_content_hash(&tx, galaxy, hash)?;
1296        tx.commit()
1297            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1298        Ok(result)
1299    }
1300
1301    /// Scan-based content hash lookup (O(n) fallback, used for testing index correctness).
1302    pub fn find_by_content_hash_scan(
1303        &self,
1304        galaxy: Galaxy,
1305        hash: &str,
1306    ) -> Result<Option<uuid::Uuid>> {
1307        let memories = self.scan(galaxy, 10_000)?;
1308        for mem in memories {
1309            if mem.metadata.content_hash == hash {
1310                return Ok(Some(mem.metadata.id));
1311            }
1312        }
1313        Ok(None)
1314    }
1315
1316    /// Store a memory with content-hash deduplication.
1317    /// If a memory with the same content already exists in the galaxy,
1318    /// returns the existing memory's ID without creating a duplicate.
1319    pub fn put_dedup(&self, galaxy: Galaxy, memory: &Memory) -> Result<uuid::Uuid> {
1320        if let Some(existing_id) =
1321            self.find_by_content_hash(galaxy, &memory.metadata.content_hash)?
1322        {
1323            return Ok(existing_id);
1324        }
1325        let id = memory.metadata.id;
1326        self.put(galaxy, memory)?;
1327        Ok(id)
1328    }
1329
1330    // ── Write Batching ─────────────────────────────────────────────────
1331
1332    /// Store multiple memories in a single LMDB transaction (batch write).
1333    /// All writes and index updates succeed or fail atomically.
1334    pub fn put_batch(&self, galaxy: Galaxy, memories: &[Memory]) -> Result<()> {
1335        let db = self.galaxy_db(galaxy)?;
1336        let mut tx = self
1337            .env
1338            .begin_rw_txn()
1339            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1340
1341        for memory in memories {
1342            let key = memory.metadata.id.as_bytes();
1343            let val = self.encode_record_value(galaxy, memory)?;
1344            tx.put(db, key, &val, WriteFlags::default())
1345                .map_err(|e| CoreError::Memory(format!("LMDB put_batch failed: {e}")))?;
1346            self.index_dbs.add(&mut tx, galaxy, memory)?;
1347        }
1348
1349        tx.commit()
1350            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1351        self.mutation_count
1352            .fetch_add(memories.len() as u64, Ordering::Relaxed);
1353        Ok(())
1354    }
1355
1356    // ── Query API ──────────────────────────────────────────────────────
1357
1358    /// Query memories in a galaxy with filtering.
1359    /// Uses secondary indexes when the query is a pure single-dimension filter
1360    /// (single tag, importance range, or time range with no other filters).
1361    /// Falls back to scan for complex multi-dimensional queries.
1362    pub fn query(&self, galaxy: Galaxy, query: &MemoryQuery) -> Result<Vec<Memory>> {
1363        // Try indexed fast paths for single-dimension queries. A substring
1364        // filter forces the full scan — the indexes cannot evaluate it.
1365        if query.content_substring.is_none()
1366            && query.tags.len() == 1
1367            && query.min_importance.is_none()
1368            && query.max_importance.is_none()
1369            && query.created_after.is_none()
1370            && query.created_before.is_none()
1371        {
1372            return self.query_by_tag_indexed(galaxy, &query.tags[0], query.limit);
1373        }
1374
1375        if query.content_substring.is_none()
1376            && query.tags.is_empty()
1377            && let Some(min) = query.min_importance
1378            && let Some(max) = query.max_importance
1379            && query.created_after.is_none()
1380            && query.created_before.is_none()
1381        {
1382            return self.query_by_importance_indexed(galaxy, min, max, query.limit);
1383        }
1384
1385        if query.content_substring.is_none()
1386            && query.tags.is_empty()
1387            && query.min_importance.is_none()
1388            && query.max_importance.is_none()
1389            && let Some(after) = query.created_after
1390            && let Some(before) = query.created_before
1391        {
1392            return self.query_by_time_indexed(galaxy, after, before, query.limit);
1393        }
1394
1395        // Fallback: full scan with in-memory filter
1396        let memories = self.scan(galaxy, 10_000)?;
1397        let mut results = Vec::new();
1398        for mem in memories {
1399            if query.matches(&mem) {
1400                results.push(mem);
1401                if results.len() >= query.limit {
1402                    break;
1403                }
1404            }
1405        }
1406        Ok(results)
1407    }
1408
1409    /// Tag-based indexed query → memories with the given tag.
1410    fn query_by_tag_indexed(&self, galaxy: Galaxy, tag: &str, limit: usize) -> Result<Vec<Memory>> {
1411        let db = self.galaxy_db(galaxy)?;
1412        let tx = self
1413            .env
1414            .begin_ro_txn()
1415            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1416        let ids = self.index_dbs.find_by_tag(&tx, galaxy, tag)?;
1417        let mut results = Vec::new();
1418        for id in &ids {
1419            if results.len() >= limit {
1420                break;
1421            }
1422            if let Ok(bytes) = tx.get(db, id.as_bytes()) {
1423                if let Ok(mem) = self.decode_record_value(galaxy, id.as_bytes(), bytes) {
1424                    results.push(mem);
1425                }
1426            }
1427        }
1428        tx.commit()
1429            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1430        Ok(results)
1431    }
1432
1433    /// Importance-range indexed query → memories with importance in [min, max].
1434    fn query_by_importance_indexed(
1435        &self,
1436        galaxy: Galaxy,
1437        min: f32,
1438        max: f32,
1439        limit: usize,
1440    ) -> Result<Vec<Memory>> {
1441        let db = self.galaxy_db(galaxy)?;
1442        let tx = self
1443            .env
1444            .begin_ro_txn()
1445            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1446        let ids = self
1447            .index_dbs
1448            .find_by_importance_range(&tx, galaxy, min, max)?;
1449        let mut results = Vec::new();
1450        for id in &ids {
1451            if results.len() >= limit {
1452                break;
1453            }
1454            if let Ok(bytes) = tx.get(db, id.as_bytes()) {
1455                if let Ok(mem) = self.decode_record_value(galaxy, id.as_bytes(), bytes) {
1456                    results.push(mem);
1457                }
1458            }
1459        }
1460        tx.commit()
1461            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1462        Ok(results)
1463    }
1464
1465    /// Time-range indexed query → memories created in [after, before].
1466    fn query_by_time_indexed(
1467        &self,
1468        galaxy: Galaxy,
1469        after: chrono::DateTime<chrono::Utc>,
1470        before: chrono::DateTime<chrono::Utc>,
1471        limit: usize,
1472    ) -> Result<Vec<Memory>> {
1473        let db = self.galaxy_db(galaxy)?;
1474        let tx = self
1475            .env
1476            .begin_ro_txn()
1477            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1478        let ids = self
1479            .index_dbs
1480            .find_by_time_range(&tx, galaxy, after, before)?;
1481        let mut results = Vec::new();
1482        for id in &ids {
1483            if results.len() >= limit {
1484                break;
1485            }
1486            if let Ok(bytes) = tx.get(db, id.as_bytes()) {
1487                if let Ok(mem) = self.decode_record_value(galaxy, id.as_bytes(), bytes) {
1488                    results.push(mem);
1489                }
1490            }
1491        }
1492        tx.commit()
1493            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1494        Ok(results)
1495    }
1496
1497    // ── Semantic Coordinate Encoding ─────────────────────────────────────
1498
1499    /// Store a memory with semantically-derived 5D coordinates.
1500    ///
1501    /// Replaces the SHA-256 hash-based `Coordinate5D::encode()` with
1502    /// anchor-based TF projection. The memory's `coord5d` field is updated
1503    /// with semantically meaningful x/y/z values before storage.
1504    pub fn put_semantic(&self, galaxy: Galaxy, memory: &mut Memory) -> Result<()> {
1505        let temporal_weight = memory.metadata.coord5d.w;
1506        let importance = memory.metadata.importance;
1507        memory.metadata.coord5d =
1508            self.semantic_encoder
1509                .encode_coordinate(&memory.content, temporal_weight, importance);
1510        self.put(galaxy, memory)
1511    }
1512
1513    /// Find memories in a galaxy with content semantically similar to the query text.
1514    ///
1515    /// Encodes the query text into a 5D coordinate and scans the galaxy,
1516    /// returning memories sorted by semantic distance (nearest first).
1517    pub fn find_similar(
1518        &self,
1519        galaxy: Galaxy,
1520        query_text: &str,
1521        limit: usize,
1522    ) -> Result<Vec<(Memory, f32)>> {
1523        let query_coord = self
1524            .semantic_encoder
1525            .encode_coordinate(query_text, 0.5, 0.5);
1526        let memories = self.scan(galaxy, 10_000)?;
1527        let mut results: Vec<(Memory, f32)> = memories
1528            .into_iter()
1529            .map(|m| {
1530                let dist = query_coord.semantic_distance_to(&m.metadata.coord5d);
1531                (m, dist)
1532            })
1533            .collect();
1534        results.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal));
1535        results.truncate(limit);
1536        Ok(results)
1537    }
1538
1539    // ── Embedding Storage ──────────────────────────────────────────────
1540
1541    /// Store an embedding vector for a memory in the Embeddings galaxy.
1542    /// Keyed by the memory's UUID.
1543    pub fn put_embedding(&self, memory_id: uuid::Uuid, embedding: &[f32]) -> Result<()> {
1544        let db = self.galaxy_db(Galaxy::Embeddings)?;
1545        let key = memory_id.as_bytes();
1546        let val = encode_embedding(embedding);
1547
1548        let mut tx = self
1549            .env
1550            .begin_rw_txn()
1551            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1552        tx.put(db, key, &val, WriteFlags::default())
1553            .map_err(|e| CoreError::Memory(format!("LMDB put_embedding failed: {e}")))?;
1554        tx.commit()
1555            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1556        self.mutation_count.fetch_add(1, Ordering::Relaxed);
1557        Ok(())
1558    }
1559
1560    /// Retrieve an embedding vector for a memory from the Embeddings galaxy.
1561    pub fn get_embedding(&self, memory_id: uuid::Uuid) -> Result<Option<Vec<f32>>> {
1562        let db = self.galaxy_db(Galaxy::Embeddings)?;
1563        let key = memory_id.as_bytes();
1564
1565        let tx = self
1566            .env
1567            .begin_ro_txn()
1568            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1569        match tx.get(db, key) {
1570            Ok(bytes) => {
1571                let embedding = decode_embedding(bytes);
1572                tx.commit()
1573                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1574                Ok(Some(embedding))
1575            }
1576            Err(lmdb::Error::NotFound) => {
1577                tx.commit()
1578                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1579                Ok(None)
1580            }
1581            Err(e) => Err(CoreError::Memory(format!("LMDB get_embedding failed: {e}"))),
1582        }
1583    }
1584
1585    /// Delete an embedding vector from the Embeddings galaxy.
1586    pub fn delete_embedding(&self, memory_id: uuid::Uuid) -> Result<bool> {
1587        let db = self.galaxy_db(Galaxy::Embeddings)?;
1588        let key = memory_id.as_bytes();
1589
1590        let mut tx = self
1591            .env
1592            .begin_rw_txn()
1593            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1594        let exists = tx.get(db, key).is_ok();
1595        if exists {
1596            tx.del(db, key, None)
1597                .map_err(|e| CoreError::Memory(format!("LMDB del_embedding failed: {e}")))?;
1598        }
1599        tx.commit()
1600            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1601        if exists {
1602            self.mutation_count.fetch_add(1, Ordering::Relaxed);
1603        }
1604        Ok(exists)
1605    }
1606
1607    // ── Embedding cache (content-hash → vector) ────────────────────────
1608
1609    /// Store a cached embedding under a caller-computed cache key
1610    /// (embedder namespace + content hash). Vectors for the same content
1611    /// differ across models, so the key must carry the namespace.
1612    pub fn put_embedding_cache(&self, cache_key: &str, embedding: &[f32]) -> Result<()> {
1613        let mut tx = self
1614            .env
1615            .begin_rw_txn()
1616            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1617        tx.put(
1618            self.embedding_cache_db,
1619            &cache_key.as_bytes().to_vec(),
1620            &encode_embedding(embedding),
1621            WriteFlags::default(),
1622        )
1623        .map_err(|e| CoreError::Memory(format!("LMDB put_embedding_cache failed: {e}")))?;
1624        tx.commit()
1625            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1626        self.mutation_count.fetch_add(1, Ordering::Relaxed);
1627        Ok(())
1628    }
1629
1630    /// Batched cache write: one transaction for the whole ingest chunk.
1631    pub fn put_embedding_cache_batch(&self, entries: &[(String, Vec<f32>)]) -> Result<()> {
1632        if entries.is_empty() {
1633            return Ok(());
1634        }
1635        let mut tx = self
1636            .env
1637            .begin_rw_txn()
1638            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1639        for (key, embedding) in entries {
1640            tx.put(
1641                self.embedding_cache_db,
1642                &key.as_bytes().to_vec(),
1643                &encode_embedding(embedding),
1644                WriteFlags::default(),
1645            )
1646            .map_err(|e| CoreError::Memory(format!("LMDB put_embedding_cache failed: {e}")))?;
1647        }
1648        tx.commit()
1649            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1650        self.mutation_count
1651            .fetch_add(entries.len() as u64, Ordering::Relaxed);
1652        Ok(())
1653    }
1654
1655    /// Look up a cached embedding. `Ok(None)` = miss; the caller embeds.
1656    pub fn get_embedding_cache(&self, cache_key: &str) -> Result<Option<Vec<f32>>> {
1657        let tx = self
1658            .env
1659            .begin_ro_txn()
1660            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1661        match tx.get(self.embedding_cache_db, &cache_key.as_bytes().to_vec()) {
1662            Ok(bytes) => {
1663                let embedding = decode_embedding(bytes);
1664                tx.commit()
1665                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1666                Ok(Some(embedding))
1667            }
1668            Err(lmdb::Error::NotFound) => {
1669                tx.commit()
1670                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1671                Ok(None)
1672            }
1673            Err(e) => Err(CoreError::Memory(format!(
1674                "LMDB get_embedding_cache failed: {e}"
1675            ))),
1676        }
1677    }
1678
1679    /// Batched lookup: one read transaction for the whole ingest chunk.
1680    /// Result aligns 1:1 with `keys`.
1681    pub fn get_embedding_cache_batch(&self, keys: &[String]) -> Result<Vec<Option<Vec<f32>>>> {
1682        let tx = self
1683            .env
1684            .begin_ro_txn()
1685            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1686        let mut out = Vec::with_capacity(keys.len());
1687        for key in keys {
1688            out.push(
1689                tx.get(self.embedding_cache_db, &key.as_bytes().to_vec())
1690                    .ok()
1691                    .map(decode_embedding),
1692            );
1693        }
1694        tx.commit()
1695            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1696        Ok(out)
1697    }
1698
1699    /// Number of cached vectors (doctor / honesty surfaces).
1700    pub fn embedding_cache_count(&self) -> Result<u64> {
1701        let tx = self
1702            .env
1703            .begin_ro_txn()
1704            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1705        let mut cursor = tx
1706            .open_ro_cursor(self.embedding_cache_db)
1707            .map_err(|e| CoreError::Memory(format!("LMDB cursor embedding_cache failed: {e}")))?;
1708        let mut count = 0u64;
1709        for _ in cursor.iter() {
1710            count += 1;
1711        }
1712        Ok(count)
1713    }
1714
1715    // ── Revision chain (V8 S11c) ────────────────────────────────────────
1716
1717    /// Append one content revision to a memory's chain. Seq is derived
1718    /// from the current chain tail (append-only by convention); the write
1719    /// bumps the store mutation counter so dispatch windows attribute it.
1720    pub fn record_revision(
1721        &self,
1722        galaxy: Galaxy,
1723        id: MemoryId,
1724        old_hash: &str,
1725        new_hash: &str,
1726        actor: crate::revision::RevisionActor,
1727    ) -> Result<crate::revision::MemoryRevision> {
1728        let seq = self.revisions(galaxy, id)?.len() as u32;
1729        let entry = crate::revision::MemoryRevision {
1730            seq,
1731            timestamp: wm_core::time::now_unix_secs(),
1732            old_hash: old_hash.to_string(),
1733            new_hash: new_hash.to_string(),
1734            actor_session: actor.session,
1735            actor_user: actor.user,
1736            actor_compartment: actor.compartment,
1737        };
1738        let key = crate::revision::revision_key(galaxy, id, seq);
1739        let val = serde_json::to_vec(&entry)
1740            .map_err(|e| CoreError::Memory(format!("revision serialize failed: {e}")))?;
1741        let mut tx = self
1742            .env
1743            .begin_rw_txn()
1744            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1745        tx.put(self.revisions_db, &key, &val, WriteFlags::default())
1746            .map_err(|e| CoreError::Memory(format!("LMDB put revision failed: {e}")))?;
1747        tx.commit()
1748            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1749        self.mutation_count.fetch_add(1, Ordering::Relaxed);
1750        Ok(entry)
1751    }
1752
1753    /// Full revision chain for a memory, ordered by seq. Empty for
1754    /// memories never content-updated (or pre-S11c).
1755    pub fn revisions(
1756        &self,
1757        galaxy: Galaxy,
1758        id: MemoryId,
1759    ) -> Result<Vec<crate::revision::MemoryRevision>> {
1760        // Cursor-op constants from lmdb.h (frozen LMDB ABI): the `lmdb`
1761        // crate's `iter_from` panics on a SetRange miss, and a miss is a
1762        // normal state here (a memory with no revisions yet sorts after
1763        // every existing key), so the cursor is driven manually.
1764        const MDB_GET_CURRENT: u32 = 4;
1765        const MDB_NEXT: u32 = 8;
1766        const MDB_SET_RANGE: u32 = 17;
1767        let prefix = crate::revision::revision_prefix(galaxy, id);
1768        let tx = self
1769            .env
1770            .begin_ro_txn()
1771            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1772        let cursor = tx
1773            .open_ro_cursor(self.revisions_db)
1774            .map_err(|e| CoreError::Memory(format!("LMDB cursor revisions failed: {e}")))?;
1775        let mut out = Vec::new();
1776        if cursor.get(Some(&prefix), None, MDB_SET_RANGE).is_ok() {
1777            while let Ok((key, val)) = cursor.get(None, None, MDB_GET_CURRENT) {
1778                // `key` is None only for ops that return no key — GET_CURRENT
1779                // after a positioned read always carries one.
1780                if !key.is_some_and(|k| k.starts_with(&prefix)) {
1781                    break;
1782                }
1783                let entry: crate::revision::MemoryRevision = serde_json::from_slice(val)
1784                    .map_err(|e| CoreError::Memory(format!("revision deserialize failed: {e}")))?;
1785                out.push(entry);
1786                // Bounded walk — corrupt data can never spin this loop.
1787                if out.len() >= 10_000 || cursor.get(None, None, MDB_NEXT).is_err() {
1788                    break;
1789                }
1790            }
1791        }
1792        drop(cursor);
1793        tx.commit()
1794            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1795        Ok(out)
1796    }
1797
1798    /// Walk one memory's revision chain and grade it against the memory's
1799    /// current content hash (seq continuity, hash linkage, head match).
1800    pub fn verify_revision_chain(
1801        &self,
1802        galaxy: Galaxy,
1803        id: MemoryId,
1804        current_hash: &str,
1805    ) -> Result<crate::revision::RevisionChainReport> {
1806        let entries = self.revisions(galaxy, id)?;
1807        Ok(crate::revision::verify_chain(&entries, current_hash))
1808    }
1809
1810    // ── Record attestations (Track F Slice A, D5) ───────────────────────
1811
1812    /// Record one creation attestation for a memory. Upsert by key
1813    /// (`att:{galaxy}:{memory_id}`): re-attestation overwrites, which is
1814    /// safe because attestation covers the *creation* event and memory ids
1815    /// are unique per create. The write bumps the mutation counter like
1816    /// every other store mutation.
1817    pub fn record_attestation(
1818        &self,
1819        galaxy: Galaxy,
1820        id: MemoryId,
1821        entry: &crate::attestation::RecordAttestation,
1822    ) -> Result<()> {
1823        let key = crate::attestation::attestation_key(galaxy, id);
1824        let val = serde_json::to_vec(entry)
1825            .map_err(|e| CoreError::Memory(format!("attestation serialize failed: {e}")))?;
1826        let mut tx = self
1827            .env
1828            .begin_rw_txn()
1829            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1830        tx.put(self.attestations_db, &key, &val, WriteFlags::default())
1831            .map_err(|e| CoreError::Memory(format!("LMDB put attestation failed: {e}")))?;
1832        tx.commit()
1833            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1834        self.mutation_count.fetch_add(1, Ordering::Relaxed);
1835        Ok(())
1836    }
1837
1838    /// This memory's creation attestation, if the creating dispatch signed
1839    /// one (key-available creates only — absence is honest, not an error).
1840    pub fn attestation(
1841        &self,
1842        galaxy: Galaxy,
1843        id: MemoryId,
1844    ) -> Result<Option<crate::attestation::RecordAttestation>> {
1845        let key = crate::attestation::attestation_key(galaxy, id);
1846        let tx = self
1847            .env
1848            .begin_ro_txn()
1849            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1850        let out =
1851            match tx.get(self.attestations_db, &key) {
1852                Ok(val) => Some(serde_json::from_slice(val).map_err(|e| {
1853                    CoreError::Memory(format!("attestation deserialize failed: {e}"))
1854                })?),
1855                Err(lmdb::Error::NotFound) => None,
1856                Err(e) => {
1857                    return Err(CoreError::Memory(format!(
1858                        "LMDB get attestation failed: {e}"
1859                    )));
1860                }
1861            };
1862        drop(tx);
1863        Ok(out)
1864    }
1865
1866    /// Every attestation in the store (for `wm anchor`). Bounded walk —
1867    /// corrupt data can never spin this loop.
1868    pub fn scan_attestations(&self) -> Result<Vec<crate::attestation::RecordAttestation>> {
1869        const MDB_GET_CURRENT: u32 = 4;
1870        const MDB_NEXT: u32 = 8;
1871        const MDB_FIRST: u32 = 9;
1872        let tx = self
1873            .env
1874            .begin_ro_txn()
1875            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
1876        let cursor = tx
1877            .open_ro_cursor(self.attestations_db)
1878            .map_err(|e| CoreError::Memory(format!("LMDB cursor attestations failed: {e}")))?;
1879        let mut out = Vec::new();
1880        if cursor.get(None, None, MDB_FIRST).is_ok() {
1881            while let Ok((_, val)) = cursor.get(None, None, MDB_GET_CURRENT) {
1882                let entry: crate::attestation::RecordAttestation = serde_json::from_slice(val)
1883                    .map_err(|e| {
1884                        CoreError::Memory(format!("attestation deserialize failed: {e}"))
1885                    })?;
1886                out.push(entry);
1887                // Bounded walk — corrupt data can never spin this loop.
1888                if out.len() >= 1_000_000 || cursor.get(None, None, MDB_NEXT).is_err() {
1889                    break;
1890                }
1891            }
1892        }
1893        drop(cursor);
1894        tx.commit()
1895            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
1896        Ok(out)
1897    }
1898
1899    /// Grade one memory's attestation: presence, signature validity, and
1900    /// whether the attested hash still matches the live content hash.
1901    /// A content *update* after creation flips `matches_head` to false by
1902    /// design — updates are covered by the revisions chain, not by
1903    /// re-attestation.
1904    pub fn verify_attestation(
1905        &self,
1906        galaxy: Galaxy,
1907        id: MemoryId,
1908    ) -> Result<crate::attestation::AttestationReport> {
1909        use crate::attestation::AttestationReport;
1910        let Some(att) = self.attestation(galaxy, id)? else {
1911            return Ok(AttestationReport {
1912                attested: false,
1913                signature_valid: false,
1914                matches_head: false,
1915                memory_present: self.get(galaxy, id)?.is_some(),
1916                breaks: vec!["no attestation recorded for this memory".to_string()],
1917            });
1918        };
1919        let mut breaks = Vec::new();
1920        let signature_valid = crate::attestation::verify_attestation(&att);
1921        if !signature_valid {
1922            breaks.push("signature does not verify against recorded pubkey".to_string());
1923        }
1924        let (matches_head, memory_present) = if let Some(memory) = self.get(galaxy, id)? {
1925            let matches = memory.metadata.content_hash == att.record_hash;
1926            if !matches {
1927                breaks.push(
1928                    "attested record_hash != live content_hash (memory updated after attestation)"
1929                        .to_string(),
1930                );
1931            }
1932            (matches, true)
1933        } else {
1934            breaks.push("attested memory id not present in galaxy".to_string());
1935            (false, false)
1936        };
1937        Ok(AttestationReport {
1938            attested: true,
1939            signature_valid,
1940            matches_head,
1941            memory_present,
1942            breaks,
1943        })
1944    }
1945
1946    /// Grade every attestation in the store (for `wm anchor`). Entries
1947    /// whose galaxy/id no longer parse are reported as broken — never
1948    /// skipped silently.
1949    pub fn attestation_sweep(
1950        &self,
1951    ) -> Result<
1952        Vec<(
1953            crate::attestation::RecordAttestation,
1954            crate::attestation::AttestationReport,
1955        )>,
1956    > {
1957        use crate::attestation::AttestationReport;
1958        let mut out = Vec::new();
1959        for att in self.scan_attestations()? {
1960            let parsed = match (
1961                Galaxy::from_db_name(&att.galaxy),
1962                uuid::Uuid::parse_str(&att.memory_id),
1963            ) {
1964                (Some(galaxy), Ok(id)) => Some((galaxy, id)),
1965                _ => None,
1966            };
1967            match parsed {
1968                Some((galaxy, id)) => out.push((att, self.verify_attestation(galaxy, id)?)),
1969                None => out.push((
1970                    att,
1971                    AttestationReport {
1972                        attested: true,
1973                        signature_valid: false,
1974                        matches_head: false,
1975                        memory_present: false,
1976                        breaks: vec![
1977                            "attestation row has unparseable galaxy or memory id".to_string(),
1978                        ],
1979                    },
1980                )),
1981            }
1982        }
1983        Ok(out)
1984    }
1985
1986    // ── Non-Destructive Phagic Cold-Storage (the project's sacred rule) ───────
1987
1988    /// Store a compressed cold record in the cold_storage DBI.
1989    pub fn put_cold_record(&self, record: &crate::cold_storage::ColdRecord) -> Result<()> {
1990        let key = record.id.as_bytes();
1991        let val = rmp_serde::to_vec_named(record)
1992            .map_err(|e| CoreError::Memory(format!("Cold record serialization failed: {e}")))?;
1993        let mut tx = self
1994            .env
1995            .begin_rw_txn()
1996            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
1997        tx.put(self.cold_storage_db, key, &val, WriteFlags::default())
1998            .map_err(|e| CoreError::Memory(format!("LMDB put cold_storage failed: {e}")))?;
1999        tx.commit()
2000            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
2001        self.mutation_count.fetch_add(1, Ordering::Relaxed);
2002        Ok(())
2003    }
2004
2005    /// Retrieve a compressed cold record by memory id.
2006    pub fn get_cold_record(&self, id: MemoryId) -> Result<Option<crate::cold_storage::ColdRecord>> {
2007        let key = id.as_bytes();
2008        let tx = self
2009            .env
2010            .begin_ro_txn()
2011            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
2012        match tx.get(self.cold_storage_db, key) {
2013            Ok(bytes) => {
2014                let record: crate::cold_storage::ColdRecord = rmp_serde::from_slice(bytes)
2015                    .map_err(|e| {
2016                        CoreError::Memory(format!("Cold record deserialization failed: {e}"))
2017                    })?;
2018                Ok(Some(record))
2019            }
2020            Err(lmdb::Error::NotFound) => Ok(None),
2021            Err(e) => Err(CoreError::Memory(format!(
2022                "LMDB get cold_storage failed: {e}"
2023            ))),
2024        }
2025    }
2026
2027    /// Delete a cold record from the cold storage DBI (used when thawing back to hot).
2028    pub fn delete_cold_record(&self, id: MemoryId) -> Result<bool> {
2029        let key = id.as_bytes();
2030        let mut tx = self
2031            .env
2032            .begin_rw_txn()
2033            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
2034        let deleted = match tx.del(self.cold_storage_db, key, None) {
2035            Ok(()) => true,
2036            Err(lmdb::Error::NotFound) => false,
2037            Err(e) => {
2038                return Err(CoreError::Memory(format!(
2039                    "LMDB del cold_storage failed: {e}"
2040                )));
2041            }
2042        };
2043        tx.commit()
2044            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
2045        if deleted {
2046            self.mutation_count.fetch_add(1, Ordering::Relaxed);
2047        }
2048        Ok(deleted)
2049    }
2050
2051    /// Count cold records in the cold storage database, optionally filtered by galaxy.
2052    pub fn count_cold(&self, galaxy: Option<Galaxy>) -> Result<usize> {
2053        let tx = self
2054            .env
2055            .begin_ro_txn()
2056            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
2057        let mut cursor = tx
2058            .open_ro_cursor(self.cold_storage_db)
2059            .map_err(|e| CoreError::Memory(format!("LMDB open_ro_cursor failed: {e}")))?;
2060        let mut count = 0;
2061        for (_key, val) in cursor.iter() {
2062            if let Some(target_g) = galaxy {
2063                let record: crate::cold_storage::ColdRecord =
2064                    rmp_serde::from_slice(val).map_err(|e| {
2065                        CoreError::Memory(format!("Cold record deserialization failed: {e}"))
2066                    })?;
2067                if record.galaxy == target_g {
2068                    count += 1;
2069                }
2070            } else {
2071                count += 1;
2072            }
2073        }
2074        Ok(count)
2075    }
2076
2077    /// List cold records (summaries) with optional galaxy filter and limit.
2078    pub fn list_cold_records(
2079        &self,
2080        galaxy: Option<Galaxy>,
2081        limit: usize,
2082    ) -> Result<Vec<crate::cold_storage::ColdRecordSummary>> {
2083        let query = crate::cold_storage::ColdQuery {
2084            galaxy,
2085            limit: if limit == 0 { 100 } else { limit },
2086            ..Default::default()
2087        };
2088        self.query_cold_records(&query)
2089    }
2090
2091    /// Query cold records matching a `ColdQuery` filter.
2092    pub fn query_cold_records(
2093        &self,
2094        query: &crate::cold_storage::ColdQuery,
2095    ) -> Result<Vec<crate::cold_storage::ColdRecordSummary>> {
2096        let tx = self
2097            .env
2098            .begin_ro_txn()
2099            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
2100        let mut cursor = tx
2101            .open_ro_cursor(self.cold_storage_db)
2102            .map_err(|e| CoreError::Memory(format!("LMDB open_ro_cursor failed: {e}")))?;
2103        let mut results = Vec::new();
2104        let limit = if query.limit == 0 {
2105            usize::MAX
2106        } else {
2107            query.limit
2108        };
2109
2110        for (_key, val) in cursor.iter() {
2111            let record: crate::cold_storage::ColdRecord =
2112                rmp_serde::from_slice(val).map_err(|e| {
2113                    CoreError::Memory(format!("Cold record deserialization failed: {e}"))
2114                })?;
2115            let summary = record.summary();
2116            if query.matches(&summary) {
2117                results.push(summary);
2118                if results.len() >= limit {
2119                    break;
2120                }
2121            }
2122        }
2123        Ok(results)
2124    }
2125
2126    /// Bounded, identity-bound cold discovery.
2127    ///
2128    /// Scans at most `max_scan` cold records (LMDB `cold_storage` DBI),
2129    /// filters by galaxy when given, decompresses each candidate, verifies
2130    /// the id/galaxy/content-hash chain, applies visibility (private never
2131    /// surfaces; superseded/non-current records are skipped), and returns
2132    /// up to `limit` full cold records whose content or tags contain every
2133    /// query term (case-insensitive). Nothing is thawed or mutated.
2134    pub fn find_cold_matching(
2135        &self,
2136        terms: &[String],
2137        galaxy: Option<Galaxy>,
2138        limit: usize,
2139        max_scan: usize,
2140    ) -> Result<crate::cold_storage::ColdDiscoveryOutcome> {
2141        self.find_cold_matching_eligible(terms, galaxy, limit, max_scan, |_| true)
2142    }
2143
2144    /// Apply caller eligibility to verified payloads before consuming result
2145    /// capacity. Rejected matches still consume the bounded scan budget.
2146    pub fn find_cold_matching_eligible(
2147        &self,
2148        terms: &[String],
2149        galaxy: Option<Galaxy>,
2150        limit: usize,
2151        max_scan: usize,
2152        eligible: impl Fn(&Memory) -> bool,
2153    ) -> Result<crate::cold_storage::ColdDiscoveryOutcome> {
2154        use crate::cold_storage::ColdDiscoveryStop;
2155        let mut out = crate::cold_storage::ColdDiscoveryOutcome::default();
2156        if terms.is_empty() || limit == 0 || max_scan == 0 {
2157            return Ok(out);
2158        }
2159        let tx = self
2160            .env
2161            .begin_ro_txn()
2162            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
2163        let mut cursor = tx
2164            .open_ro_cursor(self.cold_storage_db)
2165            .map_err(|e| CoreError::Memory(format!("LMDB open_ro_cursor failed: {e}")))?;
2166        let mut iter = cursor.iter();
2167        loop {
2168            if out.scanned >= max_scan {
2169                out.stop_reason = ColdDiscoveryStop::ScanLimit;
2170                break;
2171            }
2172            if out.records.len() >= limit {
2173                out.stop_reason = ColdDiscoveryStop::ResultLimit;
2174                break;
2175            }
2176            let Some((key, val)) = iter.next() else {
2177                out.stop_reason = ColdDiscoveryStop::Exhausted;
2178                break;
2179            };
2180            out.scanned += 1;
2181            let record: crate::cold_storage::ColdRecord = if let Ok(r) = rmp_serde::from_slice(val)
2182            {
2183                r
2184            } else {
2185                out.integrity_rejected += 1;
2186                continue;
2187            };
2188            if let Some(g) = galaxy {
2189                if record.galaxy != g {
2190                    continue;
2191                }
2192            }
2193            out.candidates += 1;
2194            let mem = if let Ok(m) = record.decompress() {
2195                m
2196            } else {
2197                out.integrity_rejected += 1;
2198                continue;
2199            };
2200            if mem.metadata.is_private {
2201                out.private_skipped += 1;
2202                continue;
2203            }
2204            if !mem.metadata.validity.is_current() {
2205                out.non_current_skipped += 1;
2206                continue;
2207            }
2208            let integrity_ok = key == record.id.as_bytes()
2209                && mem.metadata.id == record.id
2210                && mem.metadata.galaxy == record.galaxy
2211                && mem.metadata.content_hash == record.content_hash
2212                && crate::content_hash(&mem.content) == record.content_hash;
2213            if !integrity_ok {
2214                out.integrity_rejected += 1;
2215                continue;
2216            }
2217            let haystack = format!(
2218                "{} {}",
2219                mem.content.to_lowercase(),
2220                mem.metadata.tags.join(" ").to_lowercase()
2221            );
2222            if !terms.iter().all(|t| haystack.contains(t.as_str())) {
2223                continue;
2224            }
2225            out.matched += 1;
2226            if !eligible(&mem) {
2227                out.eligibility_skipped += 1;
2228                continue;
2229            }
2230            out.records.push(record);
2231        }
2232        Ok(out)
2233    }
2234
2235    /// Freeze an active hot memory into the compressed cold archive.
2236    ///
2237    /// Non-destructive: preserves complete metadata, vector clocks, content, embeddings,
2238    /// and provenance. The memory transitions to `Tier::Archival`, is stored in `cold_storage_db`,
2239    /// is deindexed from Tantivy (if `search` provided), and is removed from the active hot galaxy.
2240    pub fn freeze_to_cold(
2241        &self,
2242        search: Option<&crate::SearchEngine>,
2243        memory_id: MemoryId,
2244        distance: f32,
2245        factors: crate::cold_storage::OuterRimFactors,
2246        digest_id: Option<MemoryId>,
2247        notes: Option<String>,
2248        codec: crate::cold_storage::CompressionCodec,
2249    ) -> Result<crate::cold_storage::ColdRecord> {
2250        let (galaxy, mut mem) = self.find_across_galaxies(memory_id)?.ok_or_else(|| {
2251            CoreError::NotFound(format!("Memory {memory_id} not found in hot store"))
2252        })?;
2253
2254        // Transition tier to Archival
2255        if mem.metadata.tier != crate::memory::Tier::Archival {
2256            let _ = mem.transition_tier(crate::memory::Tier::Archival);
2257        }
2258
2259        let record =
2260            crate::cold_storage::ColdRecord::new(&mem, distance, factors, digest_id, notes, codec)?;
2261
2262        // Store into cold archive
2263        self.put_cold_record(&record)?;
2264
2265        // Remove from active hot galaxy
2266        self.delete(galaxy, memory_id)?;
2267
2268        // Deindex from Tantivy search engine if provided
2269        if let Some(engine) = search {
2270            if let Ok(mut writer_guard) = engine.writer() {
2271                let _ = engine.delete_document(&mut writer_guard, &memory_id.to_string());
2272                let _ = engine.commit(&mut writer_guard);
2273            }
2274        }
2275
2276        Ok(record)
2277    }
2278
2279    /// Thaw a memory from compressed cold storage back into the hot active tier.
2280    ///
2281    /// Zero data loss: restores the original memory with all fields, transitions tier
2282    /// back to `Tier::Episodic`, bumps access/recall count, updates `accessed_at`,
2283    /// stores into the hot galaxy, and reindexes into Tantivy search (if provided).
2284    pub fn thaw_from_cold(
2285        &self,
2286        search: Option<&crate::SearchEngine>,
2287        memory_id: MemoryId,
2288    ) -> Result<Memory> {
2289        let record = self.get_cold_record(memory_id)?.ok_or_else(|| {
2290            CoreError::NotFound(format!("Memory {memory_id} not found in cold storage"))
2291        })?;
2292
2293        let mut mem = record.decompress()?;
2294
2295        // Transition tier back to Episodic (warm serving)
2296        let _ = mem.transition_tier(crate::memory::Tier::Episodic);
2297        mem.metadata.accessed_at = chrono::Utc::now();
2298        mem.metadata.access_count += 1;
2299        mem.metadata.recall_count += 1;
2300        if !mem.metadata.tags.iter().any(|t| t == "thawed:phagic") {
2301            mem.metadata.tags.push("thawed:phagic".to_string());
2302        }
2303
2304        // Put back into active hot galaxy
2305        self.put(record.galaxy, &mem)?;
2306
2307        // Reindex in Tantivy if provided
2308        if let Some(engine) = search {
2309            if let Ok(mut writer_guard) = engine.writer() {
2310                let _ = engine.index_memory(&mut writer_guard, &mem);
2311                let _ = engine.commit(&mut writer_guard);
2312            }
2313        }
2314
2315        // Remove from cold archive
2316        self.delete_cold_record(memory_id)?;
2317
2318        Ok(mem)
2319    }
2320
2321    /// Find a memory anywhere: in the active hot galaxies, or decompressed from cold storage.
2322    ///
2323    /// Returns `(galaxy, memory, is_cold)`.
2324    pub fn find_anywhere(&self, id: MemoryId) -> Result<Option<(Galaxy, Memory, bool)>> {
2325        if let Some((galaxy, mem)) = self.find_across_galaxies(id)? {
2326            return Ok(Some((galaxy, mem, false)));
2327        }
2328        if let Some(cold_record) = self.get_cold_record(id)? {
2329            let galaxy = cold_record.galaxy;
2330            let mem = cold_record.decompress()?;
2331            return Ok(Some((galaxy, mem, true)));
2332        }
2333        Ok(None)
2334    }
2335}
2336
2337#[cfg(test)]
2338mod tests {
2339    use super::*;
2340    use crate::content_hash;
2341
2342    #[test]
2343    fn open_and_create_galaxies() {
2344        let tmp = tempfile::tempdir().unwrap();
2345        let store = MemoryStore::open_default(tmp.path()).unwrap();
2346        for galaxy in Galaxy::all() {
2347            let _db = store.galaxy_db(galaxy).unwrap();
2348        }
2349    }
2350
2351    #[test]
2352    fn ensure_schema_completes_a_pre_cold_store() {
2353        use lmdb::{DatabaseFlags as LmdbFlags, Environment as LmdbEnv};
2354        use uuid::Uuid;
2355
2356        // Synthetic pre-cold store: every DBI a 9.0.0 store had, but no
2357        // `cold_storage` (the exact 2026-09-14 restore-drill finding).
2358        let tmp = tempfile::tempdir().unwrap();
2359        let path = tmp.path().join("old-store");
2360        std::fs::create_dir_all(&path).unwrap();
2361        {
2362            let env = LmdbEnv::new().set_max_dbs(64).open(&path).unwrap();
2363            for galaxy in Galaxy::all() {
2364                env.create_db(Some(galaxy.db_name()), LmdbFlags::default())
2365                    .unwrap();
2366            }
2367            for (name, flags) in crate::indexes::INDEX_DBS {
2368                env.create_db(Some(name), *flags).unwrap();
2369            }
2370            for (name, flags) in [
2371                ("episodic_records", LmdbFlags::default()),
2372                ("episodic_terms_v2", LmdbFlags::DUP_SORT),
2373                ("embedding_cache", LmdbFlags::default()),
2374                ("revisions", LmdbFlags::default()),
2375                (crate::attestation::ATTESTATIONS_DB, LmdbFlags::default()),
2376            ] {
2377                env.create_db(Some(name), flags).unwrap();
2378            }
2379        }
2380
2381        let files_before: Vec<String> = {
2382            let mut names: Vec<String> = std::fs::read_dir(&path)
2383                .unwrap()
2384                .map(|e| e.unwrap().file_name().to_string_lossy().to_string())
2385                .collect();
2386            names.sort();
2387            names
2388        };
2389        let data_before = std::fs::read(path.join("data.mdb")).unwrap();
2390
2391        let error = match MemoryStore::open_readonly(&path) {
2392            Ok(_) => panic!("strict read-only open must refuse an incomplete store"),
2393            Err(e) => e.to_string(),
2394        };
2395        assert!(error.contains("cold_storage"), "{error}");
2396        assert_eq!(
2397            std::fs::read(path.join("data.mdb")).unwrap(),
2398            data_before,
2399            "readonly refusal must not mutate the pre-cold store"
2400        );
2401        let files_after: Vec<String> = {
2402            let mut names: Vec<String> = std::fs::read_dir(&path)
2403                .unwrap()
2404                .map(|e| e.unwrap().file_name().to_string_lossy().to_string())
2405                .collect();
2406            names.sort();
2407            names
2408        };
2409        assert_eq!(
2410            files_after, files_before,
2411            "readonly refusal changed the store directory"
2412        );
2413
2414        let created = MemoryStore::ensure_schema(&path).unwrap();
2415        assert_eq!(created, vec!["cold_storage".to_string()], "{created:?}");
2416
2417        let store = MemoryStore::open_readonly(&path).unwrap();
2418        assert!(store.get_cold_record(Uuid::nil()).unwrap().is_none());
2419
2420        // Idempotent: a complete store reports nothing missing.
2421        assert!(MemoryStore::ensure_schema(&path).unwrap().is_empty());
2422    }
2423
2424    #[test]
2425    fn cold_discovery_hydrates_verifies_and_respects_visibility() {
2426        use crate::cold_storage::{ColdRecord, CompressionCodec, OuterRimFactors};
2427        let tmp = tempfile::tempdir().unwrap();
2428        let store = MemoryStore::open_default(tmp.path()).unwrap();
2429        let factors = OuterRimFactors {
2430            age_factor: 0.5,
2431            access_factor: 0.5,
2432            resonance_factor: 0.5,
2433            emotional_factor: 0.5,
2434            importance_factor: 0.5,
2435            distance: 0.5,
2436        };
2437
2438        let pub_mem = Memory::new(
2439            Galaxy::Codex,
2440            "public needle zxquniquecoldfact741 buried".into(),
2441        );
2442        let rec = ColdRecord::new(
2443            &pub_mem,
2444            0.5,
2445            factors.clone(),
2446            None,
2447            None,
2448            CompressionCodec::Gzip,
2449        )
2450        .unwrap();
2451        store.put_cold_record(&rec).unwrap();
2452        let out = store
2453            .find_cold_matching(&["zxquniquecoldfact741".to_string()], None, 10, 100)
2454            .unwrap();
2455        assert_eq!(out.matched, 1);
2456        assert_eq!(out.integrity_rejected, 0);
2457        assert_eq!(out.records.len(), 1);
2458
2459        // Private originals never surface over MCP discovery.
2460        let mut priv_mem = Memory::new(Galaxy::Codex, "private needle zxquniquecoldfact742".into());
2461        priv_mem.metadata.is_private = true;
2462        store
2463            .put_cold_record(
2464                &ColdRecord::new(&priv_mem, 0.5, factors, None, None, CompressionCodec::Gzip)
2465                    .unwrap(),
2466            )
2467            .unwrap();
2468        let out_priv = store
2469            .find_cold_matching(&["zxquniquecoldfact742".to_string()], None, 10, 100)
2470            .unwrap();
2471        assert_eq!(out_priv.matched, 0);
2472        assert_eq!(out_priv.private_skipped, 1);
2473
2474        // Tamper: the payload decompresses to content that no longer matches
2475        // the advertised content hash — refuse, never return.
2476        let mut tampered = rec.clone();
2477        let mut bad = Memory::new(Galaxy::Codex, "tampered needle zxquniquecoldfact743".into());
2478        bad.metadata.id = rec.id;
2479        let (payload, size) =
2480            crate::cold_storage::compress_memory(&bad, CompressionCodec::Gzip).unwrap();
2481        tampered.compressed_payload = payload;
2482        tampered.uncompressed_size = size;
2483        store.put_cold_record(&tampered).unwrap();
2484        let out_tamper = store
2485            .find_cold_matching(&["zxquniquecoldfact743".to_string()], None, 10, 100)
2486            .unwrap();
2487        assert_eq!(out_tamper.matched, 0);
2488        assert_eq!(out_tamper.integrity_rejected, 1);
2489
2490        // Even an internally consistent header/payload must not be accepted
2491        // under another physical UUID key (which would break known-ID read).
2492        store.delete_cold_record(rec.id).unwrap();
2493        let wrong_key = uuid::Uuid::from_u128(741);
2494        assert_ne!(wrong_key, rec.id);
2495        let value = rmp_serde::to_vec_named(&rec).unwrap();
2496        let mut tx = store.env.begin_rw_txn().unwrap();
2497        tx.put(
2498            store.cold_storage_db,
2499            wrong_key.as_bytes(),
2500            &value,
2501            WriteFlags::default(),
2502        )
2503        .unwrap();
2504        tx.commit().unwrap();
2505        let wrong_key_out = store
2506            .find_cold_matching(&["zxquniquecoldfact741".into()], None, 10, 100)
2507            .unwrap();
2508        assert!(wrong_key_out.records.is_empty());
2509        assert_eq!(wrong_key_out.integrity_rejected, 1);
2510    }
2511
2512    /// Substring filter (memory.query trap fix, 2026-08-29): literal
2513    /// case-insensitive content match, galaxy-wide — never an arbitrary
2514    /// page, never routed through the indexed fast paths.
2515    #[test]
2516    fn query_substring_filters_galaxy_wide() {
2517        let tmp = tempfile::tempdir().unwrap();
2518        let store = MemoryStore::open_default(tmp.path()).unwrap();
2519
2520        for (i, content) in [
2521            "the mesh joins at dawn",
2522            "unrelated content entirely",
2523            "MESH joins at dusk",
2524        ]
2525        .iter()
2526        .enumerate()
2527        {
2528            let mut m = Memory::new(Galaxy::Codex, content.to_string());
2529            m.metadata.importance = 0.5 + i as f32 / 10.0;
2530            store.put(Galaxy::Codex, &m).unwrap();
2531        }
2532
2533        let hits = store
2534            .query(
2535                Galaxy::Codex,
2536                &MemoryQuery::new().with_content_substring("mesh joins"),
2537            )
2538            .unwrap();
2539        assert_eq!(hits.len(), 2, "CI substring must match both: {hits:?}");
2540        assert!(
2541            hits.iter()
2542                .all(|m| m.content.to_lowercase().contains("mesh joins"))
2543        );
2544
2545        let none = store
2546            .query(
2547                Galaxy::Codex,
2548                &MemoryQuery::new().with_content_substring("quantum calendar"),
2549            )
2550            .unwrap();
2551        assert!(none.is_empty(), "no match must be an honest empty set");
2552
2553        // Substring + tag combined still applies (no fast-path bypass).
2554        let mut tagged = Memory::new(Galaxy::Codex, "mesh joins again".to_string());
2555        tagged.metadata.tags = vec!["mesh".into()];
2556        store.put(Galaxy::Codex, &tagged).unwrap();
2557        let combined = store
2558            .query(
2559                Galaxy::Codex,
2560                &MemoryQuery::new()
2561                    .with_tags(vec!["mesh".into()])
2562                    .with_content_substring("again"),
2563            )
2564            .unwrap();
2565        assert_eq!(combined.len(), 1);
2566        assert_eq!(combined[0].content, "mesh joins again");
2567    }
2568
2569    #[cfg(unix)]
2570    #[test]
2571    fn store_dir_has_restrictive_permissions() {
2572        let tmp = tempfile::tempdir().unwrap();
2573        let store_path = tmp.path().join("lmdb");
2574        let _store = MemoryStore::open_default(&store_path).unwrap();
2575        let perms = std::fs::metadata(&store_path).unwrap().permissions().mode();
2576        assert_eq!(
2577            perms & 0o777,
2578            0o700,
2579            "store directory should have 0700 permissions, got {:o}",
2580            perms & 0o777
2581        );
2582    }
2583
2584    #[test]
2585    fn put_get_delete_memory() {
2586        let tmp = tempfile::tempdir().unwrap();
2587        let store = MemoryStore::open_default(tmp.path()).unwrap();
2588
2589        let mem = Memory::new(Galaxy::Codex, "Hello world".to_string());
2590        let id = mem.metadata.id;
2591
2592        store.put(Galaxy::Codex, &mem).unwrap();
2593        let retrieved = store.get(Galaxy::Codex, id).unwrap();
2594        assert!(retrieved.is_some());
2595        assert_eq!(retrieved.unwrap().content, "Hello world");
2596
2597        let deleted = store.delete(Galaxy::Codex, id).unwrap();
2598        assert!(deleted);
2599
2600        let gone = store.get(Galaxy::Codex, id).unwrap();
2601        assert!(gone.is_none());
2602    }
2603
2604    #[test]
2605    fn scan_memories() {
2606        let tmp = tempfile::tempdir().unwrap();
2607        let store = MemoryStore::open_default(tmp.path()).unwrap();
2608
2609        for i in 0..5 {
2610            let mem = Memory::new(Galaxy::Codex, format!("memory-{i}"));
2611            store.put(Galaxy::Codex, &mem).unwrap();
2612        }
2613
2614        let all = store.scan(Galaxy::Codex, 100).unwrap();
2615        assert_eq!(all.len(), 5);
2616
2617        let limited = store.scan(Galaxy::Codex, 3).unwrap();
2618        assert_eq!(limited.len(), 3);
2619    }
2620
2621    #[test]
2622    fn overwrite_removes_stale_index_entries() {
2623        let tmp = tempfile::tempdir().unwrap();
2624        let store = MemoryStore::open_default(tmp.path()).unwrap();
2625
2626        // Original record: tag "alpha", importance 0.9.
2627        let mut mem = Memory::new(Galaxy::Codex, "overwrite target".to_string());
2628        mem.metadata.tags = vec!["alpha".to_string()];
2629        mem.metadata.importance = 0.9;
2630        let id = mem.metadata.id;
2631        store.put(Galaxy::Codex, &mem).unwrap();
2632
2633        // Overwrite with tag "beta", importance 0.1, new content hash.
2634        let mut updated = Memory::new(Galaxy::Codex, "overwritten content".to_string());
2635        updated.metadata.id = id;
2636        updated.metadata.tags = vec!["beta".to_string()];
2637        updated.metadata.importance = 0.1;
2638        store.put(Galaxy::Codex, &updated).unwrap();
2639
2640        // Stale entries must be gone, new entries must be queryable.
2641        let tx = store.env().begin_ro_txn().unwrap();
2642        let by_alpha = store
2643            .index_dbs()
2644            .find_by_tag(&tx, Galaxy::Codex, "alpha")
2645            .unwrap();
2646        let by_beta = store
2647            .index_dbs()
2648            .find_by_tag(&tx, Galaxy::Codex, "beta")
2649            .unwrap();
2650        assert!(
2651            by_alpha.is_empty(),
2652            "stale tag index entries must be removed on overwrite"
2653        );
2654        assert_eq!(by_beta, vec![id]);
2655
2656        let by_importance = store
2657            .index_dbs()
2658            .find_by_importance_range(&tx, Galaxy::Codex, 0.0, 0.2)
2659            .unwrap();
2660        assert!(
2661            by_importance.contains(&id),
2662            "new importance must be indexed"
2663        );
2664        let by_high = store
2665            .index_dbs()
2666            .find_by_importance_range(&tx, Galaxy::Codex, 0.8, 1.0)
2667            .unwrap();
2668        assert!(
2669            !by_high.contains(&id),
2670            "stale importance index entries must be removed on overwrite"
2671        );
2672
2673        let old_hash = content_hash("overwrite target");
2674        let new_hash = content_hash("overwritten content");
2675        assert_eq!(
2676            store
2677                .index_dbs()
2678                .find_by_content_hash(&tx, Galaxy::Codex, &old_hash)
2679                .unwrap(),
2680            None,
2681            "stale content-hash index entry must be removed"
2682        );
2683        assert_eq!(
2684            store
2685                .index_dbs()
2686                .find_by_content_hash(&tx, Galaxy::Codex, &new_hash)
2687                .unwrap(),
2688            Some(id)
2689        );
2690    }
2691
2692    #[test]
2693    fn count_memories() {
2694        let tmp = tempfile::tempdir().unwrap();
2695        let store = MemoryStore::open_default(tmp.path()).unwrap();
2696
2697        assert_eq!(store.count(Galaxy::Codex).unwrap(), 0);
2698
2699        for i in 0..3 {
2700            let mem = Memory::new(Galaxy::Codex, format!("count-{i}"));
2701            store.put(Galaxy::Codex, &mem).unwrap();
2702        }
2703
2704        assert_eq!(store.count(Galaxy::Codex).unwrap(), 3);
2705    }
2706
2707    /// `count_by_tag` uses the tag index and counts distinct records — the
2708    /// status surface reports logical sessions with it (one `start` tag per
2709    /// session, ignoring turns/checkpoints).
2710    #[test]
2711    fn count_by_tag_counts_indexed_records() {
2712        let tmp = tempfile::tempdir().unwrap();
2713        let store = MemoryStore::open_default(tmp.path()).unwrap();
2714
2715        assert_eq!(store.count_by_tag(Galaxy::Sessions, "start").unwrap(), 0);
2716
2717        let mut start = Memory::new(Galaxy::Sessions, "{\"type\":\"session_start\"}".into());
2718        start.metadata.tags = vec!["session".into(), "start".into()];
2719        store.put(Galaxy::Sessions, &start).unwrap();
2720        for i in 0..2 {
2721            let mut turn =
2722                Memory::new(Galaxy::Sessions, format!("{{\"type\":\"turn\",\"i\":{i}}}"));
2723            turn.metadata.tags = vec!["session".into(), "turn".into()];
2724            store.put(Galaxy::Sessions, &turn).unwrap();
2725        }
2726
2727        assert_eq!(store.count(Galaxy::Sessions).unwrap(), 3);
2728        assert_eq!(store.count_by_tag(Galaxy::Sessions, "start").unwrap(), 1);
2729        assert_eq!(store.count_by_tag(Galaxy::Sessions, "turn").unwrap(), 2);
2730        assert_eq!(store.count_by_tag(Galaxy::Sessions, "absent").unwrap(), 0);
2731    }
2732
2733    #[test]
2734    fn get_nonexistent_returns_none() {
2735        let tmp = tempfile::tempdir().unwrap();
2736        let store = MemoryStore::open_default(tmp.path()).unwrap();
2737        let result = store.get(Galaxy::Codex, uuid::Uuid::new_v4()).unwrap();
2738        assert!(result.is_none());
2739    }
2740
2741    #[test]
2742    fn raw_put_get() {
2743        let tmp = tempfile::tempdir().unwrap();
2744        let store = MemoryStore::open_default(tmp.path()).unwrap();
2745
2746        store
2747            .put_raw(Galaxy::Substrate, b"config:key", b"value123")
2748            .unwrap();
2749        let val = store.get_raw(Galaxy::Substrate, b"config:key").unwrap();
2750        assert_eq!(val, Some(b"value123".to_vec()));
2751    }
2752
2753    #[test]
2754    fn put_dedup_prevents_duplicates() {
2755        let tmp = tempfile::tempdir().unwrap();
2756        let store = MemoryStore::open_default(tmp.path()).unwrap();
2757
2758        let mem1 = Memory::new(Galaxy::Codex, "duplicate content".into());
2759        let id1 = store.put_dedup(Galaxy::Codex, &mem1).unwrap();
2760
2761        let mem2 = Memory::new(Galaxy::Codex, "duplicate content".into());
2762        let id2 = store.put_dedup(Galaxy::Codex, &mem2).unwrap();
2763
2764        assert_eq!(id1, id2, "dedup should return same ID for same content");
2765        assert_eq!(store.count(Galaxy::Codex).unwrap(), 1);
2766    }
2767
2768    #[test]
2769    fn put_dedup_allows_different_content() {
2770        let tmp = tempfile::tempdir().unwrap();
2771        let store = MemoryStore::open_default(tmp.path()).unwrap();
2772
2773        let mem1 = Memory::new(Galaxy::Codex, "content A".into());
2774        store.put_dedup(Galaxy::Codex, &mem1).unwrap();
2775
2776        let mem2 = Memory::new(Galaxy::Codex, "content B".into());
2777        store.put_dedup(Galaxy::Codex, &mem2).unwrap();
2778
2779        assert_eq!(store.count(Galaxy::Codex).unwrap(), 2);
2780    }
2781
2782    #[test]
2783    fn put_batch_atomic_write() {
2784        let tmp = tempfile::tempdir().unwrap();
2785        let store = MemoryStore::open_default(tmp.path()).unwrap();
2786
2787        let memories: Vec<Memory> = (0..10)
2788            .map(|i| Memory::new(Galaxy::Codex, format!("batch-{i}")))
2789            .collect();
2790
2791        store.put_batch(Galaxy::Codex, &memories).unwrap();
2792        assert_eq!(store.count(Galaxy::Codex).unwrap(), 10);
2793    }
2794
2795    #[test]
2796    fn query_by_tags() {
2797        let tmp = tempfile::tempdir().unwrap();
2798        let store = MemoryStore::open_default(tmp.path()).unwrap();
2799
2800        let mem1 = Memory::new(Galaxy::Codex, "tagged memory".into())
2801            .with_tags(vec!["rust".into(), "memory".into()]);
2802        let mem2 =
2803            Memory::new(Galaxy::Codex, "other memory".into()).with_tags(vec!["python".into()]);
2804        store.put(Galaxy::Codex, &mem1).unwrap();
2805        store.put(Galaxy::Codex, &mem2).unwrap();
2806
2807        let query = MemoryQuery::new().with_tags(vec!["rust".into()]);
2808        let results = store.query(Galaxy::Codex, &query).unwrap();
2809        assert_eq!(results.len(), 1);
2810        assert_eq!(results[0].content, "tagged memory");
2811    }
2812
2813    #[test]
2814    fn query_by_importance_range() {
2815        let tmp = tempfile::tempdir().unwrap();
2816        let store = MemoryStore::open_default(tmp.path()).unwrap();
2817
2818        store
2819            .put(
2820                Galaxy::Codex,
2821                &Memory::new(Galaxy::Codex, "low".into()).with_importance(0.1),
2822            )
2823            .unwrap();
2824        store
2825            .put(
2826                Galaxy::Codex,
2827                &Memory::new(Galaxy::Codex, "mid".into()).with_importance(0.5),
2828            )
2829            .unwrap();
2830        store
2831            .put(
2832                Galaxy::Codex,
2833                &Memory::new(Galaxy::Codex, "high".into()).with_importance(0.9),
2834            )
2835            .unwrap();
2836
2837        let query = MemoryQuery::new().with_importance_range(0.4, 0.6);
2838        let results = store.query(Galaxy::Codex, &query).unwrap();
2839        assert_eq!(results.len(), 1);
2840        assert_eq!(results[0].content, "mid");
2841    }
2842
2843    #[test]
2844    fn memory_query_one_sided_time_bounds() {
2845        // `created_after` / `created_before` map onto the temporal filter
2846        // one side at a time (the API passthrough for memory.query).
2847        let old = Memory::new(Galaxy::Codex, "old".into());
2848        let mut recent = Memory::new(Galaxy::Codex, "recent".into());
2849        recent.metadata.created_at = old.metadata.created_at + chrono::Duration::days(30);
2850
2851        let cutoff = old.metadata.created_at + chrono::Duration::days(10);
2852        let after = MemoryQuery::new().with_created_after(cutoff);
2853        assert!(!after.matches(&old), "pre-cutoff memory must not match");
2854        assert!(after.matches(&recent), "post-cutoff memory must match");
2855
2856        let before = MemoryQuery::new().with_created_before(cutoff);
2857        assert!(before.matches(&old), "pre-cutoff memory must match");
2858        assert!(
2859            !before.matches(&recent),
2860            "post-cutoff memory must not match"
2861        );
2862
2863        // Bounds are inclusive.
2864        let edge = MemoryQuery::new().with_created_after(cutoff);
2865        let mut at = Memory::new(Galaxy::Codex, "at cutoff".into());
2866        at.metadata.created_at = cutoff;
2867        assert!(edge.matches(&at), "created_at == after bound is inclusive");
2868    }
2869
2870    #[test]
2871    fn embedding_put_get_delete() {
2872        let tmp = tempfile::tempdir().unwrap();
2873        let store = MemoryStore::open_default(tmp.path()).unwrap();
2874
2875        let id = uuid::Uuid::new_v4();
2876        let embedding = vec![0.1, 0.2, 0.3, 0.4, 0.5];
2877
2878        store.put_embedding(id, &embedding).unwrap();
2879        let retrieved = store.get_embedding(id).unwrap();
2880        assert!(retrieved.is_some());
2881        let retrieved = retrieved.unwrap();
2882        assert_eq!(retrieved.len(), 5);
2883        assert!((retrieved[0] - 0.1).abs() < f32::EPSILON);
2884
2885        assert!(store.delete_embedding(id).unwrap());
2886        assert!(store.get_embedding(id).unwrap().is_none());
2887    }
2888
2889    #[test]
2890    fn embedding_cache_roundtrip_batch_and_count() {
2891        let tmp = tempfile::tempdir().unwrap();
2892        let store = MemoryStore::open_default(tmp.path()).unwrap();
2893
2894        let entries: Vec<(String, Vec<f32>)> = (0..5)
2895            .map(|i| (format!("ns:model:{i:016x}"), vec![i as f32; 8]))
2896            .collect();
2897        store.put_embedding_cache_batch(&entries).unwrap();
2898        assert_eq!(store.embedding_cache_count().unwrap(), 5);
2899
2900        // Single read
2901        let hit = store
2902            .get_embedding_cache("ns:model:0000000000000003")
2903            .unwrap();
2904        assert_eq!(hit.unwrap(), vec![3.0f32; 8]);
2905        assert!(
2906            store
2907                .get_embedding_cache("ns:model:absent")
2908                .unwrap()
2909                .is_none()
2910        );
2911
2912        // Batched read aligns 1:1, misses are None
2913        let keys: Vec<String> = (0..6).map(|i| format!("ns:model:{i:016x}")).collect();
2914        let batch = store.get_embedding_cache_batch(&keys).unwrap();
2915        assert_eq!(batch.len(), 6);
2916        assert!(batch[0..5].iter().all(Option::is_some));
2917        assert!(batch[5].is_none());
2918
2919        // Overwrite is a put, not a duplicate
2920        store
2921            .put_embedding_cache("ns:model:0000000000000001", &[9.0; 8])
2922            .unwrap();
2923        assert_eq!(store.embedding_cache_count().unwrap(), 5);
2924        assert_eq!(
2925            store
2926                .get_embedding_cache("ns:model:0000000000000001")
2927                .unwrap()
2928                .unwrap(),
2929            vec![9.0f32; 8]
2930        );
2931    }
2932
2933    #[test]
2934    fn embedding_cache_survives_store_reopen() {
2935        // V8 ship list #2 acceptance shape: vectors persist across restart.
2936        let tmp = tempfile::tempdir().unwrap();
2937        {
2938            let store = MemoryStore::open_default(tmp.path()).unwrap();
2939            store
2940                .put_embedding_cache("onnx:bge-small:abc", &[0.5; 384])
2941                .unwrap();
2942        }
2943        let reopened = MemoryStore::open_default(tmp.path()).unwrap();
2944        let cached = reopened.get_embedding_cache("onnx:bge-small:abc").unwrap();
2945        assert_eq!(cached.unwrap(), vec![0.5f32; 384]);
2946    }
2947
2948    #[test]
2949    fn content_hash_is_sha256() {
2950        let hash1 = content_hash("test content");
2951        let hash2 = content_hash("test content");
2952        let hash3 = content_hash("different content");
2953
2954        assert_eq!(hash1, hash2, "same content should produce same hash");
2955        assert_ne!(
2956            hash1, hash3,
2957            "different content should produce different hash"
2958        );
2959        assert_eq!(hash1.len(), 64, "SHA-256 hex should be 64 chars");
2960    }
2961
2962    #[test]
2963    fn query_by_tag_uses_index() {
2964        let tmp = tempfile::tempdir().unwrap();
2965        let store = MemoryStore::open_default(tmp.path()).unwrap();
2966
2967        let mem1 = Memory::new(Galaxy::Codex, "tagged".into())
2968            .with_tags(vec!["rust".into(), "memory".into()]);
2969        let mem2 = Memory::new(Galaxy::Codex, "other".into()).with_tags(vec!["python".into()]);
2970        store.put(Galaxy::Codex, &mem1).unwrap();
2971        store.put(Galaxy::Codex, &mem2).unwrap();
2972
2973        let query = MemoryQuery::new().with_tags(vec!["rust".into()]);
2974        let results = store.query(Galaxy::Codex, &query).unwrap();
2975        assert_eq!(results.len(), 1);
2976        assert_eq!(results[0].content, "tagged");
2977    }
2978
2979    #[test]
2980    fn query_by_importance_uses_index() {
2981        let tmp = tempfile::tempdir().unwrap();
2982        let store = MemoryStore::open_default(tmp.path()).unwrap();
2983
2984        store
2985            .put(
2986                Galaxy::Codex,
2987                &Memory::new(Galaxy::Codex, "low".into()).with_importance(0.1),
2988            )
2989            .unwrap();
2990        store
2991            .put(
2992                Galaxy::Codex,
2993                &Memory::new(Galaxy::Codex, "mid".into()).with_importance(0.5),
2994            )
2995            .unwrap();
2996        store
2997            .put(
2998                Galaxy::Codex,
2999                &Memory::new(Galaxy::Codex, "high".into()).with_importance(0.9),
3000            )
3001            .unwrap();
3002
3003        let query = MemoryQuery::new().with_importance_range(0.4, 0.6);
3004        let results = store.query(Galaxy::Codex, &query).unwrap();
3005        assert_eq!(results.len(), 1);
3006        assert_eq!(results[0].content, "mid");
3007    }
3008
3009    #[test]
3010    fn query_by_time_uses_index() {
3011        let tmp = tempfile::tempdir().unwrap();
3012        let store = MemoryStore::open_default(tmp.path()).unwrap();
3013
3014        let t0 = chrono::Utc::now();
3015        std::thread::sleep(std::time::Duration::from_millis(10));
3016        let mem = Memory::new(Galaxy::Codex, "timed".into());
3017        store.put(Galaxy::Codex, &mem).unwrap();
3018        std::thread::sleep(std::time::Duration::from_millis(10));
3019        let t2 = chrono::Utc::now();
3020
3021        let query = MemoryQuery::new().with_time_range(t0, t2);
3022        let results = store.query(Galaxy::Codex, &query).unwrap();
3023        assert_eq!(results.len(), 1);
3024        assert_eq!(results[0].content, "timed");
3025    }
3026
3027    #[test]
3028    fn delete_removes_index_entries() {
3029        let tmp = tempfile::tempdir().unwrap();
3030        let store = MemoryStore::open_default(tmp.path()).unwrap();
3031
3032        let mem = Memory::new(Galaxy::Codex, "test".into())
3033            .with_tags(vec!["tag1".into()])
3034            .with_importance(0.7);
3035        let id = mem.metadata.id;
3036        let hash = mem.metadata.content_hash.clone();
3037        store.put(Galaxy::Codex, &mem).unwrap();
3038
3039        // Verify index entries exist
3040        assert!(
3041            store
3042                .find_by_content_hash(Galaxy::Codex, &hash)
3043                .unwrap()
3044                .is_some()
3045        );
3046
3047        // Delete
3048        store.delete(Galaxy::Codex, id).unwrap();
3049
3050        // Verify index entries are gone
3051        assert!(
3052            store
3053                .find_by_content_hash(Galaxy::Codex, &hash)
3054                .unwrap()
3055                .is_none()
3056        );
3057
3058        // Tag query should return 0
3059        let query = MemoryQuery::new().with_tags(vec!["tag1".into()]);
3060        let results = store.query(Galaxy::Codex, &query).unwrap();
3061        assert!(results.is_empty());
3062    }
3063
3064    #[test]
3065    fn put_batch_updates_indexes() {
3066        let tmp = tempfile::tempdir().unwrap();
3067        let store = MemoryStore::open_default(tmp.path()).unwrap();
3068
3069        let memories: Vec<Memory> = (0..5)
3070            .map(|i| {
3071                Memory::new(Galaxy::Codex, format!("batch-{i}"))
3072                    .with_tags(vec![format!("tag{i}")])
3073                    .with_importance(i as f32 * 0.2)
3074            })
3075            .collect();
3076        store.put_batch(Galaxy::Codex, &memories).unwrap();
3077
3078        for i in 0..5 {
3079            let query = MemoryQuery::new().with_tags(vec![format!("tag{i}")]);
3080            let results = store.query(Galaxy::Codex, &query).unwrap();
3081            assert_eq!(results.len(), 1, "tag{i} should have 1 result");
3082        }
3083    }
3084
3085    #[test]
3086    fn find_by_content_hash_indexed_matches_scan() {
3087        let tmp = tempfile::tempdir().unwrap();
3088        let store = MemoryStore::open_default(tmp.path()).unwrap();
3089
3090        let mem = Memory::new(Galaxy::Codex, "dedup test".into());
3091        let id = mem.metadata.id;
3092        let hash = mem.metadata.content_hash.clone();
3093        store.put(Galaxy::Codex, &mem).unwrap();
3094
3095        let indexed = store.find_by_content_hash(Galaxy::Codex, &hash).unwrap();
3096        let scanned = store
3097            .find_by_content_hash_scan(Galaxy::Codex, &hash)
3098            .unwrap();
3099
3100        assert_eq!(indexed, scanned);
3101        assert_eq!(indexed, Some(id));
3102    }
3103
3104    #[test]
3105    fn put_dedup_uses_index() {
3106        let tmp = tempfile::tempdir().unwrap();
3107        let store = MemoryStore::open_default(tmp.path()).unwrap();
3108
3109        let mem1 = Memory::new(Galaxy::Codex, "duplicate content".into());
3110        let id1 = store.put_dedup(Galaxy::Codex, &mem1).unwrap();
3111
3112        let mem2 = Memory::new(Galaxy::Codex, "duplicate content".into());
3113        let id2 = store.put_dedup(Galaxy::Codex, &mem2).unwrap();
3114
3115        assert_eq!(id1, id2, "dedup should return same ID for same content");
3116        assert_eq!(store.count(Galaxy::Codex).unwrap(), 1);
3117    }
3118
3119    #[test]
3120    fn put_semantic_updates_coord5d() {
3121        let tmp = tempfile::tempdir().unwrap();
3122        let store = MemoryStore::open_default(tmp.path()).unwrap();
3123
3124        let mut mem = Memory::new(
3125            Galaxy::Codex,
3126            "The algorithm computes data using a systematic method".to_string(),
3127        );
3128        let original_coord = mem.metadata.coord5d.clone();
3129        store.put_semantic(Galaxy::Codex, &mut mem).unwrap();
3130
3131        // coord5d should have changed from the SHA-256 hash-based encoding
3132        assert_ne!(
3133            mem.metadata.coord5d.x, original_coord.x,
3134            "semantic encoding should change x"
3135        );
3136        assert_ne!(
3137            mem.metadata.coord5d.y, original_coord.y,
3138            "semantic encoding should change y"
3139        );
3140
3141        // Verify it was stored with the semantic coordinate
3142        let retrieved = store.get(Galaxy::Codex, mem.metadata.id).unwrap().unwrap();
3143        assert_eq!(retrieved.metadata.coord5d.x, mem.metadata.coord5d.x);
3144    }
3145
3146    #[test]
3147    fn put_semantic_preserves_temporal_and_importance() {
3148        let tmp = tempfile::tempdir().unwrap();
3149        let store = MemoryStore::open_default(tmp.path()).unwrap();
3150
3151        let mut mem = Memory::new(Galaxy::Codex, "test content".into()).with_importance(0.8);
3152        mem.metadata.coord5d.w = 0.6;
3153        store.put_semantic(Galaxy::Codex, &mut mem).unwrap();
3154
3155        assert!((mem.metadata.coord5d.w - 0.6).abs() < f32::EPSILON);
3156        assert!((mem.metadata.coord5d.v - 0.8).abs() < f32::EPSILON);
3157    }
3158
3159    #[test]
3160    fn find_similar_returns_nearest_first() {
3161        let tmp = tempfile::tempdir().unwrap();
3162        let store = MemoryStore::open_default(tmp.path()).unwrap();
3163
3164        let mut logic_mem = Memory::new(
3165            Galaxy::Codex,
3166            "The algorithm computes data using systematic logic and analysis".to_string(),
3167        );
3168        store.put_semantic(Galaxy::Codex, &mut logic_mem).unwrap();
3169
3170        let mut emotion_mem = Memory::new(
3171            Galaxy::Codex,
3172            "I feel love and joy with deep passion and empathy in my heart".to_string(),
3173        );
3174        store.put_semantic(Galaxy::Codex, &mut emotion_mem).unwrap();
3175
3176        // Query with logic-like text should find the logic memory first
3177        let results = store
3178            .find_similar(Galaxy::Codex, "algorithm data systematic method", 10)
3179            .unwrap();
3180        assert!(!results.is_empty());
3181        assert_eq!(results[0].0.metadata.id, logic_mem.metadata.id);
3182
3183        // Query with emotion-like text should find the emotion memory first
3184        let results = store
3185            .find_similar(Galaxy::Codex, "love joy passion heart feeling", 10)
3186            .unwrap();
3187        assert!(!results.is_empty());
3188        assert_eq!(results[0].0.metadata.id, emotion_mem.metadata.id);
3189    }
3190
3191    #[test]
3192    fn find_similar_empty_galaxy() {
3193        let tmp = tempfile::tempdir().unwrap();
3194        let store = MemoryStore::open_default(tmp.path()).unwrap();
3195
3196        let results = store.find_similar(Galaxy::Codex, "anything", 10).unwrap();
3197        assert!(results.is_empty());
3198    }
3199
3200    #[test]
3201    fn find_similar_respects_limit() {
3202        let tmp = tempfile::tempdir().unwrap();
3203        let store = MemoryStore::open_default(tmp.path()).unwrap();
3204
3205        for i in 0..5 {
3206            let mut mem = Memory::new(Galaxy::Codex, format!("algorithm data method {i}"));
3207            store.put_semantic(Galaxy::Codex, &mut mem).unwrap();
3208        }
3209
3210        let results = store
3211            .find_similar(Galaxy::Codex, "algorithm data", 3)
3212            .unwrap();
3213        assert_eq!(results.len(), 3);
3214    }
3215
3216    #[test]
3217    fn semantic_encoder_accessible() {
3218        let tmp = tempfile::tempdir().unwrap();
3219        let store = MemoryStore::open_default(tmp.path()).unwrap();
3220
3221        let scores = store.semantic_encoder().encode("algorithm data logic");
3222        // Logic-heavy text → x < 0.5
3223        assert!(scores.x < 0.5);
3224    }
3225
3226    #[test]
3227    fn put_raw_batch_writes_atomically() {
3228        let tmp = tempfile::tempdir().unwrap();
3229        let store = MemoryStore::open_default(tmp.path()).unwrap();
3230
3231        let entries: &[(&[u8], &[u8])] =
3232            &[(b"key1", b"val1"), (b"key2", b"val2"), (b"key3", b"val3")];
3233        store.put_raw_batch(Galaxy::Karma, entries).unwrap();
3234
3235        assert_eq!(
3236            store.get_raw(Galaxy::Karma, b"key1").unwrap().unwrap(),
3237            b"val1"
3238        );
3239        assert_eq!(
3240            store.get_raw(Galaxy::Karma, b"key2").unwrap().unwrap(),
3241            b"val2"
3242        );
3243        assert_eq!(
3244            store.get_raw(Galaxy::Karma, b"key3").unwrap().unwrap(),
3245            b"val3"
3246        );
3247    }
3248
3249    #[test]
3250    fn put_raw_batch_empty_is_noop() {
3251        let tmp = tempfile::tempdir().unwrap();
3252        let store = MemoryStore::open_default(tmp.path()).unwrap();
3253
3254        store.put_raw_batch(Galaxy::Karma, &[]).unwrap();
3255        assert_eq!(store.count(Galaxy::Karma).unwrap(), 0);
3256    }
3257
3258    #[test]
3259    fn entry_limit_rejects_excess_writes() {
3260        let tmp = tempfile::tempdir().unwrap();
3261        let store = MemoryStore::open_default(tmp.path())
3262            .unwrap()
3263            .with_entry_limit(3);
3264
3265        for i in 0..3 {
3266            let mem = Memory::new(Galaxy::Codex, format!("memory {i}"));
3267            store.put(Galaxy::Codex, &mem).unwrap();
3268        }
3269
3270        // 4th write should be rejected
3271        let mem = Memory::new(Galaxy::Codex, "overflow memory".to_string());
3272        let result = store.put(Galaxy::Codex, &mem);
3273        assert!(result.is_err(), "write beyond limit should be rejected");
3274        let err_msg = result.unwrap_err().to_string();
3275        assert!(
3276            err_msg.contains("entry limit reached"),
3277            "error should mention entry limit: {err_msg}"
3278        );
3279        assert_eq!(store.count(Galaxy::Codex).unwrap(), 3);
3280    }
3281
3282    #[test]
3283    fn entry_limit_per_galaxy_independent() {
3284        let tmp = tempfile::tempdir().unwrap();
3285        let store = MemoryStore::open_default(tmp.path())
3286            .unwrap()
3287            .with_entry_limit(2);
3288
3289        // Fill Codex to limit
3290        for i in 0..2 {
3291            let mem = Memory::new(Galaxy::Codex, format!("codex {i}"));
3292            store.put(Galaxy::Codex, &mem).unwrap();
3293        }
3294
3295        // Writing to a different galaxy should still work
3296        let mem = Memory::new(Galaxy::Research, "science memory".to_string());
3297        let result = store.put(Galaxy::Research, &mem);
3298        assert!(
3299            result.is_ok(),
3300            "different galaxy should not be affected by limit"
3301        );
3302    }
3303
3304    #[test]
3305    fn entry_limit_none_allows_unlimited() {
3306        let tmp = tempfile::tempdir().unwrap();
3307        let store = MemoryStore::open_default(tmp.path()).unwrap();
3308
3309        // No limit set — should allow many writes
3310        for i in 0..50 {
3311            let mem = Memory::new(Galaxy::Codex, format!("memory {i}"));
3312            store.put(Galaxy::Codex, &mem).unwrap();
3313        }
3314        assert_eq!(store.count(Galaxy::Codex).unwrap(), 50);
3315    }
3316
3317    #[test]
3318    fn map_full_error_is_graceful() {
3319        // Small map: opening succeeds (galaxy + sidecar databases fit), but
3320        // the padded write loop fills it and MapFull surfaces from put().
3321        // 64KB proved too tight for eager create_db on macOS/arm64.
3322        let tmp = tempfile::tempdir().unwrap();
3323        let store = MemoryStore::open(tmp.path(), 512 * 1024).unwrap();
3324
3325        // Write memories until map is full
3326        let mut written = 0;
3327        let mut got_map_full = false;
3328        for i in 0..1000 {
3329            let mem = Memory::new(
3330                Galaxy::Codex,
3331                format!("memory content {i} {}", "with padding ".repeat(50)),
3332            );
3333            match store.put(Galaxy::Codex, &mem) {
3334                Ok(()) => written += 1,
3335                Err(e) => {
3336                    let msg = e.to_string();
3337                    if msg.contains("map full") {
3338                        got_map_full = true;
3339                        break;
3340                    }
3341                    // Other errors are fine too (e.g., serialize failed)
3342                    break;
3343                }
3344            }
3345        }
3346
3347        assert!(
3348            got_map_full || written < 1000,
3349            "should eventually hit map full or error"
3350        );
3351        assert!(written > 0, "should have written at least some memories");
3352    }
3353
3354    #[test]
3355    fn test_find_across_galaxies() {
3356        let tmp = tempfile::tempdir().unwrap();
3357        let store = MemoryStore::open_default(tmp.path()).unwrap();
3358
3359        let mem = Memory::new(Galaxy::Research, "Cross-galaxy research memo".into());
3360        let id = mem.metadata.id;
3361        store.put(Galaxy::Research, &mem).unwrap();
3362
3363        let found = store.find_across_galaxies(id).unwrap();
3364        assert!(found.is_some());
3365        let (galaxy, retrieved) = found.unwrap();
3366        assert_eq!(galaxy, Galaxy::Research);
3367        assert_eq!(retrieved.metadata.id, id);
3368        assert_eq!(retrieved.content, "Cross-galaxy research memo");
3369
3370        // Non-existent id returns None
3371        assert!(
3372            store
3373                .find_across_galaxies(uuid::Uuid::new_v4())
3374                .unwrap()
3375                .is_none()
3376        );
3377    }
3378}