zeph_memory/store/

mod.rs

// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
// SPDX-License-Identifier: MIT OR Apache-2.0

//! SQLite-backed relational store for all persistent agent data.
//!
//! [`SqliteStore`] wraps a [`zeph_db::DbPool`] and provides typed sub-store modules
//! for every data domain:
//!
//! | Module | Contents |
//! |--------|----------|
//! | `messages` | Conversation messages, role strings, metadata |
//! | `summaries` | Compression summaries per conversation |
//! | `persona` | Long-lived user attributes ([`PersonaFactRow`]) |
//! | `trajectory` | Goal trajectory entries ([`TrajectoryEntryRow`]) |
//! | `memory_tree` | Hierarchical note consolidation tree ([`MemoryTreeRow`]) |
//! | `session_digest` | Per-session digest records |
//! | `experiments` | A/B experiment results and session summaries |
//! | `corrections` | User-issued corrections stored for fine-tuning |
//! | `graph_store` | Entity/edge adjacency tables for the knowledge graph |
//! | `overflow` | Context-overflow handling metadata |
//! | `preferences` | User preference key-value store |
//! | `skills` | Skill metrics, usage, and version records |
//! | `trust` | Skill trust scores by source |
//! | `acp_sessions` | ACP protocol session events |
//! | `mem_scenes` | Scene segmentation records |
//! | `compression_guidelines` | LLM compression policy guidelines |
//! | `admission_training` | A-MAC admission training data |
//! | `channel_preferences` | Per-channel UX preferences (e.g. last active provider) |
//! | `agent_sessions` | Fleet session lifecycle records ([`AgentSessionRow`]) |
//! | `cross_thread` | Generic namespaced cross-thread key-value store ([`StoreItem`], spec-080) |
//! | `usage_records` | Per-LLM-call usage/cost/latency ledger ([`crate::UsageRecord`], issue #6549) |

mod acp_sessions;
pub mod admission_training;
pub mod agent_sessions;
pub mod channel_preferences;
pub mod compression_guidelines;
pub mod corrections;
pub mod cross_thread;
pub mod experiments;
pub mod graph_store;
mod history;
mod mem_scenes;
pub mod memory_tree;
pub(crate) mod messages;
pub mod overflow;
pub mod persona;
pub mod preferences;
pub mod retrieval_failures;
pub mod session_digest;
mod skills;
mod summaries;
pub mod trajectory;
mod trust;
pub mod usage_records;

#[allow(unused_imports)]
use zeph_db::sql;
use zeph_db::{DbConfig, DbPool};

use crate::error::MemoryError;

pub use acp_sessions::{AcpSessionConfigSnapshot, AcpSessionEvent, AcpSessionInfo};
pub use agent_sessions::{AgentSessionRow, SessionChannel, SessionKind, SessionStatus};
pub use cross_thread::StoreItem;
pub use memory_tree::MemoryTreeRow;
pub use messages::role_str;
pub use persona::PersonaFactRow;
pub use skills::{SkillMetricsRow, SkillUsageRow, SkillVersionRow};
pub use trajectory::{NewTrajectoryEntry, TrajectoryEntryRow};
pub use trust::{SkillTrustRow, SourceKind};

/// Primary relational data store backed by a [`DbPool`].
///
/// Opening a `SqliteStore` runs all pending `SQLite` migrations automatically.
///
/// # Examples
///
/// ```rust,no_run
/// # async fn example() -> Result<(), zeph_memory::MemoryError> {
/// use zeph_memory::store::SqliteStore;
///
/// let store = SqliteStore::new(":memory:").await?;
/// let cid = store.create_conversation().await?;
/// # Ok(())
/// # }
/// ```
#[derive(Debug, Clone)]
pub struct SqliteStore {
    pool: DbPool,
    db_instance_id: String,
}

impl SqliteStore {
    /// Open (or create) the database and run migrations.
    ///
    /// # Errors
    ///
    /// Returns an error if the database cannot be opened or migrations fail.
    pub async fn new(path: &str) -> Result<Self, MemoryError> {
        Self::with_pool_size(path, 5).await
    }

