wm_memory/

store.rs

//! LMDB-backed memory store.
//!
//! Each galaxy is an LMDB named database (sub-DB within the same file).
//! Reads are zero-copy (mmap'd). Writes are batched.

use lmdb::{
    Cursor, Database, DatabaseFlags, Environment, EnvironmentFlags, Transaction, WriteFlags,
};
use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use wm_core::{CoreError, Galaxy, Result};

#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;

use crate::episodic::EpisodicStore;
use crate::indexes::IndexDbs;
use crate::memory::{Memory, MemoryId, decode_embedding, encode_embedding};
use crate::semantic::SemanticEncoder;

/// Query filter for memories.
#[derive(Debug, Clone, Default)]
pub struct MemoryQuery {
    /// Filter by tags (memory must contain ALL specified tags).
    pub tags: Vec<String>,
    /// Minimum importance (inclusive).
    pub min_importance: Option<f32>,
    /// Maximum importance (inclusive).
    pub max_importance: Option<f32>,
    /// Only memories created after this timestamp.
    pub created_after: Option<chrono::DateTime<chrono::Utc>>,
    /// Only memories created before this timestamp.
    pub created_before: Option<chrono::DateTime<chrono::Utc>>,
    /// Case-insensitive substring filter over content (literal match —
    /// not tokenized or ranked; that is what the search engine is for).
    pub content_substring: Option<String>,
    /// Maximum number of results.
    pub limit: usize,
}

impl MemoryQuery {
    /// Create an empty query (matches all, limit 100).
    #[must_use]
    pub fn new() -> Self {
        Self {
            limit: 100,
            ..Default::default()
        }
    }

    /// Set tag filter.
    #[must_use]
    pub fn with_tags(mut self, tags: Vec<String>) -> Self {
        self.tags = tags;
        self
    }

    /// Set importance range.
    #[must_use]
    pub const fn with_importance_range(mut self, min: f32, max: f32) -> Self {
        self.min_importance = Some(min);
        self.max_importance = Some(max);
        self
    }

    /// Set temporal range.
    #[must_use]
    pub const fn with_time_range(
        mut self,
        after: chrono::DateTime<chrono::Utc>,
        before: chrono::DateTime<chrono::Utc>,
    ) -> Self {
        self.created_after = Some(after);
        self.created_before = Some(before);
        self
    }

    /// One-sided temporal bound: only memories created at or after this
    /// timestamp (`created_after` API passthrough).
    #[must_use]
    pub const fn with_created_after(mut self, after: chrono::DateTime<chrono::Utc>) -> Self {
        self.created_after = Some(after);
        self
    }

    /// One-sided temporal bound: only memories created at or before this
    /// timestamp (`created_before` API passthrough).
    #[must_use]
    pub const fn with_created_before(mut self, before: chrono::DateTime<chrono::Utc>) -> Self {
        self.created_before = Some(before);
        self
    }

    /// Set limit.
    #[must_use]
    pub const fn with_limit(mut self, limit: usize) -> Self {
        self.limit = limit;
        self
    }

    /// Set a case-insensitive substring filter over content.
    #[must_use]
    pub fn with_content_substring(mut self, substring: impl Into<String>) -> Self {
        self.content_substring = Some(substring.into().to_lowercase());
        self
    }

    /// Check if a memory matches this query.
    #[must_use]
    pub fn matches(&self, mem: &Memory) -> bool {
        // Tag filter: memory must contain all specified tags
        if !self.tags.is_empty() {
            for tag in &self.tags {
                if !mem.metadata.tags.iter().any(|t| t == tag) {
                    return false;
                }
            }
        }

        // Importance filter
        if let Some(min) = self.min_importance {
            if mem.metadata.importance < min {
                return false;
            }
        }
        if let Some(max) = self.max_importance {
            if mem.metadata.importance > max {
                return false;
            }
        }

        // Temporal filter
        if let Some(after) = self.created_after {
            if mem.metadata.created_at < after {
                return false;
            }
        }
        if let Some(before) = self.created_before {
            if mem.metadata.created_at > before {
                return false;
            }
        }

        // Substring filter (literal, case-insensitive — never ranked).
        if let Some(sub) = &self.content_substring {
            if !mem.content.to_lowercase().contains(sub) {
                return false;
            }
        }

        true
    }
}

/// The LMDB environment containing all 14 galaxy sub-databases plus 4 index DBs.
pub struct MemoryStore {
    /// Path to the LMDB file
    path: std::path::PathBuf,
    /// LMDB environment (opened once, shared across threads)
    env: Environment,
    /// Cached handles to the 4 secondary index sub-databases
    index_dbs: IndexDbs,
    /// Semantic encoder for content-derived coordinates
    semantic_encoder: SemanticEncoder,
    /// Optional per-galaxy entry limit (DoS prevention)
    max_entries_per_galaxy: Option<usize>,
    /// Monotonic counter of successful mutations since this handle was
    /// opened. Lets the dispatch pipeline cheaply detect actual store
    /// writes (write-audit journal) without scanning galaxies.
    mutation_count: AtomicU64,
    /// Dedicated database for lossless v6 episodic records.
    episodic_db: Database,
    /// DUP_SORT term→id postings for bounded episodic search (v2 sidecar).
    episodic_terms_v2_db: Database,
    /// Content-hash → vector cache (v26 "Tier 2" idea, finally wired):
    /// warm-start for re-ingest and re-runs. Keyed by the embedder
    /// namespace + content hash, so switching models never serves stale
    /// vectors.
    embedding_cache_db: Database,
    /// Per-memory revision chains (V8 S11c): append-only content-change
    /// history, self-verifying (seq continuity + hash linkage + head
    /// match). See [`crate::revision`].
    revisions_db: Database,
    /// Per-memory creation attestations (Track F Slice A, D5): one signed
    /// record per created memory, keyed `att:{galaxy}:{memory_id}`.
    /// See [`crate::attestation`].
    attestations_db: Database,
    /// Dedicated database for compressed cold-stored memories.
    pub(crate) cold_storage_db: Database,
    /// Optional at-rest keyring DBI (Q39 slice A). `Some` when the store has
    /// a keyring; read-only paths open it optionally and never create it.
    keyring_db: Option<Database>,
    /// Unlocked galaxy DEKs for writable at-rest stores (slice A verifies
    /// them at open; record AEAD is slice B). `None` for plaintext stores.
    at_rest: Option<crate::at_rest::AtRestState>,
    /// Warm term-posting cache shared by episodic search views.
    episodic_term_cache: std::sync::Arc<RwLock<HashMap<String, Vec<uuid::Uuid>>>>,
    /// Optional embedder for episodic vector reranking.
    episodic_embedder:
        std::sync::OnceLock<Option<Arc<dyn crate::embedder::Embedder + Send + Sync>>>,
    /// One-shot guard: rebuild the v2 episodic sidecar once per process.
    episodic_sidecar_ensured: std::sync::OnceLock<()>,
    /// Optional adaptive aliases for episodic key expansion.
    episodic_aliases: std::sync::OnceLock<Option<crate::episodic_keys::AdaptiveAliases>>,
    /// Optional vocabulary enrichment for episodic index-time term expansion.
    episodic_enrichment: std::sync::OnceLock<Option<crate::enrichment::VocabularyEnrichment>>,
}

impl MemoryStore {
    /// Probe whether another process holds the LMDB writer lock on this store.
    ///
    /// LMDB's write env-open falls back to a blocking *shared* lock when the
    /// exclusive writer lock is held, then opens `data.mdb` for writing
    /// anyway and wedges on its internal mutex — `wm grimoire`/`wm status`
    /// hung forever against a live store until a SIGKILL (9.1.6). Callers
    /// that need exclusive access probe first and fail loudly instead of
    /// deadlocking.
    ///
    /// The probe is a non-blocking `fcntl(F_SETLK, F_WRLCK)` over the whole
    /// `lock.mdb` (record locks overlap LMDB's byte-range writer lock), so
    /// it never blocks and never mutates the store.
    ///
    /// Returns `Ok(())` when the writer lock is free, `Err(WouldBlock)`
    /// when another process holds it.
    #[cfg(unix)]
    pub fn probe_write_lock(store_root: &Path) -> std::io::Result<()> {
        use rustix::fs::{FlockOperation, fcntl_lock};
        let lock_path = store_root.join("lmdb").join("lock.mdb");
        let file = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(&lock_path)?;
        match fcntl_lock(&file, FlockOperation::NonBlockingLockExclusive) {
            Ok(()) => Ok(()),
            Err(rustix::io::Errno::AGAIN | rustix::io::Errno::ACCESS) => Err(std::io::Error::new(
                std::io::ErrorKind::WouldBlock,
                "LMDB writer lock held by another process",
            )),
            Err(e) => Err(e.into()),
        }
    }

    /// Non-unix: LMDB locking differs (LockFileEx on Windows); the probe is
    /// best-effort there and reports the lock as free.
    #[cfg(not(unix))]
    pub fn probe_write_lock(_store_root: &Path) -> std::io::Result<()> {
        Ok(())
    }

    /// Open or create an LMDB store at the given path.
    ///
    /// At-rest mode comes from the environment (`WM_AT_REST_MODE`, default
    /// `off`) — see [`Self::open_with_at_rest`]. `off` is a provable no-op:
    /// no keyring DBI is created and no key files are written. A
    /// set-but-unrecognized `WM_AT_REST_MODE` value refuses the open
    /// (fail-closed).
    ///
    /// On Unix, the store directory is created with mode 0o700 (owner-only
    /// access) if it does not already exist. Existing directories are
    /// left untouched.
    pub fn open(path: impl AsRef<Path>, map_size: usize) -> Result<Self> {
        Self::open_with_at_rest(path, map_size, &crate::at_rest::AtRestConfig::from_env()?)
    }

