1use 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#[derive(Debug, Clone, Default)]
25pub struct MemoryQuery {
26 pub tags: Vec<String>,
28 pub min_importance: Option<f32>,
30 pub max_importance: Option<f32>,
32 pub created_after: Option<chrono::DateTime<chrono::Utc>>,
34 pub created_before: Option<chrono::DateTime<chrono::Utc>>,
36 pub content_substring: Option<String>,
39 pub limit: usize,
41}
42
43impl MemoryQuery {
44 #[must_use]
46 pub fn new() -> Self {
47 Self {
48 limit: 100,
49 ..Default::default()
50 }
51 }
52
53 #[must_use]
55 pub fn with_tags(mut self, tags: Vec<String>) -> Self {
56 self.tags = tags;
57 self
58 }
59
60 #[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 #[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 #[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 #[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 #[must_use]
98 pub const fn with_limit(mut self, limit: usize) -> Self {
99 self.limit = limit;
100 self
101 }
102
103 #[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 #[must_use]
112 pub fn matches(&self, mem: &Memory) -> bool {
113 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 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 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 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
157pub struct MemoryStore {
159 path: std::path::PathBuf,
161 env: Environment,
163 index_dbs: IndexDbs,
165 semantic_encoder: SemanticEncoder,
167 max_entries_per_galaxy: Option<usize>,
169 mutation_count: AtomicU64,
173 episodic_db: Database,
175 episodic_terms_v2_db: Database,
177 embedding_cache_db: Database,
182 revisions_db: Database,
186 attestations_db: Database,
190 pub(crate) cold_storage_db: Database,
192 keyring_db: Option<Database>,
195 at_rest: Option<crate::at_rest::AtRestState>,
198 episodic_term_cache: std::sync::Arc<RwLock<HashMap<String, Vec<uuid::Uuid>>>>,
200 episodic_embedder:
202 std::sync::OnceLock<Option<Arc<dyn crate::embedder::Embedder + Send + Sync>>>,
203 episodic_sidecar_ensured: std::sync::OnceLock<()>,
205 episodic_aliases: std::sync::OnceLock<Option<crate::episodic_keys::AdaptiveAliases>>,
207 episodic_enrichment: std::sync::OnceLock<Option<crate::enrichment::VocabularyEnrichment>>,
209}
210
211impl MemoryStore {
212 #[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 #[cfg(not(unix))]
248 pub fn probe_write_lock(_store_root: &Path) -> std::io::Result<()> {
249 Ok(())
250 }
251
252 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 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 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 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 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 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 #[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 #[must_use]
395 pub const fn at_rest_state(&self) -> Option<&crate::at_rest::AtRestState> {
396 self.at_rest.as_ref()
397 }
398
399 #[must_use]
403 pub const fn keyring_db(&self) -> Option<Database> {
404 self.keyring_db
405 }
406
407 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 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 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 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 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 #[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 pub fn open_default(path: impl AsRef<Path>) -> Result<Self> {
531 let size = Self::default_map_size();
536 Self::open(path, size)
537 }
538
539 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 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 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 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 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 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 let store = Self::open_default(&path)?;
734 drop(store);
735 Ok(missing)
736 }
737
738 #[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 pub fn path(&self) -> &Path {
750 &self.path
751 }
752
753 pub const fn env(&self) -> &Environment {
755 &self.env
756 }
757
758 pub fn mutation_count(&self) -> u64 {
762 self.mutation_count.load(Ordering::Relaxed)
763 }
764
765 pub const fn index_dbs(&self) -> &IndexDbs {
767 &self.index_dbs
768 }
769
770 pub const fn semantic_encoder(&self) -> &SemanticEncoder {
772 &self.semantic_encoder
773 }
774
775 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 #[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 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 pub fn set_episodic_aliases(&self, aliases: crate::episodic_keys::AdaptiveAliases) {
838 let _ = self.episodic_aliases.set(Some(aliases));
839 }
840
841 pub fn set_episodic_enrichment(&self, enrichment: crate::enrichment::VocabularyEnrichment) {
843 let _ = self.episodic_enrichment.set(Some(enrichment));
844 }
845
846 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 pub fn put(&self, galaxy: Galaxy, memory: &Memory) -> Result<()> {
861 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 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 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 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 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 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 let exists = tx.get(db, key).is_ok();
968 if exists {
969 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 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 pub fn scan_all(&self, galaxy: Galaxy) -> Result<Vec<Memory>> {
1025 self.scan_all_impl(galaxy, false)
1026 }
1027
1028 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 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 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 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 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 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 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 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 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 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 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 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 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 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 pub fn query(&self, galaxy: Galaxy, query: &MemoryQuery) -> Result<Vec<Memory>> {
1363 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 pub fn revisions(
1756 &self,
1757 galaxy: Galaxy,
1758 id: MemoryId,
1759 ) -> Result<Vec<crate::revision::MemoryRevision>> {
1760 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 self.put_cold_record(&record)?;
2264
2265 self.delete(galaxy, memory_id)?;
2267
2268 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 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 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 self.put(record.galaxy, &mem)?;
2306
2307 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 self.delete_cold_record(memory_id)?;
2317
2318 Ok(mem)
2319 }
2320
2321 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 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 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 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 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 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 #[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 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 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 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 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 #[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 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 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 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 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 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 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 assert!(
3041 store
3042 .find_by_content_hash(Galaxy::Codex, &hash)
3043 .unwrap()
3044 .is_some()
3045 );
3046
3047 store.delete(Galaxy::Codex, id).unwrap();
3049
3050 assert!(
3052 store
3053 .find_by_content_hash(Galaxy::Codex, &hash)
3054 .unwrap()
3055 .is_none()
3056 );
3057
3058 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 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 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 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 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 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 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 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 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 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 let tmp = tempfile::tempdir().unwrap();
3323 let store = MemoryStore::open(tmp.path(), 512 * 1024).unwrap();
3324
3325 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 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 assert!(
3372 store
3373 .find_across_galaxies(uuid::Uuid::new_v4())
3374 .unwrap()
3375 .is_none()
3376 );
3377 }
3378}