    /// Open (or create) the database with a configurable connection pool size.
    ///
    /// # Errors
    ///
    /// Returns an error if the database cannot be opened or migrations fail.
    pub async fn with_pool_size(path: &str, pool_size: u32) -> Result<Self, MemoryError> {
        let pool = DbConfig {
            url: path.to_string(),
            pool_size,
        }
        .connect()
        .await?;
        let db_instance_id = zeph_db::get_or_create_instance_id(&pool).await?;

        Ok(Self {
            pool,
            db_instance_id,
        })
    }

    /// Create a store from an already-open pool (no migrations run).
    ///
    /// Use this when the pool was obtained from an existing store (e.g. the main
    /// agent memory store) to avoid redundant migration runs.
    ///
    /// # Errors
    ///
    /// Returns an error if this database's stable [`Self::db_instance_id`] cannot be
    /// fetched or created (#5742).
    pub async fn from_pool(pool: DbPool) -> Result<Self, MemoryError> {
        let db_instance_id = zeph_db::get_or_create_instance_id(&pool).await?;
        Ok(Self {
            pool,
            db_instance_id,
        })
    }

    /// Expose the underlying pool for shared access by other stores.
    #[must_use]
    pub fn pool(&self) -> &DbPool {
        &self.pool
    }

    /// Stable UUID identifying this physical database (#5742).
    ///
    /// Generated once and persisted in the `db_instance` table. Used to disambiguate
    /// autoincrementing IDs (e.g. `conversation_id`) when multiple independent databases
    /// share one Qdrant instance.
    #[must_use]
    pub fn db_instance_id(&self) -> &str {
        &self.db_instance_id
    }

    /// Run all migrations on the given pool.
    ///
    /// # Errors
    ///
    /// Returns an error if any migration fails.
    pub async fn run_migrations(pool: &DbPool) -> Result<(), MemoryError> {
        zeph_db::run_migrations(pool).await?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::NamedTempFile;

    // Matches `DbConfig` busy_timeout default (5 seconds in ms)
    const DEFAULT_BUSY_TIMEOUT_MS: i64 = 5_000;

    #[tokio::test]
    async fn wal_journal_mode_enabled_on_file_db() {
        let file = NamedTempFile::new().expect("tempfile");
        let path = file.path().to_str().expect("valid path");

        let store = SqliteStore::new(path).await.expect("SqliteStore::new");

        let mode: String = zeph_db::query_scalar(sql!("PRAGMA journal_mode"))
            .fetch_one(store.pool())
            .await
            .expect("PRAGMA query");

        assert_eq!(mode, "wal", "expected WAL journal mode, got: {mode}");
    }

    #[tokio::test]
    async fn busy_timeout_enabled_on_file_db() {
        let file = NamedTempFile::new().expect("tempfile");
        let path = file.path().to_str().expect("valid path");

        let store = SqliteStore::new(path).await.expect("SqliteStore::new");

        let timeout_ms: i64 = zeph_db::query_scalar(sql!("PRAGMA busy_timeout"))
            .fetch_one(store.pool())
            .await
            .expect("PRAGMA query");

        assert_eq!(
            timeout_ms, DEFAULT_BUSY_TIMEOUT_MS,
            "expected busy_timeout pragma to match configured timeout"
        );
    }

    #[tokio::test]
    async fn creates_parent_dirs() {
        let dir = tempfile::tempdir().expect("tempdir");
        let deep = dir.path().join("a/b/c/zeph.db");
        let path = deep.to_str().expect("valid path");
        let _store = SqliteStore::new(path).await.expect("SqliteStore::new");
        assert!(deep.exists(), "database file should exist");
    }

    #[tokio::test]
    async fn with_pool_size_accepts_custom_size() {
        let store = SqliteStore::with_pool_size(":memory:", 2)
            .await
            .expect("with_pool_size");
        // Verify the store is operational with the custom pool size.
        let _cid = store
            .create_conversation()
            .await
            .expect("create_conversation");
    }
}