    /// Open or create an LMDB store with an explicit at-rest configuration.
    ///
    /// Q39 slice A: when the mode is `keyfile`/`passphrase`, the store's
    /// keyring DBI is read (or initialized as `meta` + `rk:check` + 16
    /// wrapped galaxy DEKs, all in one transaction) and the DEKs are
    /// unwrapped at open. Records stay plaintext in slice A.
    pub fn open_with_at_rest(
        path: impl AsRef<Path>,
        map_size: usize,
        at_rest_config: &crate::at_rest::AtRestConfig,
    ) -> Result<Self> {
        let path = path.as_ref().to_path_buf();

        // Ensure the directory exists with restrictive permissions.
        std::fs::create_dir_all(&path)
            .map_err(|e| CoreError::Memory(format!("Cannot create store dir: {e}")))?;
        #[cfg(unix)]
        {
            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700))
                .map_err(|e| CoreError::Memory(format!("Cannot set store dir permissions: {e}")))?;
        }

        let env = Environment::new()
            .set_map_size(map_size)
            .set_max_dbs(64)
            .open(&path)
            .map_err(|e| CoreError::Memory(format!("LMDB open failed: {e}")))?;

        // Create all 14 galaxy sub-databases
        for galaxy in Galaxy::all() {
            let db = env
                .create_db(Some(galaxy.db_name()), DatabaseFlags::default())
                .map_err(|e| {
                    CoreError::Memory(format!(
                        "LMDB create_db failed for {}: {e}",
                        galaxy.db_name()
                    ))
                })?;
            let _ = db;
        }

        // Create 4 secondary index sub-databases
        for (name, flags) in crate::indexes::INDEX_DBS {
            let db = env
                .create_db(Some(name), *flags)
                .map_err(|e| CoreError::Memory(format!("LMDB create_db failed for {name}: {e}")))?;
            let _ = db;
        }

        let index_dbs = IndexDbs::open(&env)?;
        let episodic_db = env
            .create_db(Some("episodic_records"), DatabaseFlags::default())
            .map_err(|e| {
                CoreError::Memory(format!("LMDB create_db failed for episodic_records: {e}"))
            })?;
        // v2 sidecar: DUP_SORT postings (term -> set of record ids). Append of
        // a record touches only the (term, id) pairs it introduces instead of
        // rewriting whole posting lists, so ingest cost stays O(new records)
        // as the store grows. The v1 msgpack-Vec database is retained unused
        // on legacy stores; v2 is rebuilt from the authoritative records when
        // found empty.
        let episodic_terms_v2_db = env
            .create_db(Some("episodic_terms_v2"), DatabaseFlags::DUP_SORT)
            .map_err(|e| {
                CoreError::Memory(format!("LMDB create_db failed for episodic_terms_v2: {e}"))
            })?;
        let embedding_cache_db = env
            .create_db(Some("embedding_cache"), DatabaseFlags::default())
            .map_err(|e| {
                CoreError::Memory(format!("LMDB create_db failed for embedding_cache: {e}"))
            })?;
        let revisions_db = env
            .create_db(Some("revisions"), DatabaseFlags::default())
            .map_err(|e| CoreError::Memory(format!("LMDB create_db failed for revisions: {e}")))?;
        let attestations_db = env
            .create_db(
                Some(crate::attestation::ATTESTATIONS_DB),
                DatabaseFlags::default(),
            )
            .map_err(|e| {
                CoreError::Memory(format!("LMDB create_db failed for attestations: {e}"))
            })?;
        let cold_storage_db = env
            .create_db(Some("cold_storage"), DatabaseFlags::default())
            .map_err(|e| {
                CoreError::Memory(format!("LMDB create_db failed for cold_storage: {e}"))
            })?;
        let (keyring_db, at_rest) = crate::at_rest::open_at_rest(&env, &path, at_rest_config)?;
        Ok(Self {
            path,
            env,
            index_dbs,
            semantic_encoder: SemanticEncoder::new(),
            max_entries_per_galaxy: None,
            mutation_count: AtomicU64::new(0),
            episodic_db,
            episodic_terms_v2_db,
            embedding_cache_db,
            revisions_db,
            attestations_db,
            cold_storage_db,
            keyring_db,
            at_rest,
            episodic_term_cache: std::sync::Arc::new(RwLock::new(HashMap::new())),
            episodic_embedder: std::sync::OnceLock::new(),
            episodic_sidecar_ensured: std::sync::OnceLock::new(),
            episodic_aliases: std::sync::OnceLock::new(),
            episodic_enrichment: std::sync::OnceLock::new(),
        })
    }

    /// At-rest disclosure status (Q39 slice A): keyring meta only — the RK is
    /// never resolved here, and nothing is created or written. `Absent` means
    /// plaintext pass-through.
    #[must_use]
    pub fn at_rest_status(&self) -> crate::at_rest::AtRestStatus {
        if let Some(state) = &self.at_rest {
            return crate::at_rest::AtRestStatus::Present(state.status());
        }
        match &self.keyring_db {
            None => crate::at_rest::AtRestStatus::Absent,
            Some(db) => crate::at_rest::read_status(&self.env, *db, &self.path),
        }
    }

    /// Unlocked keyring state for a writable at-rest store (slice B seam);
    /// `None` for plaintext-pass-through stores and all read paths.
    #[must_use]
    pub const fn at_rest_state(&self) -> Option<&crate::at_rest::AtRestState> {
        self.at_rest.as_ref()
    }

    /// Bounded env open for inspection paths (9.1.6).
    ///
    /// LMDB env opens can block forever: against a live writer the
    /// exclusive-lock fallback wedges on an internal mutex, and a crashed
    /// server can leave the lock file's in-file mutex locked so even
    /// read-only opens hang. Inspection callers (status, doctor, grimoire)
    /// must never hang — run the open on a worker thread and give up after
    /// `timeout`, returning `Ok(None)` so the caller degrades loudly
    /// instead of deadlocking.
    pub fn open_readonly_bounded(
        path: impl AsRef<Path>,
        timeout: std::time::Duration,
    ) -> Result<Option<Self>> {
        let path = path.as_ref().to_path_buf();
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let result = Self::open_readonly(&path);
            let _ = tx.send(result);
        });
        match rx.recv_timeout(timeout) {
            Ok(result) => result.map(Some),
            Err(_) => Ok(None),
        }
    }

    /// Bounded writable env open for exclusive-access paths (9.1.6).
    /// See [`Self::open_readonly_bounded`] for the wedge rationale; write
    /// paths bail with an actionable message on timeout instead of hanging.
    pub fn open_default_bounded(
        path: impl AsRef<Path>,
        timeout: std::time::Duration,
    ) -> Result<Option<Self>> {
        let path = path.as_ref().to_path_buf();
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let result = Self::open_default(&path);
            let _ = tx.send(result);
        });
        match rx.recv_timeout(timeout) {
            Ok(result) => result.map(Some),
            Err(_) => Ok(None),
        }
    }

    /// Open with the default map size.
    ///
    /// 4 GB on Unix: LMDB truncates the data file sparsely (ftruncate), so
    /// reservation costs nothing until pages are written. On Windows NTFS
    /// materializes the file at full map size immediately — a 4 GB default
    /// would allocate 4 GB on disk per store the moment it opens — so the
    /// Windows default is smaller; pass an explicit size to `open()` for
    /// large stores. (Auto-grow on MapFull is a planned follow-up.)
    pub fn open_default(path: impl AsRef<Path>) -> Result<Self> {
        // Deployment knob: override the platform default explicitly (bytes).
        // CI uses this on Windows, where NTFS materializes the map file at
        // full size and hundreds of parallel test stores would exhaust the
        // runner disk even at the 256MB Windows default.
        let platform_default = if cfg!(windows) {
            256 * 1024 * 1024
        } else {
            4 * 1024 * 1024 * 1024
        };
        let size = std::env::var("WM_DEFAULT_MAP_SIZE")
            .ok()
            .and_then(|v| v.parse::<usize>().ok())
            .filter(|&v| v > 0)
            .unwrap_or(platform_default);
        Self::open(path, size)
    }

    /// Open an existing LMDB store without creating a directory, database, or
    /// writable LMDB environment. This is the preservation boundary used by
    /// read-only evaluator servers: an incomplete or incompatible store must
    /// fail closed for the caller to investigate, never be initialized or
    /// repaired in place.
    pub fn open_readonly(path: impl AsRef<Path>) -> Result<Self> {
        let path = path.as_ref().to_path_buf();
        if !path.is_dir() {
            return Err(CoreError::Memory(format!(
                "Read-only LMDB store directory does not exist: {}",
                path.display()
            )));
        }
        if !path.join("data.mdb").is_file() {
            return Err(CoreError::Memory(format!(
                "Read-only LMDB store is missing data.mdb: {}",
                path.display()
            )));
        }

        let env = Environment::new()
            .set_max_dbs(32)
            .set_flags(EnvironmentFlags::READ_ONLY)
            .open(&path)
            .map_err(|e| CoreError::Memory(format!("Read-only LMDB open failed: {e}")))?;

        let index_dbs = IndexDbs::open(&env)?;
        let open_named = |name: &str| {
            env.open_db(Some(name)).map_err(|e| {
                CoreError::Memory(format!("Read-only LMDB missing database {name}: {e}"))
            })
        };
        let episodic_db = open_named("episodic_records")?;
        let episodic_terms_v2_db = open_named("episodic_terms_v2")?;
        let embedding_cache_db = open_named("embedding_cache")?;
        let revisions_db = open_named("revisions")?;
        let attestations_db = open_named(crate::attestation::ATTESTATIONS_DB)?;
        let cold_storage_db = open_named("cold_storage")?;
        // The at-rest keyring is optional on read paths: opened when present,
        // never created, and its RK is never resolved here (status only).
        let keyring_db = crate::at_rest::open_keyring_optional(&env)?;

        Ok(Self {
            path,
            env,
            index_dbs,
            semantic_encoder: SemanticEncoder::new(),
            max_entries_per_galaxy: None,
            mutation_count: AtomicU64::new(0),
            episodic_db,
            episodic_terms_v2_db,
            embedding_cache_db,
            revisions_db,
            attestations_db,
            cold_storage_db,
            keyring_db,
            at_rest: None,
            episodic_term_cache: std::sync::Arc::new(RwLock::new(HashMap::new())),
            episodic_embedder: std::sync::OnceLock::new(),
            episodic_sidecar_ensured: std::sync::OnceLock::new(),
            episodic_aliases: std::sync::OnceLock::new(),
            episodic_enrichment: std::sync::OnceLock::new(),
        })
    }

    /// Named databases `open()` creates and `open_readonly()` requires,
    /// beyond the galaxy and secondary-index sets.
    const NAMED_DBIS: [&'static str; 6] = [
        "episodic_records",
        "episodic_terms_v2",
        "embedding_cache",
        "revisions",
        crate::attestation::ATTESTATIONS_DB,
        "cold_storage",
    ];

    /// Open an existing store for inspection without taking any lock
    /// (`MDB_NOLOCK | MDB_RDONLY`), 9.1.6.
    ///
    /// Read-only env opens still block forever in two real situations:
    /// a live writer holds the exclusive lock (lmdb-master falls back to a
    /// blocking shared-lock wait), and a crashed server can leave the lock
    /// file's in-file mutex wedged so every open hangs. Inspection paths
    /// (status, doctor, grimoire) never need the lock file — no locks, no
    /// reader slots, no mutex — just an mmap read of the store. The store
    /// must exist and be schema-complete (same strict refusal as
    /// [`Self::open_readonly`]); torn-meta-page reads are theoretically
    /// possible mid-write and acceptable for display counts.
    pub fn open_inspection(path: impl AsRef<Path>) -> Result<Self> {
        let path = path.as_ref().to_path_buf();
        if !path.is_dir() {
            return Err(CoreError::Memory(format!(
                "Read-only LMDB store directory does not exist: {}",
                path.display()
            )));
        }
        if !path.join("data.mdb").is_file() {
            return Err(CoreError::Memory(format!(
                "Read-only LMDB store is missing data.mdb: {}",
                path.display()
            )));
        }

        let env = Environment::new()
            .set_max_dbs(32)
            .set_flags(EnvironmentFlags::READ_ONLY | EnvironmentFlags::NO_LOCK)
            .open(&path)
            .map_err(|e| CoreError::Memory(format!("Inspection LMDB open failed: {e}")))?;

        let index_dbs = IndexDbs::open(&env)?;
        let open_named = |name: &str| {
            env.open_db(Some(name)).map_err(|e| {
                CoreError::Memory(format!("Inspection LMDB missing database {name}: {e}"))
            })
        };
        let episodic_db = open_named("episodic_records")?;
        let episodic_terms_v2_db = open_named("episodic_terms_v2")?;
        let embedding_cache_db = open_named("embedding_cache")?;
        let revisions_db = open_named("revisions")?;
        let attestations_db = open_named(crate::attestation::ATTESTATIONS_DB)?;
        let cold_storage_db = open_named("cold_storage")?;
        // The at-rest keyring is optional on read paths: opened when present,
        // never created, and its RK is never resolved here (status only).
        let keyring_db = crate::at_rest::open_keyring_optional(&env)?;

        Ok(Self {
            path,
            env,
            index_dbs,
            semantic_encoder: SemanticEncoder::new(),
            max_entries_per_galaxy: None,
            mutation_count: AtomicU64::new(0),
            episodic_db,
            episodic_terms_v2_db,
            embedding_cache_db,
            revisions_db,
            attestations_db,
            cold_storage_db,
            keyring_db,
            at_rest: None,
            episodic_term_cache: std::sync::Arc::new(RwLock::new(HashMap::new())),
            episodic_embedder: std::sync::OnceLock::new(),
            episodic_sidecar_ensured: std::sync::OnceLock::new(),
            episodic_aliases: std::sync::OnceLock::new(),
            episodic_enrichment: std::sync::OnceLock::new(),
        })
    }

    /// Complete a store's schema in place: create any galaxy, index, or named
    /// database this build expects but an older store lacks, then let the
    /// caller reopen normally. Returns the database names that were missing.
    ///
    /// Restores from older builds can be byte-exact yet not openable (found
    /// 2026-09-14: a 9.0.0 backup lacked `cold_storage`). This is the only
    /// in-place repair path; `open_readonly` deliberately stays strict so
    /// preservation callers see an incomplete store instead of a silent fix.
    ///
    /// The at-rest `keyring` DBI is **not** required schema (Q39 slice A
    /// decision): legacy stores and strict reads stay working, and this
    /// function neither creates nor repairs a keyring.
    pub fn ensure_schema(path: impl AsRef<Path>) -> Result<Vec<String>> {
        let path = path.as_ref().to_path_buf();
        if !path.is_dir() {
            return Err(CoreError::Memory(format!(
                "Store directory does not exist: {}",
                path.display()
            )));
        }
        let expected = || {
            Galaxy::all()
                .into_iter()
                .map(|galaxy| galaxy.db_name().to_string())
                .chain(
                    crate::indexes::INDEX_DBS
                        .iter()
                        .map(|(name, _)| (*name).to_string()),
                )
                .chain(Self::NAMED_DBIS.iter().map(|name| (*name).to_string()))
        };
        let missing: Vec<String> = {
            let env = Environment::new()
                .set_max_dbs(64)
                .set_flags(EnvironmentFlags::READ_ONLY)
                .open(&path)
                .map_err(|e| CoreError::Memory(format!("Read-only LMDB open failed: {e}")))?;
            expected()
                .filter(|name| env.open_db(Some(name.as_str())).is_err())
                .collect()
        };
        if missing.is_empty() {
            return Ok(missing);
        }
        // A writable open creates every missing galaxy, index, and named
        // database. Drop it immediately; the caller reopens as usual.
        let store = Self::open_default(&path)?;
        drop(store);
        Ok(missing)
    }

    /// Set a per-galaxy entry limit for DoS prevention.
    ///
    /// When set, `put` will reject writes that would exceed the limit.
    /// This prevents a single galaxy from exhausting the LMDB map.
    #[must_use]
    pub const fn with_entry_limit(mut self, limit: usize) -> Self {
        self.max_entries_per_galaxy = Some(limit);
        self
    }

    /// Path to the LMDB file.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Get the LMDB environment handle.
    pub const fn env(&self) -> &Environment {
        &self.env
    }

    /// Monotonic counter of successful mutations since this handle was
    /// opened (puts, deletes, clears, raw writes). Used by the dispatch
    /// pipeline's write-audit journal to detect actual store writes.
    pub fn mutation_count(&self) -> u64 {
        self.mutation_count.load(Ordering::Relaxed)
    }

    /// Get the cached index database handles.
    pub const fn index_dbs(&self) -> &IndexDbs {
        &self.index_dbs
    }

    /// Get the semantic encoder.
    pub const fn semantic_encoder(&self) -> &SemanticEncoder {
        &self.semantic_encoder
    }

    /// Rebuild the episodic DUP_SORT sidecar once per process when it is
    /// empty while authoritative records exist (legacy v1 stores and
    /// lost-sidecar recovery). Raw records are never modified; a failed
    /// rebuild leaves search on its raw-scan fallback.
    fn ensure_episodic_sidecar(&self) {
        if self.episodic_sidecar_ensured.get().is_some() {
            return;
        }
        let _ = self.episodic_sidecar_ensured.set(());
        let view = EpisodicStore::new(
            &self.env,
            self.episodic_db,
            self.episodic_terms_v2_db,
            self.episodic_term_cache.clone(),
            &self.mutation_count,
        );
        let needs_rebuild = matches!(
            (view.sidecar_is_empty(), view.record_count()),
            (Ok(true), Ok(n)) if n > 0
        );
        if needs_rebuild {
            match view.rebuild_sidecar() {
                Ok(n) => tracing::info!("episodic sidecar rebuilt from {n} records"),
                Err(e) => {
                    tracing::warn!("episodic sidecar rebuild failed: {e}");
                }
            }
        }
    }

    /// Open the v6 lossless episodic record view.
    #[must_use]
    pub fn episodic(&self) -> EpisodicStore<'_> {
        self.ensure_episodic_sidecar();
        let mut store = EpisodicStore::new(
            &self.env,
            self.episodic_db,
            self.episodic_terms_v2_db,
            self.episodic_term_cache.clone(),
            &self.mutation_count,
        );
        if let Some(Some(embedder)) = self.episodic_embedder.get() {
            store = store.with_embedder(embedder.clone());
        }
        if let Some(Some(aliases)) = self.episodic_aliases.get() {
            store = store.with_adaptive_aliases(aliases.clone());
        }
        if let Some(Some(enrichment)) = self.episodic_enrichment.get() {
            store = store.with_enrichment(enrichment.clone());
        }
        store
    }

    /// Attach an embedder for episodic vector reranking.
    pub fn set_episodic_embedder(
        &self,
        embedder: Arc<dyn crate::embedder::Embedder + Send + Sync>,
    ) {
        let _ = self.episodic_embedder.set(Some(embedder));
    }

    /// Attach adaptive aliases for episodic key expansion.
    pub fn set_episodic_aliases(&self, aliases: crate::episodic_keys::AdaptiveAliases) {
        let _ = self.episodic_aliases.set(Some(aliases));
    }

    /// Attach vocabulary enrichment for episodic index-time term expansion.
    pub fn set_episodic_enrichment(&self, enrichment: crate::enrichment::VocabularyEnrichment) {
        let _ = self.episodic_enrichment.set(Some(enrichment));
    }

    /// Get a named database handle for a galaxy.
    pub fn galaxy_db(&self, galaxy: Galaxy) -> Result<Database> {
        self.env.open_db(Some(galaxy.db_name())).map_err(|e| {
            CoreError::Memory(format!("LMDB open_db failed for {}: {e}", galaxy.db_name()))
        })
    }

    // ── Memory CRUD ───────────────────────────────────────────────────

    /// Store a memory in the given galaxy. Keyed by memory.metadata.id.
    /// Also updates all secondary indexes.
    ///
    /// Returns a clear error if the per-galaxy entry limit is exceeded
    /// or if the LMDB map is full.
    pub fn put(&self, galaxy: Galaxy, memory: &Memory) -> Result<()> {
        // Check per-galaxy entry limit (DoS prevention)
        if let Some(limit) = self.max_entries_per_galaxy {
            let current = self.count(galaxy)?;
            if current >= limit {
                return Err(CoreError::Memory(format!(
                    "galaxy {} entry limit reached ({current}/{limit}), write rejected",
                    galaxy.db_name()
                )));
            }
        }

        let db = self.galaxy_db(galaxy)?;
        let key = memory.metadata.id.as_bytes();
        let val = rmp_serde::to_vec_named(memory)
            .map_err(|e| CoreError::Memory(format!("serialize failed: {e}")))?;

        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;

        // Overwrite semantics: capture the previous record (if any) so its
        // index entries can be removed before the new ones are added.
        // Otherwise stale tags, importance values, timestamps, and content
        // hashes stay queryable after updates.
        let existing = tx
            .get(db, key)
            .ok()
            .and_then(|bytes| crate::codec::decode(bytes).ok());

        match tx.put(db, key, &val, lmdb::WriteFlags::default()) {
            Ok(()) => {}
            Err(lmdb::Error::MapFull) => {
                tx.abort();
                return Err(CoreError::Memory(format!(
                    "LMDB map full: galaxy {}, consider growing map size or pruning old memories",
                    galaxy.db_name()
                )));
            }
            Err(e) => {
                tx.abort();
                return Err(CoreError::Memory(format!("LMDB put failed: {e}")));
            }
        }
        if let Some(existing) = existing {
            self.index_dbs.remove(&mut tx, galaxy, &existing)?;
        }
        self.index_dbs.add(&mut tx, galaxy, memory)?;
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        self.mutation_count.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    /// Retrieve a memory by ID from the given galaxy.
    pub fn get(&self, galaxy: Galaxy, id: uuid::Uuid) -> Result<Option<Memory>> {
        let db = self.galaxy_db(galaxy)?;
        let key = id.as_bytes();

        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let result = tx.get(db, key);
        match result {
            Ok(bytes) => {
                let memory: Memory = crate::codec::decode(bytes)
                    .map_err(|e| CoreError::Memory(format!("deserialize failed: {e}")))?;
                tx.commit()
                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
                Ok(Some(memory))
            }
            Err(lmdb::Error::NotFound) => {
                // ReadOnly transactions don't strictly need commit, but it's good practice
                tx.commit()
                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
                Ok(None)
            }
            Err(e) => Err(CoreError::Memory(format!("LMDB get failed: {e}"))),
        }
    }

    /// Retrieve a memory by ID searching across all memory galaxies (S9 cross-galaxy traversal).
    ///
    /// Cross-galaxy associations and edges reference galaxy-blind UUIDs.
    /// This searches through all memory galaxies in canonical order and returns
    /// the first matching (Galaxy, Memory) pair, or None if not found.
    pub fn find_across_galaxies(&self, id: uuid::Uuid) -> Result<Option<(Galaxy, Memory)>> {
        for galaxy in Galaxy::memory_galaxies() {
            if let Some(mem) = self.get(galaxy, id)? {
                return Ok(Some((galaxy, mem)));
            }
        }
        Ok(None)
    }

    /// Delete a memory by ID from the given galaxy. Returns true if a key was removed.
    /// Also removes all secondary index entries for the memory.
    pub fn delete(&self, galaxy: Galaxy, id: uuid::Uuid) -> Result<bool> {
        let db = self.galaxy_db(galaxy)?;
        let key = id.as_bytes();

        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;

        // Check if key exists and deserialize for index cleanup
        let exists = tx.get(db, key).is_ok();
        if exists {
            // Read memory to get index values for cleanup
            if let Ok(bytes) = tx.get(db, key) {
                if let Ok(memory) = crate::codec::decode(bytes) {
                    let _ = self.index_dbs.remove(&mut tx, galaxy, &memory);
                }
            }
            tx.del(db, key, None)
                .map_err(|e| CoreError::Memory(format!("LMDB del failed: {e}")))?;
        }
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        if exists {
            self.mutation_count.fetch_add(1, Ordering::Relaxed);
        }
        Ok(exists)
    }

    /// Scan up to `limit` memories from the given galaxy (unordered by LMDB page layout).
    pub fn scan(&self, galaxy: Galaxy, limit: usize) -> Result<Vec<Memory>> {
        let db = self.galaxy_db(galaxy)?;
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;

        let mut cursor = tx
            .open_ro_cursor(db)
            .map_err(|e| CoreError::Memory(format!("LMDB cursor failed: {e}")))?;

        let mut memories = Vec::with_capacity(limit.min(256));
        for (i, (_key, val)) in cursor.iter().enumerate() {
            if memories.len() >= limit {
                break;
            }
            match crate::codec::decode(val) {
                Ok(memory) => memories.push(memory),
                Err(e) => {
                    tracing::warn!(
                        "Skipping corrupted entry at index {i} in galaxy {:?}: {e}",
                        galaxy
                    );
                }
            }
        }

        drop(cursor);
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        Ok(memories)
    }

    /// Scan every memory in the galaxy (unordered by LMDB page layout).
    ///
    /// Used by maintenance tooling (e.g. index rebuild). The full galaxy is
    /// materialized in memory — prefer [`Self::scan`] for bounded reads.
    pub fn scan_all(&self, galaxy: Galaxy) -> Result<Vec<Memory>> {
        self.scan_all_impl(galaxy, false)
    }

    /// Maintenance scan that refuses to omit an undecodable source record.
    /// Use before replacing derived indexes; a tolerant scan is not a complete
    /// authoritative snapshot when any record fails decoding.
    pub fn scan_all_strict(&self, galaxy: Galaxy) -> Result<Vec<Memory>> {
        self.scan_all_impl(galaxy, true)
    }

    fn scan_all_impl(&self, galaxy: Galaxy, strict: bool) -> Result<Vec<Memory>> {
        let db = self.galaxy_db(galaxy)?;
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;

        let mut cursor = tx
            .open_ro_cursor(db)
            .map_err(|e| CoreError::Memory(format!("LMDB cursor failed: {e}")))?;

        let mut memories = Vec::new();
        for (i, (_key, val)) in cursor.iter().enumerate() {
            match crate::codec::decode(val) {
                Ok(memory) => memories.push(memory),
                Err(e) => {
                    if strict {
                        return Err(CoreError::Memory(format!(
                            "refusing incomplete scan of {}: record {i} cannot be decoded: {e}",
                            galaxy.db_name()
                        )));
                    }
                    tracing::warn!(
                        "Skipping corrupted entry at index {i} in galaxy {:?}: {e}",
                        galaxy
                    );
                }
            }
        }

        drop(cursor);
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        Ok(memories)
    }

    /// Count entries in a galaxy.
    pub fn count(&self, galaxy: Galaxy) -> Result<usize> {
        let db = self.galaxy_db(galaxy)?;
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let mut cursor = tx
            .open_ro_cursor(db)
            .map_err(|e| CoreError::Memory(format!("LMDB cursor failed: {e}")))?;
        let count = cursor.iter().count();
        drop(cursor);
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        Ok(count)
    }

    /// Count entries in a galaxy carrying a tag, using the tag index (no
    /// record decoding). `wm status` uses this to report logical sessions
    /// (records tagged `start`) instead of every turn/checkpoint record
    /// stored in the Sessions galaxy.
    pub fn count_by_tag(&self, galaxy: Galaxy, tag: &str) -> Result<usize> {
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let ids = self.index_dbs.find_by_tag(&tx, galaxy, tag)?;
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        Ok(ids.len())
    }

    /// Clear all memories from a galaxy in a single transaction.
    /// Returns the number of entries cleared.
    /// Also removes all secondary index entries.
    pub fn clear_galaxy(&self, galaxy: Galaxy) -> Result<usize> {
        let db = self.galaxy_db(galaxy)?;

        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;

        let mut cursor = tx
            .open_ro_cursor(db)
            .map_err(|e| CoreError::Memory(format!("LMDB cursor failed: {e}")))?;

        let mut count = 0usize;
        let keys_to_delete: Vec<(Vec<u8>, Memory)> = cursor
            .iter()
            .filter_map(|(key, val)| {
                if let Ok(memory) = crate::codec::decode(val) {
                    Some((key.to_vec(), memory))
                } else {
                    None
                }
            })
            .collect();

        drop(cursor);

        for (key, memory) in &keys_to_delete {
            let _ = self.index_dbs.remove(&mut tx, galaxy, memory);
            tx.del(db, &key, None)
                .map_err(|e| CoreError::Memory(format!("LMDB del failed: {e}")))?;
            count += 1;
        }

        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        self.mutation_count
            .fetch_add(count as u64, Ordering::Relaxed);
        Ok(count)
    }

    /// Put multiple memories into a galaxy in a single transaction.
    /// Returns the number of memories written.
    pub fn batch_put(&self, galaxy: Galaxy, memories: &[Memory]) -> Result<usize> {
        if memories.is_empty() {
            return Ok(0);
        }

        let db = self.galaxy_db(galaxy)?;

        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;

        let mut count = 0usize;
        for memory in memories {
            let key = memory.metadata.id.as_bytes();
            let val = rmp_serde::to_vec_named(memory)
                .map_err(|e| CoreError::Memory(format!("serialize failed: {e}")))?;
            match tx.put(db, key, &val, WriteFlags::default()) {
                Ok(()) => {}
                Err(lmdb::Error::MapFull) => {
                    tx.abort();
                    return Err(CoreError::Memory(format!(
                        "LMDB map full: galaxy {}, consider growing map size",
                        galaxy.db_name()
                    )));
                }
                Err(e) => {
                    tx.abort();
                    return Err(CoreError::Memory(format!("LMDB put failed: {e}")));
                }
            }
            self.index_dbs.add(&mut tx, galaxy, memory)?;
            count += 1;
        }

        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        self.mutation_count
            .fetch_add(count as u64, Ordering::Relaxed);
        Ok(count)
    }

    /// Get raw key-value bytes (for advanced use cases).
    pub fn get_raw(&self, galaxy: Galaxy, key: &[u8]) -> Result<Option<Vec<u8>>> {
        let db = self.galaxy_db(galaxy)?;
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        match tx.get(db, &key) {
            Ok(bytes) => {
                let data = bytes.to_vec();
                tx.commit()
                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
                Ok(Some(data))
            }
            Err(lmdb::Error::NotFound) => {
                tx.commit()
                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
                Ok(None)
            }
            Err(e) => Err(CoreError::Memory(format!("LMDB get_raw failed: {e}"))),
        }
    }

    /// Put raw key-value bytes (for advanced use cases).
    pub fn put_raw(&self, galaxy: Galaxy, key: &[u8], val: &[u8]) -> Result<()> {
        let db = self.galaxy_db(galaxy)?;
        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
        tx.put(db, &key, &val, lmdb::WriteFlags::default())
            .map_err(|e| CoreError::Memory(format!("LMDB put_raw failed: {e}")))?;
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        self.mutation_count.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    /// Delete raw key-value bytes (for advanced use cases).
    /// Returns true if a key was removed.
    pub fn delete_raw(&self, galaxy: Galaxy, key: &[u8]) -> Result<bool> {
        let db = self.galaxy_db(galaxy)?;
        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
        let deleted = tx.del(db, &key, None).is_ok();
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        if deleted {
            self.mutation_count.fetch_add(1, Ordering::Relaxed);
        }
        Ok(deleted)
    }

    /// Batch multiple raw key-value writes in a single LMDB transaction.
    /// All writes succeed or fail atomically.
    pub fn put_raw_batch(&self, galaxy: Galaxy, entries: &[(&[u8], &[u8])]) -> Result<()> {
        self.put_raw_batch_impl(galaxy, entries)?;
        self.mutation_count
            .fetch_add(entries.len() as u64, Ordering::Relaxed);
        Ok(())
    }

    /// Batch multiple raw key-value writes without advancing the mutation
    /// counter.
    ///
    /// For governance bookkeeping (karma chain, write-audit journal): these
    /// writes are metadata *about* dispatches, not memory mutations. Letting
    /// them tick the counter attributes a whole batch flush to whichever
    /// dispatch happens to be in flight when the threshold trips — the
    /// 2026-08-28 restore-drill false-positive class ("read-only" tools
    /// flagged with the previous batch's size as their write delta).
    pub fn put_raw_batch_untracked(
        &self,
        galaxy: Galaxy,
        entries: &[(&[u8], &[u8])],
    ) -> Result<()> {
        self.put_raw_batch_impl(galaxy, entries)
    }

    fn put_raw_batch_impl(&self, galaxy: Galaxy, entries: &[(&[u8], &[u8])]) -> Result<()> {
        let db = self.galaxy_db(galaxy)?;
        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
        for (key, val) in entries {
            tx.put(db, key, val, WriteFlags::default())
                .map_err(|e| CoreError::Memory(format!("LMDB put_raw_batch failed: {e}")))?;
        }
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        Ok(())
    }

    // ── Content-hash Deduplication ────────────────────────────────────

    /// Check if a memory with the same content hash already exists in the galaxy.
    /// Uses the content_hash index for O(1) lookup.
    /// Returns the existing memory's ID if found.
    pub fn find_by_content_hash(&self, galaxy: Galaxy, hash: &str) -> Result<Option<uuid::Uuid>> {
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let result = self.index_dbs.find_by_content_hash(&tx, galaxy, hash)?;
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        Ok(result)
    }

    /// Scan-based content hash lookup (O(n) fallback, used for testing index correctness).
    pub fn find_by_content_hash_scan(
        &self,
        galaxy: Galaxy,
        hash: &str,
    ) -> Result<Option<uuid::Uuid>> {
        let memories = self.scan(galaxy, 10_000)?;
        for mem in memories {
            if mem.metadata.content_hash == hash {
                return Ok(Some(mem.metadata.id));
            }
        }
        Ok(None)
    }

    /// Store a memory with content-hash deduplication.
    /// If a memory with the same content already exists in the galaxy,
    /// returns the existing memory's ID without creating a duplicate.
    pub fn put_dedup(&self, galaxy: Galaxy, memory: &Memory) -> Result<uuid::Uuid> {
        if let Some(existing_id) =
            self.find_by_content_hash(galaxy, &memory.metadata.content_hash)?
        {
            return Ok(existing_id);
        }
        let id = memory.metadata.id;
        self.put(galaxy, memory)?;
        Ok(id)
    }

    // ── Write Batching ─────────────────────────────────────────────────

    /// Store multiple memories in a single LMDB transaction (batch write).
    /// All writes and index updates succeed or fail atomically.
    pub fn put_batch(&self, galaxy: Galaxy, memories: &[Memory]) -> Result<()> {
        let db = self.galaxy_db(galaxy)?;
        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;

        for memory in memories {
            let key = memory.metadata.id.as_bytes();
            let val = rmp_serde::to_vec_named(memory)
                .map_err(|e| CoreError::Memory(format!("serialize failed: {e}")))?;
            tx.put(db, key, &val, WriteFlags::default())
                .map_err(|e| CoreError::Memory(format!("LMDB put_batch failed: {e}")))?;
            self.index_dbs.add(&mut tx, galaxy, memory)?;
        }

        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        self.mutation_count
            .fetch_add(memories.len() as u64, Ordering::Relaxed);
        Ok(())
    }

    // ── Query API ──────────────────────────────────────────────────────

    /// Query memories in a galaxy with filtering.
    /// Uses secondary indexes when the query is a pure single-dimension filter
    /// (single tag, importance range, or time range with no other filters).
    /// Falls back to scan for complex multi-dimensional queries.
    pub fn query(&self, galaxy: Galaxy, query: &MemoryQuery) -> Result<Vec<Memory>> {
        // Try indexed fast paths for single-dimension queries. A substring
        // filter forces the full scan — the indexes cannot evaluate it.
        if query.content_substring.is_none()
            && query.tags.len() == 1
            && query.min_importance.is_none()
            && query.max_importance.is_none()
            && query.created_after.is_none()
            && query.created_before.is_none()
        {
            return self.query_by_tag_indexed(galaxy, &query.tags[0], query.limit);
        }

        if query.content_substring.is_none()
            && query.tags.is_empty()
            && let Some(min) = query.min_importance
            && let Some(max) = query.max_importance
            && query.created_after.is_none()
            && query.created_before.is_none()
        {
            return self.query_by_importance_indexed(galaxy, min, max, query.limit);
        }

        if query.content_substring.is_none()
            && query.tags.is_empty()
            && query.min_importance.is_none()
            && query.max_importance.is_none()
            && let Some(after) = query.created_after
            && let Some(before) = query.created_before
        {
            return self.query_by_time_indexed(galaxy, after, before, query.limit);
        }

        // Fallback: full scan with in-memory filter
        let memories = self.scan(galaxy, 10_000)?;
        let mut results = Vec::new();
        for mem in memories {
            if query.matches(&mem) {
                results.push(mem);
                if results.len() >= query.limit {
                    break;
                }
            }
        }
        Ok(results)
    }

    /// Tag-based indexed query → memories with the given tag.
    fn query_by_tag_indexed(&self, galaxy: Galaxy, tag: &str, limit: usize) -> Result<Vec<Memory>> {
        let db = self.galaxy_db(galaxy)?;
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let ids = self.index_dbs.find_by_tag(&tx, galaxy, tag)?;
        let mut results = Vec::new();
        for id in &ids {
            if results.len() >= limit {
                break;
            }
            if let Ok(bytes) = tx.get(db, id.as_bytes()) {
                if let Ok(mem) = crate::codec::decode(bytes) {
                    results.push(mem);
                }
            }
        }
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        Ok(results)
    }

    /// Importance-range indexed query → memories with importance in [min, max].
    fn query_by_importance_indexed(
        &self,
        galaxy: Galaxy,
        min: f32,
        max: f32,
        limit: usize,
    ) -> Result<Vec<Memory>> {
        let db = self.galaxy_db(galaxy)?;
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let ids = self
            .index_dbs
            .find_by_importance_range(&tx, galaxy, min, max)?;
        let mut results = Vec::new();
        for id in &ids {
            if results.len() >= limit {
                break;
            }
            if let Ok(bytes) = tx.get(db, id.as_bytes()) {
                if let Ok(mem) = crate::codec::decode(bytes) {
                    results.push(mem);
                }
            }
        }
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        Ok(results)
    }

    /// Time-range indexed query → memories created in [after, before].
    fn query_by_time_indexed(
        &self,
        galaxy: Galaxy,
        after: chrono::DateTime<chrono::Utc>,
        before: chrono::DateTime<chrono::Utc>,
        limit: usize,
    ) -> Result<Vec<Memory>> {
        let db = self.galaxy_db(galaxy)?;
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let ids = self
            .index_dbs
            .find_by_time_range(&tx, galaxy, after, before)?;
        let mut results = Vec::new();
        for id in &ids {
            if results.len() >= limit {
                break;
            }
            if let Ok(bytes) = tx.get(db, id.as_bytes()) {
                if let Ok(mem) = crate::codec::decode(bytes) {
                    results.push(mem);
                }
            }
        }
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        Ok(results)
    }

    // ── Semantic Coordinate Encoding ─────────────────────────────────────

    /// Store a memory with semantically-derived 5D coordinates.
    ///
    /// Replaces the SHA-256 hash-based `Coordinate5D::encode()` with
    /// anchor-based TF projection. The memory's `coord5d` field is updated
    /// with semantically meaningful x/y/z values before storage.
    pub fn put_semantic(&self, galaxy: Galaxy, memory: &mut Memory) -> Result<()> {
        let temporal_weight = memory.metadata.coord5d.w;
        let importance = memory.metadata.importance;
        memory.metadata.coord5d =
            self.semantic_encoder
                .encode_coordinate(&memory.content, temporal_weight, importance);
        self.put(galaxy, memory)
    }

    /// Find memories in a galaxy with content semantically similar to the query text.
    ///
    /// Encodes the query text into a 5D coordinate and scans the galaxy,
    /// returning memories sorted by semantic distance (nearest first).
    pub fn find_similar(
        &self,
        galaxy: Galaxy,
        query_text: &str,
        limit: usize,
    ) -> Result<Vec<(Memory, f32)>> {
        let query_coord = self
            .semantic_encoder
            .encode_coordinate(query_text, 0.5, 0.5);
        let memories = self.scan(galaxy, 10_000)?;
        let mut results: Vec<(Memory, f32)> = memories
            .into_iter()
            .map(|m| {
                let dist = query_coord.semantic_distance_to(&m.metadata.coord5d);
                (m, dist)
            })
            .collect();
        results.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal));
        results.truncate(limit);
        Ok(results)
    }

    // ── Embedding Storage ──────────────────────────────────────────────

    /// Store an embedding vector for a memory in the Embeddings galaxy.
    /// Keyed by the memory's UUID.
    pub fn put_embedding(&self, memory_id: uuid::Uuid, embedding: &[f32]) -> Result<()> {
        let db = self.galaxy_db(Galaxy::Embeddings)?;
        let key = memory_id.as_bytes();
        let val = encode_embedding(embedding);

        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
        tx.put(db, key, &val, WriteFlags::default())
            .map_err(|e| CoreError::Memory(format!("LMDB put_embedding failed: {e}")))?;
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        self.mutation_count.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    /// Retrieve an embedding vector for a memory from the Embeddings galaxy.
    pub fn get_embedding(&self, memory_id: uuid::Uuid) -> Result<Option<Vec<f32>>> {
        let db = self.galaxy_db(Galaxy::Embeddings)?;
        let key = memory_id.as_bytes();

        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        match tx.get(db, key) {
            Ok(bytes) => {
                let embedding = decode_embedding(bytes);
                tx.commit()
                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
                Ok(Some(embedding))
            }
            Err(lmdb::Error::NotFound) => {
                tx.commit()
                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
                Ok(None)
            }
            Err(e) => Err(CoreError::Memory(format!("LMDB get_embedding failed: {e}"))),
        }
    }

    /// Delete an embedding vector from the Embeddings galaxy.
    pub fn delete_embedding(&self, memory_id: uuid::Uuid) -> Result<bool> {
        let db = self.galaxy_db(Galaxy::Embeddings)?;
        let key = memory_id.as_bytes();

        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
        let exists = tx.get(db, key).is_ok();
        if exists {
            tx.del(db, key, None)
                .map_err(|e| CoreError::Memory(format!("LMDB del_embedding failed: {e}")))?;
        }
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        if exists {
            self.mutation_count.fetch_add(1, Ordering::Relaxed);
        }
        Ok(exists)
    }

    // ── Embedding cache (content-hash → vector) ────────────────────────

    /// Store a cached embedding under a caller-computed cache key
    /// (embedder namespace + content hash). Vectors for the same content
    /// differ across models, so the key must carry the namespace.
    pub fn put_embedding_cache(&self, cache_key: &str, embedding: &[f32]) -> Result<()> {
        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
        tx.put(
            self.embedding_cache_db,
            &cache_key.as_bytes().to_vec(),
            &encode_embedding(embedding),
            WriteFlags::default(),
        )
        .map_err(|e| CoreError::Memory(format!("LMDB put_embedding_cache failed: {e}")))?;
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        self.mutation_count.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    /// Batched cache write: one transaction for the whole ingest chunk.
    pub fn put_embedding_cache_batch(&self, entries: &[(String, Vec<f32>)]) -> Result<()> {
        if entries.is_empty() {
            return Ok(());
        }
        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
        for (key, embedding) in entries {
            tx.put(
                self.embedding_cache_db,
                &key.as_bytes().to_vec(),
                &encode_embedding(embedding),
                WriteFlags::default(),
            )
            .map_err(|e| CoreError::Memory(format!("LMDB put_embedding_cache failed: {e}")))?;
        }
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        self.mutation_count
            .fetch_add(entries.len() as u64, Ordering::Relaxed);
        Ok(())
    }

    /// Look up a cached embedding. `Ok(None)` = miss; the caller embeds.
    pub fn get_embedding_cache(&self, cache_key: &str) -> Result<Option<Vec<f32>>> {
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        match tx.get(self.embedding_cache_db, &cache_key.as_bytes().to_vec()) {
            Ok(bytes) => {
                let embedding = decode_embedding(bytes);
                tx.commit()
                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
                Ok(Some(embedding))
            }
            Err(lmdb::Error::NotFound) => {
                tx.commit()
                    .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
                Ok(None)
            }
            Err(e) => Err(CoreError::Memory(format!(
                "LMDB get_embedding_cache failed: {e}"
            ))),
        }
    }

    /// Batched lookup: one read transaction for the whole ingest chunk.
    /// Result aligns 1:1 with `keys`.
    pub fn get_embedding_cache_batch(&self, keys: &[String]) -> Result<Vec<Option<Vec<f32>>>> {
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let mut out = Vec::with_capacity(keys.len());
        for key in keys {
            out.push(
                tx.get(self.embedding_cache_db, &key.as_bytes().to_vec())
                    .ok()
                    .map(decode_embedding),
            );
        }
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        Ok(out)
    }

    /// Number of cached vectors (doctor / honesty surfaces).
    pub fn embedding_cache_count(&self) -> Result<u64> {
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let mut cursor = tx
            .open_ro_cursor(self.embedding_cache_db)
            .map_err(|e| CoreError::Memory(format!("LMDB cursor embedding_cache failed: {e}")))?;
        let mut count = 0u64;
        for _ in cursor.iter() {
            count += 1;
        }
        Ok(count)
    }

    // ── Revision chain (V8 S11c) ────────────────────────────────────────

    /// Append one content revision to a memory's chain. Seq is derived
    /// from the current chain tail (append-only by convention); the write
    /// bumps the store mutation counter so dispatch windows attribute it.
    pub fn record_revision(
        &self,
        galaxy: Galaxy,
        id: MemoryId,
        old_hash: &str,
        new_hash: &str,
        actor: crate::revision::RevisionActor,
    ) -> Result<crate::revision::MemoryRevision> {
        let seq = self.revisions(galaxy, id)?.len() as u32;
        let entry = crate::revision::MemoryRevision {
            seq,
            timestamp: wm_core::time::now_unix_secs(),
            old_hash: old_hash.to_string(),
            new_hash: new_hash.to_string(),
            actor_session: actor.session,
            actor_user: actor.user,
            actor_compartment: actor.compartment,
        };
        let key = crate::revision::revision_key(galaxy, id, seq);
        let val = serde_json::to_vec(&entry)
            .map_err(|e| CoreError::Memory(format!("revision serialize failed: {e}")))?;
        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
        tx.put(self.revisions_db, &key, &val, WriteFlags::default())
            .map_err(|e| CoreError::Memory(format!("LMDB put revision failed: {e}")))?;
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        self.mutation_count.fetch_add(1, Ordering::Relaxed);
        Ok(entry)
    }

    /// Full revision chain for a memory, ordered by seq. Empty for
    /// memories never content-updated (or pre-S11c).
    pub fn revisions(
        &self,
        galaxy: Galaxy,
        id: MemoryId,
    ) -> Result<Vec<crate::revision::MemoryRevision>> {
        // Cursor-op constants from lmdb.h (frozen LMDB ABI): the `lmdb`
        // crate's `iter_from` panics on a SetRange miss, and a miss is a
        // normal state here (a memory with no revisions yet sorts after
        // every existing key), so the cursor is driven manually.
        const MDB_GET_CURRENT: u32 = 4;
        const MDB_NEXT: u32 = 8;
        const MDB_SET_RANGE: u32 = 17;
        let prefix = crate::revision::revision_prefix(galaxy, id);
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let cursor = tx
            .open_ro_cursor(self.revisions_db)
            .map_err(|e| CoreError::Memory(format!("LMDB cursor revisions failed: {e}")))?;
        let mut out = Vec::new();
        if cursor.get(Some(&prefix), None, MDB_SET_RANGE).is_ok() {
            while let Ok((key, val)) = cursor.get(None, None, MDB_GET_CURRENT) {
                // `key` is None only for ops that return no key — GET_CURRENT
                // after a positioned read always carries one.
                if !key.is_some_and(|k| k.starts_with(&prefix)) {
                    break;
                }
                let entry: crate::revision::MemoryRevision = serde_json::from_slice(val)
                    .map_err(|e| CoreError::Memory(format!("revision deserialize failed: {e}")))?;
                out.push(entry);
                // Bounded walk — corrupt data can never spin this loop.
                if out.len() >= 10_000 || cursor.get(None, None, MDB_NEXT).is_err() {
                    break;
                }
            }
        }
        drop(cursor);
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        Ok(out)
    }

    /// Walk one memory's revision chain and grade it against the memory's
    /// current content hash (seq continuity, hash linkage, head match).
    pub fn verify_revision_chain(
        &self,
        galaxy: Galaxy,
        id: MemoryId,
        current_hash: &str,
    ) -> Result<crate::revision::RevisionChainReport> {
        let entries = self.revisions(galaxy, id)?;
        Ok(crate::revision::verify_chain(&entries, current_hash))
    }

    // ── Record attestations (Track F Slice A, D5) ───────────────────────

    /// Record one creation attestation for a memory. Upsert by key
    /// (`att:{galaxy}:{memory_id}`): re-attestation overwrites, which is
    /// safe because attestation covers the *creation* event and memory ids
    /// are unique per create. The write bumps the mutation counter like
    /// every other store mutation.
    pub fn record_attestation(
        &self,
        galaxy: Galaxy,
        id: MemoryId,
        entry: &crate::attestation::RecordAttestation,
    ) -> Result<()> {
        let key = crate::attestation::attestation_key(galaxy, id);
        let val = serde_json::to_vec(entry)
            .map_err(|e| CoreError::Memory(format!("attestation serialize failed: {e}")))?;
        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
        tx.put(self.attestations_db, &key, &val, WriteFlags::default())
            .map_err(|e| CoreError::Memory(format!("LMDB put attestation failed: {e}")))?;
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        self.mutation_count.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    /// This memory's creation attestation, if the creating dispatch signed
    /// one (key-available creates only — absence is honest, not an error).
    pub fn attestation(
        &self,
        galaxy: Galaxy,
        id: MemoryId,
    ) -> Result<Option<crate::attestation::RecordAttestation>> {
        let key = crate::attestation::attestation_key(galaxy, id);
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let out =
            match tx.get(self.attestations_db, &key) {
                Ok(val) => Some(serde_json::from_slice(val).map_err(|e| {
                    CoreError::Memory(format!("attestation deserialize failed: {e}"))
                })?),
                Err(lmdb::Error::NotFound) => None,
                Err(e) => {
                    return Err(CoreError::Memory(format!(
                        "LMDB get attestation failed: {e}"
                    )));
                }
            };
        drop(tx);
        Ok(out)
    }

    /// Every attestation in the store (for `wm anchor`). Bounded walk —
    /// corrupt data can never spin this loop.
    pub fn scan_attestations(&self) -> Result<Vec<crate::attestation::RecordAttestation>> {
        const MDB_GET_CURRENT: u32 = 4;
        const MDB_NEXT: u32 = 8;
        const MDB_FIRST: u32 = 9;
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let cursor = tx
            .open_ro_cursor(self.attestations_db)
            .map_err(|e| CoreError::Memory(format!("LMDB cursor attestations failed: {e}")))?;
        let mut out = Vec::new();
        if cursor.get(None, None, MDB_FIRST).is_ok() {
            while let Ok((_, val)) = cursor.get(None, None, MDB_GET_CURRENT) {
                let entry: crate::attestation::RecordAttestation = serde_json::from_slice(val)
                    .map_err(|e| {
                        CoreError::Memory(format!("attestation deserialize failed: {e}"))
                    })?;
                out.push(entry);
                // Bounded walk — corrupt data can never spin this loop.
                if out.len() >= 1_000_000 || cursor.get(None, None, MDB_NEXT).is_err() {
                    break;
                }
            }
        }
        drop(cursor);
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        Ok(out)
    }

    /// Grade one memory's attestation: presence, signature validity, and
    /// whether the attested hash still matches the live content hash.
    /// A content *update* after creation flips `matches_head` to false by
    /// design — updates are covered by the revisions chain, not by
    /// re-attestation.
    pub fn verify_attestation(
        &self,
        galaxy: Galaxy,
        id: MemoryId,
    ) -> Result<crate::attestation::AttestationReport> {
        use crate::attestation::AttestationReport;
        let Some(att) = self.attestation(galaxy, id)? else {
            return Ok(AttestationReport {
                attested: false,
                signature_valid: false,
                matches_head: false,
                memory_present: self.get(galaxy, id)?.is_some(),
                breaks: vec!["no attestation recorded for this memory".to_string()],
            });
        };
        let mut breaks = Vec::new();
        let signature_valid = crate::attestation::verify_attestation(&att);
        if !signature_valid {
            breaks.push("signature does not verify against recorded pubkey".to_string());
        }
        let (matches_head, memory_present) = if let Some(memory) = self.get(galaxy, id)? {
            let matches = memory.metadata.content_hash == att.record_hash;
            if !matches {
                breaks.push(
                    "attested record_hash != live content_hash (memory updated after attestation)"
                        .to_string(),
                );
            }
            (matches, true)
        } else {
            breaks.push("attested memory id not present in galaxy".to_string());
            (false, false)
        };
        Ok(AttestationReport {
            attested: true,
            signature_valid,
            matches_head,
            memory_present,
            breaks,
        })
    }

    /// Grade every attestation in the store (for `wm anchor`). Entries
    /// whose galaxy/id no longer parse are reported as broken — never
    /// skipped silently.
    pub fn attestation_sweep(
        &self,
    ) -> Result<
        Vec<(
            crate::attestation::RecordAttestation,
            crate::attestation::AttestationReport,
        )>,
    > {
        use crate::attestation::AttestationReport;
        let mut out = Vec::new();
        for att in self.scan_attestations()? {
            let parsed = match (
                Galaxy::from_db_name(&att.galaxy),
                uuid::Uuid::parse_str(&att.memory_id),
            ) {
                (Some(galaxy), Ok(id)) => Some((galaxy, id)),
                _ => None,
            };
            match parsed {
                Some((galaxy, id)) => out.push((att, self.verify_attestation(galaxy, id)?)),
                None => out.push((
                    att,
                    AttestationReport {
                        attested: true,
                        signature_valid: false,
                        matches_head: false,
                        memory_present: false,
                        breaks: vec![
                            "attestation row has unparseable galaxy or memory id".to_string(),
                        ],
                    },
                )),
            }
        }
        Ok(out)
    }

    // ── Non-Destructive Phagic Cold-Storage (the project's sacred rule) ───────

    /// Store a compressed cold record in the cold_storage DBI.
    pub fn put_cold_record(&self, record: &crate::cold_storage::ColdRecord) -> Result<()> {
        let key = record.id.as_bytes();
        let val = rmp_serde::to_vec_named(record)
            .map_err(|e| CoreError::Memory(format!("Cold record serialization failed: {e}")))?;
        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
        tx.put(self.cold_storage_db, key, &val, WriteFlags::default())
            .map_err(|e| CoreError::Memory(format!("LMDB put cold_storage failed: {e}")))?;
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        self.mutation_count.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    /// Retrieve a compressed cold record by memory id.
    pub fn get_cold_record(&self, id: MemoryId) -> Result<Option<crate::cold_storage::ColdRecord>> {
        let key = id.as_bytes();
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        match tx.get(self.cold_storage_db, key) {
            Ok(bytes) => {
                let record: crate::cold_storage::ColdRecord = rmp_serde::from_slice(bytes)
                    .map_err(|e| {
                        CoreError::Memory(format!("Cold record deserialization failed: {e}"))
                    })?;
                Ok(Some(record))
            }
            Err(lmdb::Error::NotFound) => Ok(None),
            Err(e) => Err(CoreError::Memory(format!(
                "LMDB get cold_storage failed: {e}"
            ))),
        }
    }

    /// Delete a cold record from the cold storage DBI (used when thawing back to hot).
    pub fn delete_cold_record(&self, id: MemoryId) -> Result<bool> {
        let key = id.as_bytes();
        let mut tx = self
            .env
            .begin_rw_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB rw_txn failed: {e}")))?;
        let deleted = match tx.del(self.cold_storage_db, key, None) {
            Ok(()) => true,
            Err(lmdb::Error::NotFound) => false,
            Err(e) => {
                return Err(CoreError::Memory(format!(
                    "LMDB del cold_storage failed: {e}"
                )));
            }
        };
        tx.commit()
            .map_err(|e| CoreError::Memory(format!("LMDB commit failed: {e}")))?;
        if deleted {
            self.mutation_count.fetch_add(1, Ordering::Relaxed);
        }
        Ok(deleted)
    }

    /// Count cold records in the cold storage database, optionally filtered by galaxy.
    pub fn count_cold(&self, galaxy: Option<Galaxy>) -> Result<usize> {
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let mut cursor = tx
            .open_ro_cursor(self.cold_storage_db)
            .map_err(|e| CoreError::Memory(format!("LMDB open_ro_cursor failed: {e}")))?;
        let mut count = 0;
        for (_key, val) in cursor.iter() {
            if let Some(target_g) = galaxy {
                let record: crate::cold_storage::ColdRecord =
                    rmp_serde::from_slice(val).map_err(|e| {
                        CoreError::Memory(format!("Cold record deserialization failed: {e}"))
                    })?;
                if record.galaxy == target_g {
                    count += 1;
                }
            } else {
                count += 1;
            }
        }
        Ok(count)
    }

    /// List cold records (summaries) with optional galaxy filter and limit.
    pub fn list_cold_records(
        &self,
        galaxy: Option<Galaxy>,
        limit: usize,
    ) -> Result<Vec<crate::cold_storage::ColdRecordSummary>> {
        let query = crate::cold_storage::ColdQuery {
            galaxy,
            limit: if limit == 0 { 100 } else { limit },
            ..Default::default()
        };
        self.query_cold_records(&query)
    }

    /// Query cold records matching a `ColdQuery` filter.
    pub fn query_cold_records(
        &self,
        query: &crate::cold_storage::ColdQuery,
    ) -> Result<Vec<crate::cold_storage::ColdRecordSummary>> {
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let mut cursor = tx
            .open_ro_cursor(self.cold_storage_db)
            .map_err(|e| CoreError::Memory(format!("LMDB open_ro_cursor failed: {e}")))?;
        let mut results = Vec::new();
        let limit = if query.limit == 0 {
            usize::MAX
        } else {
            query.limit
        };

        for (_key, val) in cursor.iter() {
            let record: crate::cold_storage::ColdRecord =
                rmp_serde::from_slice(val).map_err(|e| {
                    CoreError::Memory(format!("Cold record deserialization failed: {e}"))
                })?;
            let summary = record.summary();
            if query.matches(&summary) {
                results.push(summary);
                if results.len() >= limit {
                    break;
                }
            }
        }
        Ok(results)
    }

    /// Bounded, identity-bound cold discovery.
    ///
    /// Scans at most `max_scan` cold records (LMDB `cold_storage` DBI),
    /// filters by galaxy when given, decompresses each candidate, verifies
    /// the id/galaxy/content-hash chain, applies visibility (private never
    /// surfaces; superseded/non-current records are skipped), and returns
    /// up to `limit` full cold records whose content or tags contain every
    /// query term (case-insensitive). Nothing is thawed or mutated.
    pub fn find_cold_matching(
        &self,
        terms: &[String],
        galaxy: Option<Galaxy>,
        limit: usize,
        max_scan: usize,
    ) -> Result<crate::cold_storage::ColdDiscoveryOutcome> {
        self.find_cold_matching_eligible(terms, galaxy, limit, max_scan, |_| true)
    }

    /// Apply caller eligibility to verified payloads before consuming result
    /// capacity. Rejected matches still consume the bounded scan budget.
    pub fn find_cold_matching_eligible(
        &self,
        terms: &[String],
        galaxy: Option<Galaxy>,
        limit: usize,
        max_scan: usize,
        eligible: impl Fn(&Memory) -> bool,
    ) -> Result<crate::cold_storage::ColdDiscoveryOutcome> {
        use crate::cold_storage::ColdDiscoveryStop;
        let mut out = crate::cold_storage::ColdDiscoveryOutcome::default();
        if terms.is_empty() || limit == 0 || max_scan == 0 {
            return Ok(out);
        }
        let tx = self
            .env
            .begin_ro_txn()
            .map_err(|e| CoreError::Memory(format!("LMDB ro_txn failed: {e}")))?;
        let mut cursor = tx
            .open_ro_cursor(self.cold_storage_db)
            .map_err(|e| CoreError::Memory(format!("LMDB open_ro_cursor failed: {e}")))?;
        let mut iter = cursor.iter();
        loop {
            if out.scanned >= max_scan {
                out.stop_reason = ColdDiscoveryStop::ScanLimit;
                break;
            }
            if out.records.len() >= limit {
                out.stop_reason = ColdDiscoveryStop::ResultLimit;
                break;
            }
            let Some((key, val)) = iter.next() else {
                out.stop_reason = ColdDiscoveryStop::Exhausted;
                break;
            };
            out.scanned += 1;
            let record: crate::cold_storage::ColdRecord = if let Ok(r) = rmp_serde::from_slice(val)
            {
                r
            } else {
                out.integrity_rejected += 1;
                continue;
            };
            if let Some(g) = galaxy {
                if record.galaxy != g {
                    continue;
                }
            }
            out.candidates += 1;
            let mem = if let Ok(m) = record.decompress() {
                m
            } else {
                out.integrity_rejected += 1;
                continue;
            };
            if mem.metadata.is_private {
                out.private_skipped += 1;
                continue;
            }
            if !mem.metadata.validity.is_current() {
                out.non_current_skipped += 1;
                continue;
            }
            let integrity_ok = key == record.id.as_bytes()
                && mem.metadata.id == record.id
                && mem.metadata.galaxy == record.galaxy
                && mem.metadata.content_hash == record.content_hash
                && crate::content_hash(&mem.content) == record.content_hash;
            if !integrity_ok {
                out.integrity_rejected += 1;
                continue;
            }
            let haystack = format!(
                "{} {}",
                mem.content.to_lowercase(),
                mem.metadata.tags.join(" ").to_lowercase()
            );
            if !terms.iter().all(|t| haystack.contains(t.as_str())) {
                continue;
            }
            out.matched += 1;
            if !eligible(&mem) {
                out.eligibility_skipped += 1;
                continue;
            }
            out.records.push(record);
        }
        Ok(out)
    }

    /// Freeze an active hot memory into the compressed cold archive.
    ///
    /// Non-destructive: preserves complete metadata, vector clocks, content, embeddings,
    /// and provenance. The memory transitions to `Tier::Archival`, is stored in `cold_storage_db`,
    /// is deindexed from Tantivy (if `search` provided), and is removed from the active hot galaxy.
    pub fn freeze_to_cold(
        &self,
        search: Option<&crate::SearchEngine>,
        memory_id: MemoryId,
        distance: f32,
        factors: crate::cold_storage::OuterRimFactors,
        digest_id: Option<MemoryId>,
        notes: Option<String>,
        codec: crate::cold_storage::CompressionCodec,
    ) -> Result<crate::cold_storage::ColdRecord> {
        let (galaxy, mut mem) = self.find_across_galaxies(memory_id)?.ok_or_else(|| {
            CoreError::NotFound(format!("Memory {memory_id} not found in hot store"))
        })?;

        // Transition tier to Archival
        if mem.metadata.tier != crate::memory::Tier::Archival {
            let _ = mem.transition_tier(crate::memory::Tier::Archival);
        }

        let record =
            crate::cold_storage::ColdRecord::new(&mem, distance, factors, digest_id, notes, codec)?;

        // Store into cold archive
        self.put_cold_record(&record)?;

        // Remove from active hot galaxy
        self.delete(galaxy, memory_id)?;

        // Deindex from Tantivy search engine if provided
        if let Some(engine) = search {
            if let Ok(mut writer_guard) = engine.writer() {
                let _ = engine.delete_document(&mut writer_guard, &memory_id.to_string());
                let _ = engine.commit(&mut writer_guard);
            }
        }

        Ok(record)
    }

    /// Thaw a memory from compressed cold storage back into the hot active tier.
    ///
    /// Zero data loss: restores the original memory with all fields, transitions tier
    /// back to `Tier::Episodic`, bumps access/recall count, updates `accessed_at`,
    /// stores into the hot galaxy, and reindexes into Tantivy search (if provided).
    pub fn thaw_from_cold(
        &self,
        search: Option<&crate::SearchEngine>,
        memory_id: MemoryId,
    ) -> Result<Memory> {
        let record = self.get_cold_record(memory_id)?.ok_or_else(|| {
            CoreError::NotFound(format!("Memory {memory_id} not found in cold storage"))
        })?;

        let mut mem = record.decompress()?;

        // Transition tier back to Episodic (warm serving)
        let _ = mem.transition_tier(crate::memory::Tier::Episodic);
        mem.metadata.accessed_at = chrono::Utc::now();
        mem.metadata.access_count += 1;
        mem.metadata.recall_count += 1;
        if !mem.metadata.tags.iter().any(|t| t == "thawed:phagic") {
            mem.metadata.tags.push("thawed:phagic".to_string());
        }

        // Put back into active hot galaxy
        self.put(record.galaxy, &mem)?;

        // Reindex in Tantivy if provided
        if let Some(engine) = search {
            if let Ok(mut writer_guard) = engine.writer() {
                let _ = engine.index_memory(&mut writer_guard, &mem);
                let _ = engine.commit(&mut writer_guard);
            }
        }

        // Remove from cold archive
        self.delete_cold_record(memory_id)?;

        Ok(mem)
    }

    /// Find a memory anywhere: in the active hot galaxies, or decompressed from cold storage.
    ///
    /// Returns `(galaxy, memory, is_cold)`.
    pub fn find_anywhere(&self, id: MemoryId) -> Result<Option<(Galaxy, Memory, bool)>> {
        if let Some((galaxy, mem)) = self.find_across_galaxies(id)? {
            return Ok(Some((galaxy, mem, false)));
        }
        if let Some(cold_record) = self.get_cold_record(id)? {
            let galaxy = cold_record.galaxy;
            let mem = cold_record.decompress()?;
            return Ok(Some((galaxy, mem, true)));
        }
        Ok(None)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::content_hash;

    #[test]
    fn open_and_create_galaxies() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();
        for galaxy in Galaxy::all() {
            let _db = store.galaxy_db(galaxy).unwrap();
        }
    }

    #[test]
    fn ensure_schema_completes_a_pre_cold_store() {
        use lmdb::{DatabaseFlags as LmdbFlags, Environment as LmdbEnv};
        use uuid::Uuid;

        // Synthetic pre-cold store: every DBI a 9.0.0 store had, but no
        // `cold_storage` (the exact 2026-09-14 restore-drill finding).
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("old-store");
        std::fs::create_dir_all(&path).unwrap();
        {
            let env = LmdbEnv::new().set_max_dbs(64).open(&path).unwrap();
            for galaxy in Galaxy::all() {
                env.create_db(Some(galaxy.db_name()), LmdbFlags::default())
                    .unwrap();
            }
            for (name, flags) in crate::indexes::INDEX_DBS {
                env.create_db(Some(name), *flags).unwrap();
            }
            for (name, flags) in [
                ("episodic_records", LmdbFlags::default()),
                ("episodic_terms_v2", LmdbFlags::DUP_SORT),
                ("embedding_cache", LmdbFlags::default()),
                ("revisions", LmdbFlags::default()),
                (crate::attestation::ATTESTATIONS_DB, LmdbFlags::default()),
            ] {
                env.create_db(Some(name), flags).unwrap();
            }
        }

        let files_before: Vec<String> = {
            let mut names: Vec<String> = std::fs::read_dir(&path)
                .unwrap()
                .map(|e| e.unwrap().file_name().to_string_lossy().to_string())
                .collect();
            names.sort();
            names
        };
        let data_before = std::fs::read(path.join("data.mdb")).unwrap();

        let error = match MemoryStore::open_readonly(&path) {
            Ok(_) => panic!("strict read-only open must refuse an incomplete store"),
            Err(e) => e.to_string(),
        };
        assert!(error.contains("cold_storage"), "{error}");
        assert_eq!(
            std::fs::read(path.join("data.mdb")).unwrap(),
            data_before,
            "readonly refusal must not mutate the pre-cold store"
        );
        let files_after: Vec<String> = {
            let mut names: Vec<String> = std::fs::read_dir(&path)
                .unwrap()
                .map(|e| e.unwrap().file_name().to_string_lossy().to_string())
                .collect();
            names.sort();
            names
        };
        assert_eq!(
            files_after, files_before,
            "readonly refusal changed the store directory"
        );

        let created = MemoryStore::ensure_schema(&path).unwrap();
        assert_eq!(created, vec!["cold_storage".to_string()], "{created:?}");

        let store = MemoryStore::open_readonly(&path).unwrap();
        assert!(store.get_cold_record(Uuid::nil()).unwrap().is_none());

        // Idempotent: a complete store reports nothing missing.
        assert!(MemoryStore::ensure_schema(&path).unwrap().is_empty());
    }

    #[test]
    fn cold_discovery_hydrates_verifies_and_respects_visibility() {
        use crate::cold_storage::{ColdRecord, CompressionCodec, OuterRimFactors};
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();
        let factors = OuterRimFactors {
            age_factor: 0.5,
            access_factor: 0.5,
            resonance_factor: 0.5,
            emotional_factor: 0.5,
            importance_factor: 0.5,
            distance: 0.5,
        };

        let pub_mem = Memory::new(
            Galaxy::Codex,
            "public needle zxquniquecoldfact741 buried".into(),
        );
        let rec = ColdRecord::new(
            &pub_mem,
            0.5,
            factors.clone(),
            None,
            None,
            CompressionCodec::Gzip,
        )
        .unwrap();
        store.put_cold_record(&rec).unwrap();
        let out = store
            .find_cold_matching(&["zxquniquecoldfact741".to_string()], None, 10, 100)
            .unwrap();
        assert_eq!(out.matched, 1);
        assert_eq!(out.integrity_rejected, 0);
        assert_eq!(out.records.len(), 1);

        // Private originals never surface over MCP discovery.
        let mut priv_mem = Memory::new(Galaxy::Codex, "private needle zxquniquecoldfact742".into());
        priv_mem.metadata.is_private = true;
        store
            .put_cold_record(
                &ColdRecord::new(&priv_mem, 0.5, factors, None, None, CompressionCodec::Gzip)
                    .unwrap(),
            )
            .unwrap();
        let out_priv = store
            .find_cold_matching(&["zxquniquecoldfact742".to_string()], None, 10, 100)
            .unwrap();
        assert_eq!(out_priv.matched, 0);
        assert_eq!(out_priv.private_skipped, 1);

        // Tamper: the payload decompresses to content that no longer matches
        // the advertised content hash — refuse, never return.
        let mut tampered = rec.clone();
        let mut bad = Memory::new(Galaxy::Codex, "tampered needle zxquniquecoldfact743".into());
        bad.metadata.id = rec.id;
        let (payload, size) =
            crate::cold_storage::compress_memory(&bad, CompressionCodec::Gzip).unwrap();
        tampered.compressed_payload = payload;
        tampered.uncompressed_size = size;
        store.put_cold_record(&tampered).unwrap();
        let out_tamper = store
            .find_cold_matching(&["zxquniquecoldfact743".to_string()], None, 10, 100)
            .unwrap();
        assert_eq!(out_tamper.matched, 0);
        assert_eq!(out_tamper.integrity_rejected, 1);

        // Even an internally consistent header/payload must not be accepted
        // under another physical UUID key (which would break known-ID read).
        store.delete_cold_record(rec.id).unwrap();
        let wrong_key = uuid::Uuid::from_u128(741);
        assert_ne!(wrong_key, rec.id);
        let value = rmp_serde::to_vec_named(&rec).unwrap();
        let mut tx = store.env.begin_rw_txn().unwrap();
        tx.put(
            store.cold_storage_db,
            wrong_key.as_bytes(),
            &value,
            WriteFlags::default(),
        )
        .unwrap();
        tx.commit().unwrap();
        let wrong_key_out = store
            .find_cold_matching(&["zxquniquecoldfact741".into()], None, 10, 100)
            .unwrap();
        assert!(wrong_key_out.records.is_empty());
        assert_eq!(wrong_key_out.integrity_rejected, 1);
    }

    /// Substring filter (memory.query trap fix, 2026-08-29): literal
    /// case-insensitive content match, galaxy-wide — never an arbitrary
    /// page, never routed through the indexed fast paths.
    #[test]
    fn query_substring_filters_galaxy_wide() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        for (i, content) in [
            "the mesh joins at dawn",
            "unrelated content entirely",
            "MESH joins at dusk",
        ]
        .iter()
        .enumerate()
        {
            let mut m = Memory::new(Galaxy::Codex, content.to_string());
            m.metadata.importance = 0.5 + i as f32 / 10.0;
            store.put(Galaxy::Codex, &m).unwrap();
        }

        let hits = store
            .query(
                Galaxy::Codex,
                &MemoryQuery::new().with_content_substring("mesh joins"),
            )
            .unwrap();
        assert_eq!(hits.len(), 2, "CI substring must match both: {hits:?}");
        assert!(
            hits.iter()
                .all(|m| m.content.to_lowercase().contains("mesh joins"))
        );

        let none = store
            .query(
                Galaxy::Codex,
                &MemoryQuery::new().with_content_substring("quantum calendar"),
            )
            .unwrap();
        assert!(none.is_empty(), "no match must be an honest empty set");

        // Substring + tag combined still applies (no fast-path bypass).
        let mut tagged = Memory::new(Galaxy::Codex, "mesh joins again".to_string());
        tagged.metadata.tags = vec!["mesh".into()];
        store.put(Galaxy::Codex, &tagged).unwrap();
        let combined = store
            .query(
                Galaxy::Codex,
                &MemoryQuery::new()
                    .with_tags(vec!["mesh".into()])
                    .with_content_substring("again"),
            )
            .unwrap();
        assert_eq!(combined.len(), 1);
        assert_eq!(combined[0].content, "mesh joins again");
    }

    #[cfg(unix)]
    #[test]
    fn store_dir_has_restrictive_permissions() {
        let tmp = tempfile::tempdir().unwrap();
        let store_path = tmp.path().join("lmdb");
        let _store = MemoryStore::open_default(&store_path).unwrap();
        let perms = std::fs::metadata(&store_path).unwrap().permissions().mode();
        assert_eq!(
            perms & 0o777,
            0o700,
            "store directory should have 0700 permissions, got {:o}",
            perms & 0o777
        );
    }

    #[test]
    fn put_get_delete_memory() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let mem = Memory::new(Galaxy::Codex, "Hello world".to_string());
        let id = mem.metadata.id;

        store.put(Galaxy::Codex, &mem).unwrap();
        let retrieved = store.get(Galaxy::Codex, id).unwrap();
        assert!(retrieved.is_some());
        assert_eq!(retrieved.unwrap().content, "Hello world");

        let deleted = store.delete(Galaxy::Codex, id).unwrap();
        assert!(deleted);

        let gone = store.get(Galaxy::Codex, id).unwrap();
        assert!(gone.is_none());
    }

    #[test]
    fn scan_memories() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        for i in 0..5 {
            let mem = Memory::new(Galaxy::Codex, format!("memory-{i}"));
            store.put(Galaxy::Codex, &mem).unwrap();
        }

        let all = store.scan(Galaxy::Codex, 100).unwrap();
        assert_eq!(all.len(), 5);

        let limited = store.scan(Galaxy::Codex, 3).unwrap();
        assert_eq!(limited.len(), 3);
    }

    #[test]
    fn overwrite_removes_stale_index_entries() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        // Original record: tag "alpha", importance 0.9.
        let mut mem = Memory::new(Galaxy::Codex, "overwrite target".to_string());
        mem.metadata.tags = vec!["alpha".to_string()];
        mem.metadata.importance = 0.9;
        let id = mem.metadata.id;
        store.put(Galaxy::Codex, &mem).unwrap();

        // Overwrite with tag "beta", importance 0.1, new content hash.
        let mut updated = Memory::new(Galaxy::Codex, "overwritten content".to_string());
        updated.metadata.id = id;
        updated.metadata.tags = vec!["beta".to_string()];
        updated.metadata.importance = 0.1;
        store.put(Galaxy::Codex, &updated).unwrap();

        // Stale entries must be gone, new entries must be queryable.
        let tx = store.env().begin_ro_txn().unwrap();
        let by_alpha = store
            .index_dbs()
            .find_by_tag(&tx, Galaxy::Codex, "alpha")
            .unwrap();
        let by_beta = store
            .index_dbs()
            .find_by_tag(&tx, Galaxy::Codex, "beta")
            .unwrap();
        assert!(
            by_alpha.is_empty(),
            "stale tag index entries must be removed on overwrite"
        );
        assert_eq!(by_beta, vec![id]);

        let by_importance = store
            .index_dbs()
            .find_by_importance_range(&tx, Galaxy::Codex, 0.0, 0.2)
            .unwrap();
        assert!(
            by_importance.contains(&id),
            "new importance must be indexed"
        );
        let by_high = store
            .index_dbs()
            .find_by_importance_range(&tx, Galaxy::Codex, 0.8, 1.0)
            .unwrap();
        assert!(
            !by_high.contains(&id),
            "stale importance index entries must be removed on overwrite"
        );

        let old_hash = content_hash("overwrite target");
        let new_hash = content_hash("overwritten content");
        assert_eq!(
            store
                .index_dbs()
                .find_by_content_hash(&tx, Galaxy::Codex, &old_hash)
                .unwrap(),
            None,
            "stale content-hash index entry must be removed"
        );
        assert_eq!(
            store
                .index_dbs()
                .find_by_content_hash(&tx, Galaxy::Codex, &new_hash)
                .unwrap(),
            Some(id)
        );
    }

    #[test]
    fn count_memories() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        assert_eq!(store.count(Galaxy::Codex).unwrap(), 0);

        for i in 0..3 {
            let mem = Memory::new(Galaxy::Codex, format!("count-{i}"));
            store.put(Galaxy::Codex, &mem).unwrap();
        }

        assert_eq!(store.count(Galaxy::Codex).unwrap(), 3);
    }

    /// `count_by_tag` uses the tag index and counts distinct records — the
    /// status surface reports logical sessions with it (one `start` tag per
    /// session, ignoring turns/checkpoints).
    #[test]
    fn count_by_tag_counts_indexed_records() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        assert_eq!(store.count_by_tag(Galaxy::Sessions, "start").unwrap(), 0);

        let mut start = Memory::new(Galaxy::Sessions, "{\"type\":\"session_start\"}".into());
        start.metadata.tags = vec!["session".into(), "start".into()];
        store.put(Galaxy::Sessions, &start).unwrap();
        for i in 0..2 {
            let mut turn =
                Memory::new(Galaxy::Sessions, format!("{{\"type\":\"turn\",\"i\":{i}}}"));
            turn.metadata.tags = vec!["session".into(), "turn".into()];
            store.put(Galaxy::Sessions, &turn).unwrap();
        }

        assert_eq!(store.count(Galaxy::Sessions).unwrap(), 3);
        assert_eq!(store.count_by_tag(Galaxy::Sessions, "start").unwrap(), 1);
        assert_eq!(store.count_by_tag(Galaxy::Sessions, "turn").unwrap(), 2);
        assert_eq!(store.count_by_tag(Galaxy::Sessions, "absent").unwrap(), 0);
    }

    #[test]
    fn get_nonexistent_returns_none() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();
        let result = store.get(Galaxy::Codex, uuid::Uuid::new_v4()).unwrap();
        assert!(result.is_none());
    }

    #[test]
    fn raw_put_get() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        store
            .put_raw(Galaxy::Substrate, b"config:key", b"value123")
            .unwrap();
        let val = store.get_raw(Galaxy::Substrate, b"config:key").unwrap();
        assert_eq!(val, Some(b"value123".to_vec()));
    }

    #[test]
    fn put_dedup_prevents_duplicates() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let mem1 = Memory::new(Galaxy::Codex, "duplicate content".into());
        let id1 = store.put_dedup(Galaxy::Codex, &mem1).unwrap();

        let mem2 = Memory::new(Galaxy::Codex, "duplicate content".into());
        let id2 = store.put_dedup(Galaxy::Codex, &mem2).unwrap();

        assert_eq!(id1, id2, "dedup should return same ID for same content");
        assert_eq!(store.count(Galaxy::Codex).unwrap(), 1);
    }

    #[test]
    fn put_dedup_allows_different_content() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let mem1 = Memory::new(Galaxy::Codex, "content A".into());
        store.put_dedup(Galaxy::Codex, &mem1).unwrap();

        let mem2 = Memory::new(Galaxy::Codex, "content B".into());
        store.put_dedup(Galaxy::Codex, &mem2).unwrap();

        assert_eq!(store.count(Galaxy::Codex).unwrap(), 2);
    }

    #[test]
    fn put_batch_atomic_write() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let memories: Vec<Memory> = (0..10)
            .map(|i| Memory::new(Galaxy::Codex, format!("batch-{i}")))
            .collect();

        store.put_batch(Galaxy::Codex, &memories).unwrap();
        assert_eq!(store.count(Galaxy::Codex).unwrap(), 10);
    }

    #[test]
    fn query_by_tags() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let mem1 = Memory::new(Galaxy::Codex, "tagged memory".into())
            .with_tags(vec!["rust".into(), "memory".into()]);
        let mem2 =
            Memory::new(Galaxy::Codex, "other memory".into()).with_tags(vec!["python".into()]);
        store.put(Galaxy::Codex, &mem1).unwrap();
        store.put(Galaxy::Codex, &mem2).unwrap();

        let query = MemoryQuery::new().with_tags(vec!["rust".into()]);
        let results = store.query(Galaxy::Codex, &query).unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].content, "tagged memory");
    }

    #[test]
    fn query_by_importance_range() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        store
            .put(
                Galaxy::Codex,
                &Memory::new(Galaxy::Codex, "low".into()).with_importance(0.1),
            )
            .unwrap();
        store
            .put(
                Galaxy::Codex,
                &Memory::new(Galaxy::Codex, "mid".into()).with_importance(0.5),
            )
            .unwrap();
        store
            .put(
                Galaxy::Codex,
                &Memory::new(Galaxy::Codex, "high".into()).with_importance(0.9),
            )
            .unwrap();

        let query = MemoryQuery::new().with_importance_range(0.4, 0.6);
        let results = store.query(Galaxy::Codex, &query).unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].content, "mid");
    }

    #[test]
    fn memory_query_one_sided_time_bounds() {
        // `created_after` / `created_before` map onto the temporal filter
        // one side at a time (the API passthrough for memory.query).
        let old = Memory::new(Galaxy::Codex, "old".into());
        let mut recent = Memory::new(Galaxy::Codex, "recent".into());
        recent.metadata.created_at = old.metadata.created_at + chrono::Duration::days(30);

        let cutoff = old.metadata.created_at + chrono::Duration::days(10);
        let after = MemoryQuery::new().with_created_after(cutoff);
        assert!(!after.matches(&old), "pre-cutoff memory must not match");
        assert!(after.matches(&recent), "post-cutoff memory must match");

        let before = MemoryQuery::new().with_created_before(cutoff);
        assert!(before.matches(&old), "pre-cutoff memory must match");
        assert!(
            !before.matches(&recent),
            "post-cutoff memory must not match"
        );

        // Bounds are inclusive.
        let edge = MemoryQuery::new().with_created_after(cutoff);
        let mut at = Memory::new(Galaxy::Codex, "at cutoff".into());
        at.metadata.created_at = cutoff;
        assert!(edge.matches(&at), "created_at == after bound is inclusive");
    }

    #[test]
    fn embedding_put_get_delete() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let id = uuid::Uuid::new_v4();
        let embedding = vec![0.1, 0.2, 0.3, 0.4, 0.5];

        store.put_embedding(id, &embedding).unwrap();
        let retrieved = store.get_embedding(id).unwrap();
        assert!(retrieved.is_some());
        let retrieved = retrieved.unwrap();
        assert_eq!(retrieved.len(), 5);
        assert!((retrieved[0] - 0.1).abs() < f32::EPSILON);

        assert!(store.delete_embedding(id).unwrap());
        assert!(store.get_embedding(id).unwrap().is_none());
    }

    #[test]
    fn embedding_cache_roundtrip_batch_and_count() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let entries: Vec<(String, Vec<f32>)> = (0..5)
            .map(|i| (format!("ns:model:{i:016x}"), vec![i as f32; 8]))
            .collect();
        store.put_embedding_cache_batch(&entries).unwrap();
        assert_eq!(store.embedding_cache_count().unwrap(), 5);

        // Single read
        let hit = store
            .get_embedding_cache("ns:model:0000000000000003")
            .unwrap();
        assert_eq!(hit.unwrap(), vec![3.0f32; 8]);
        assert!(
            store
                .get_embedding_cache("ns:model:absent")
                .unwrap()
                .is_none()
        );

        // Batched read aligns 1:1, misses are None
        let keys: Vec<String> = (0..6).map(|i| format!("ns:model:{i:016x}")).collect();
        let batch = store.get_embedding_cache_batch(&keys).unwrap();
        assert_eq!(batch.len(), 6);
        assert!(batch[0..5].iter().all(Option::is_some));
        assert!(batch[5].is_none());

        // Overwrite is a put, not a duplicate
        store
            .put_embedding_cache("ns:model:0000000000000001", &[9.0; 8])
            .unwrap();
        assert_eq!(store.embedding_cache_count().unwrap(), 5);
        assert_eq!(
            store
                .get_embedding_cache("ns:model:0000000000000001")
                .unwrap()
                .unwrap(),
            vec![9.0f32; 8]
        );
    }

    #[test]
    fn embedding_cache_survives_store_reopen() {
        // V8 ship list #2 acceptance shape: vectors persist across restart.
        let tmp = tempfile::tempdir().unwrap();
        {
            let store = MemoryStore::open_default(tmp.path()).unwrap();
            store
                .put_embedding_cache("onnx:bge-small:abc", &[0.5; 384])
                .unwrap();
        }
        let reopened = MemoryStore::open_default(tmp.path()).unwrap();
        let cached = reopened.get_embedding_cache("onnx:bge-small:abc").unwrap();
        assert_eq!(cached.unwrap(), vec![0.5f32; 384]);
    }

    #[test]
    fn content_hash_is_sha256() {
        let hash1 = content_hash("test content");
        let hash2 = content_hash("test content");
        let hash3 = content_hash("different content");

        assert_eq!(hash1, hash2, "same content should produce same hash");
        assert_ne!(
            hash1, hash3,
            "different content should produce different hash"
        );
        assert_eq!(hash1.len(), 64, "SHA-256 hex should be 64 chars");
    }

    #[test]
    fn query_by_tag_uses_index() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let mem1 = Memory::new(Galaxy::Codex, "tagged".into())
            .with_tags(vec!["rust".into(), "memory".into()]);
        let mem2 = Memory::new(Galaxy::Codex, "other".into()).with_tags(vec!["python".into()]);
        store.put(Galaxy::Codex, &mem1).unwrap();
        store.put(Galaxy::Codex, &mem2).unwrap();

        let query = MemoryQuery::new().with_tags(vec!["rust".into()]);
        let results = store.query(Galaxy::Codex, &query).unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].content, "tagged");
    }

    #[test]
    fn query_by_importance_uses_index() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        store
            .put(
                Galaxy::Codex,
                &Memory::new(Galaxy::Codex, "low".into()).with_importance(0.1),
            )
            .unwrap();
        store
            .put(
                Galaxy::Codex,
                &Memory::new(Galaxy::Codex, "mid".into()).with_importance(0.5),
            )
            .unwrap();
        store
            .put(
                Galaxy::Codex,
                &Memory::new(Galaxy::Codex, "high".into()).with_importance(0.9),
            )
            .unwrap();

        let query = MemoryQuery::new().with_importance_range(0.4, 0.6);
        let results = store.query(Galaxy::Codex, &query).unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].content, "mid");
    }

    #[test]
    fn query_by_time_uses_index() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let t0 = chrono::Utc::now();
        std::thread::sleep(std::time::Duration::from_millis(10));
        let mem = Memory::new(Galaxy::Codex, "timed".into());
        store.put(Galaxy::Codex, &mem).unwrap();
        std::thread::sleep(std::time::Duration::from_millis(10));
        let t2 = chrono::Utc::now();

        let query = MemoryQuery::new().with_time_range(t0, t2);
        let results = store.query(Galaxy::Codex, &query).unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].content, "timed");
    }

    #[test]
    fn delete_removes_index_entries() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let mem = Memory::new(Galaxy::Codex, "test".into())
            .with_tags(vec!["tag1".into()])
            .with_importance(0.7);
        let id = mem.metadata.id;
        let hash = mem.metadata.content_hash.clone();
        store.put(Galaxy::Codex, &mem).unwrap();

        // Verify index entries exist
        assert!(
            store
                .find_by_content_hash(Galaxy::Codex, &hash)
                .unwrap()
                .is_some()
        );

        // Delete
        store.delete(Galaxy::Codex, id).unwrap();

        // Verify index entries are gone
        assert!(
            store
                .find_by_content_hash(Galaxy::Codex, &hash)
                .unwrap()
                .is_none()
        );

        // Tag query should return 0
        let query = MemoryQuery::new().with_tags(vec!["tag1".into()]);
        let results = store.query(Galaxy::Codex, &query).unwrap();
        assert!(results.is_empty());
    }

    #[test]
    fn put_batch_updates_indexes() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let memories: Vec<Memory> = (0..5)
            .map(|i| {
                Memory::new(Galaxy::Codex, format!("batch-{i}"))
                    .with_tags(vec![format!("tag{i}")])
                    .with_importance(i as f32 * 0.2)
            })
            .collect();
        store.put_batch(Galaxy::Codex, &memories).unwrap();

        for i in 0..5 {
            let query = MemoryQuery::new().with_tags(vec![format!("tag{i}")]);
            let results = store.query(Galaxy::Codex, &query).unwrap();
            assert_eq!(results.len(), 1, "tag{i} should have 1 result");
        }
    }

    #[test]
    fn find_by_content_hash_indexed_matches_scan() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let mem = Memory::new(Galaxy::Codex, "dedup test".into());
        let id = mem.metadata.id;
        let hash = mem.metadata.content_hash.clone();
        store.put(Galaxy::Codex, &mem).unwrap();

        let indexed = store.find_by_content_hash(Galaxy::Codex, &hash).unwrap();
        let scanned = store
            .find_by_content_hash_scan(Galaxy::Codex, &hash)
            .unwrap();

        assert_eq!(indexed, scanned);
        assert_eq!(indexed, Some(id));
    }

    #[test]
    fn put_dedup_uses_index() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let mem1 = Memory::new(Galaxy::Codex, "duplicate content".into());
        let id1 = store.put_dedup(Galaxy::Codex, &mem1).unwrap();

        let mem2 = Memory::new(Galaxy::Codex, "duplicate content".into());
        let id2 = store.put_dedup(Galaxy::Codex, &mem2).unwrap();

        assert_eq!(id1, id2, "dedup should return same ID for same content");
        assert_eq!(store.count(Galaxy::Codex).unwrap(), 1);
    }

    #[test]
    fn put_semantic_updates_coord5d() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let mut mem = Memory::new(
            Galaxy::Codex,
            "The algorithm computes data using a systematic method".to_string(),
        );
        let original_coord = mem.metadata.coord5d.clone();
        store.put_semantic(Galaxy::Codex, &mut mem).unwrap();

        // coord5d should have changed from the SHA-256 hash-based encoding
        assert_ne!(
            mem.metadata.coord5d.x, original_coord.x,
            "semantic encoding should change x"
        );
        assert_ne!(
            mem.metadata.coord5d.y, original_coord.y,
            "semantic encoding should change y"
        );

        // Verify it was stored with the semantic coordinate
        let retrieved = store.get(Galaxy::Codex, mem.metadata.id).unwrap().unwrap();
        assert_eq!(retrieved.metadata.coord5d.x, mem.metadata.coord5d.x);
    }

    #[test]
    fn put_semantic_preserves_temporal_and_importance() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let mut mem = Memory::new(Galaxy::Codex, "test content".into()).with_importance(0.8);
        mem.metadata.coord5d.w = 0.6;
        store.put_semantic(Galaxy::Codex, &mut mem).unwrap();

        assert!((mem.metadata.coord5d.w - 0.6).abs() < f32::EPSILON);
        assert!((mem.metadata.coord5d.v - 0.8).abs() < f32::EPSILON);
    }

    #[test]
    fn find_similar_returns_nearest_first() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let mut logic_mem = Memory::new(
            Galaxy::Codex,
            "The algorithm computes data using systematic logic and analysis".to_string(),
        );
        store.put_semantic(Galaxy::Codex, &mut logic_mem).unwrap();

        let mut emotion_mem = Memory::new(
            Galaxy::Codex,
            "I feel love and joy with deep passion and empathy in my heart".to_string(),
        );
        store.put_semantic(Galaxy::Codex, &mut emotion_mem).unwrap();

        // Query with logic-like text should find the logic memory first
        let results = store
            .find_similar(Galaxy::Codex, "algorithm data systematic method", 10)
            .unwrap();
        assert!(!results.is_empty());
        assert_eq!(results[0].0.metadata.id, logic_mem.metadata.id);

        // Query with emotion-like text should find the emotion memory first
        let results = store
            .find_similar(Galaxy::Codex, "love joy passion heart feeling", 10)
            .unwrap();
        assert!(!results.is_empty());
        assert_eq!(results[0].0.metadata.id, emotion_mem.metadata.id);
    }

    #[test]
    fn find_similar_empty_galaxy() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let results = store.find_similar(Galaxy::Codex, "anything", 10).unwrap();
        assert!(results.is_empty());
    }

    #[test]
    fn find_similar_respects_limit() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        for i in 0..5 {
            let mut mem = Memory::new(Galaxy::Codex, format!("algorithm data method {i}"));
            store.put_semantic(Galaxy::Codex, &mut mem).unwrap();
        }

        let results = store
            .find_similar(Galaxy::Codex, "algorithm data", 3)
            .unwrap();
        assert_eq!(results.len(), 3);
    }

    #[test]
    fn semantic_encoder_accessible() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let scores = store.semantic_encoder().encode("algorithm data logic");
        // Logic-heavy text → x < 0.5
        assert!(scores.x < 0.5);
    }

    #[test]
    fn put_raw_batch_writes_atomically() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let entries: &[(&[u8], &[u8])] =
            &[(b"key1", b"val1"), (b"key2", b"val2"), (b"key3", b"val3")];
        store.put_raw_batch(Galaxy::Karma, entries).unwrap();

        assert_eq!(
            store.get_raw(Galaxy::Karma, b"key1").unwrap().unwrap(),
            b"val1"
        );
        assert_eq!(
            store.get_raw(Galaxy::Karma, b"key2").unwrap().unwrap(),
            b"val2"
        );
        assert_eq!(
            store.get_raw(Galaxy::Karma, b"key3").unwrap().unwrap(),
            b"val3"
        );
    }

    #[test]
    fn put_raw_batch_empty_is_noop() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        store.put_raw_batch(Galaxy::Karma, &[]).unwrap();
        assert_eq!(store.count(Galaxy::Karma).unwrap(), 0);
    }

    #[test]
    fn entry_limit_rejects_excess_writes() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path())
            .unwrap()
            .with_entry_limit(3);

        for i in 0..3 {
            let mem = Memory::new(Galaxy::Codex, format!("memory {i}"));
            store.put(Galaxy::Codex, &mem).unwrap();
        }

        // 4th write should be rejected
        let mem = Memory::new(Galaxy::Codex, "overflow memory".to_string());
        let result = store.put(Galaxy::Codex, &mem);
        assert!(result.is_err(), "write beyond limit should be rejected");
        let err_msg = result.unwrap_err().to_string();
        assert!(
            err_msg.contains("entry limit reached"),
            "error should mention entry limit: {err_msg}"
        );
        assert_eq!(store.count(Galaxy::Codex).unwrap(), 3);
    }

    #[test]
    fn entry_limit_per_galaxy_independent() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path())
            .unwrap()
            .with_entry_limit(2);

        // Fill Codex to limit
        for i in 0..2 {
            let mem = Memory::new(Galaxy::Codex, format!("codex {i}"));
            store.put(Galaxy::Codex, &mem).unwrap();
        }

        // Writing to a different galaxy should still work
        let mem = Memory::new(Galaxy::Research, "science memory".to_string());
        let result = store.put(Galaxy::Research, &mem);
        assert!(
            result.is_ok(),
            "different galaxy should not be affected by limit"
        );
    }

    #[test]
    fn entry_limit_none_allows_unlimited() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        // No limit set — should allow many writes
        for i in 0..50 {
            let mem = Memory::new(Galaxy::Codex, format!("memory {i}"));
            store.put(Galaxy::Codex, &mem).unwrap();
        }
        assert_eq!(store.count(Galaxy::Codex).unwrap(), 50);
    }

    #[test]
    fn map_full_error_is_graceful() {
        // Small map: opening succeeds (galaxy + sidecar databases fit), but
        // the padded write loop fills it and MapFull surfaces from put().
        // 64KB proved too tight for eager create_db on macOS/arm64.
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open(tmp.path(), 512 * 1024).unwrap();

        // Write memories until map is full
        let mut written = 0;
        let mut got_map_full = false;
        for i in 0..1000 {
            let mem = Memory::new(
                Galaxy::Codex,
                format!("memory content {i} {}", "with padding ".repeat(50)),
            );
            match store.put(Galaxy::Codex, &mem) {
                Ok(()) => written += 1,
                Err(e) => {
                    let msg = e.to_string();
                    if msg.contains("map full") {
                        got_map_full = true;
                        break;
                    }
                    // Other errors are fine too (e.g., serialize failed)
                    break;
                }
            }
        }

        assert!(
            got_map_full || written < 1000,
            "should eventually hit map full or error"
        );
        assert!(written > 0, "should have written at least some memories");
    }

    #[test]
    fn test_find_across_galaxies() {
        let tmp = tempfile::tempdir().unwrap();
        let store = MemoryStore::open_default(tmp.path()).unwrap();

        let mem = Memory::new(Galaxy::Research, "Cross-galaxy research memo".into());
        let id = mem.metadata.id;
        store.put(Galaxy::Research, &mem).unwrap();

        let found = store.find_across_galaxies(id).unwrap();
        assert!(found.is_some());
        let (galaxy, retrieved) = found.unwrap();
        assert_eq!(galaxy, Galaxy::Research);
        assert_eq!(retrieved.metadata.id, id);
        assert_eq!(retrieved.content, "Cross-galaxy research memo");

        // Non-existent id returns None
        assert!(
            store
                .find_across_galaxies(uuid::Uuid::new_v4())
                .unwrap()
                .is_none()
        );
    }
}