zeph_memory/semantic/

cross_session.rs

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

use zeph_llm::provider::LlmProvider as _;

use crate::error::MemoryError;
use crate::types::ConversationId;
use crate::vector_store::{FieldCondition, FieldValue, VectorFilter};

use super::{SESSION_SUMMARIES_COLLECTION, SemanticMemory};

#[derive(Debug, Clone)]
pub struct SessionSummaryResult {
    pub summary_text: String,
    pub score: f32,
    pub conversation_id: ConversationId,
}

impl SemanticMemory {
    /// Check whether a session summary already exists for the given conversation.
    ///
    /// Returns `true` if at least one session summary is stored in `SQLite` for this conversation.
    /// Used as the primary guard in the shutdown summary path to handle cases where hard
    /// compaction fired but its Qdrant write failed (the `SQLite` record is the authoritative source).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    pub async fn has_session_summary(
        &self,
        conversation_id: ConversationId,
    ) -> Result<bool, MemoryError> {
        let summaries = self.sqlite.load_summaries(conversation_id).await?;
        Ok(!summaries.is_empty())
    }

    /// Store a shutdown session summary: persists to `SQLite`, embeds into the
    /// `zeph_session_summaries` Qdrant collection (so cross-session search can find it),
    /// and stores key facts into the key-facts collection.
    ///
    /// Unlike the hard-compaction path, `first_message_id` and `last_message_id` are `None`
    /// because the shutdown hook does not track exact message boundaries.
    ///
    /// # Errors
    ///
    /// Returns an error if the `SQLite` insert fails. Qdrant errors are logged as warnings
    /// and do not propagate — the `SQLite` record is the authoritative summary store.
    pub async fn store_shutdown_summary(
        &self,
        conversation_id: ConversationId,
        summary_text: &str,
        key_facts: &[String],
    ) -> Result<(), MemoryError> {
        let token_estimate =
            i64::try_from(self.token_counter.count_tokens(summary_text)).unwrap_or(0);
        // Persist to SQLite first — this is the authoritative record and the source of truth
        // for has_session_summary(). NULL message range = session-level summary.
        let summary_id = self
            .sqlite
            .save_summary(conversation_id, summary_text, None, None, token_estimate)
            .await?;

        // Embed into SESSION_SUMMARIES_COLLECTION so search_session_summaries() can find it.
        if let Err(e) = self
            .store_session_summary(conversation_id, summary_text)
            .await
        {
            tracing::warn!("shutdown summary: failed to embed into session summaries: {e:#}");
        }

        if !key_facts.is_empty() {
            self.store_key_facts(conversation_id, summary_id, key_facts)
                .await;
        }

        tracing::debug!(
            conversation_id = conversation_id.0,
            summary_id,
            "stored shutdown session summary"
        );
        Ok(())
    }

    /// Store a session summary into the dedicated `zeph_session_summaries` Qdrant collection.
    ///
    /// # Errors
    ///
    /// Returns an error if embedding or Qdrant storage fails.
    pub async fn store_session_summary(
        &self,
        conversation_id: ConversationId,
        summary_text: &str,
    ) -> Result<(), MemoryError> {
        let Some(qdrant) = &self.qdrant else {
            return Ok(());
        };
        if !self.effective_embed_provider().supports_embeddings() {
            return Ok(());
        }

        let vector = match tokio::time::timeout(
            self.embed_timeout,
            self.effective_embed_provider().embed(summary_text),
        )
        .await
        {
            Ok(Ok(v)) => v,
            Ok(Err(e)) => return Err(e.into()),
            Err(_) => {
                tracing::warn!("store_session_summary: embed timed out — skipping store");
                return Ok(());
            }
        };
        qdrant
            .ensure_named_collection_for_vector(SESSION_SUMMARIES_COLLECTION, &vector)
            .await?;

        let point_id = {
            const NS: uuid::Uuid = uuid::Uuid::NAMESPACE_OID;
            let name = format!("{}:{}", qdrant.db_instance_id(), conversation_id.0);
            uuid::Uuid::new_v5(&NS, name.as_bytes()).to_string()
        };
        let payload = serde_json::json!({
            "conversation_id": conversation_id.0,
            "db_instance_id": qdrant.db_instance_id(),
            "summary_text": summary_text,
        });

        qdrant
            .upsert_to_collection(SESSION_SUMMARIES_COLLECTION, &point_id, payload, vector)
            .await?;

        tracing::debug!(
            conversation_id = conversation_id.0,
            "stored session summary"
        );
        Ok(())
    }

    /// Search session summaries from other conversations.
    ///
    /// # Errors
    ///
    /// Returns an error if embedding or Qdrant search fails.
    #[cfg_attr(
        feature = "profiling",
        tracing::instrument(name = "memory.cross_session", skip_all, fields(result_count = tracing::field::Empty))
    )]
    pub async fn search_session_summaries(
        &self,
        query: &str,
        limit: usize,
        exclude_conversation_id: Option<ConversationId>,
    ) -> Result<Vec<SessionSummaryResult>, MemoryError> {
        let Some(qdrant) = &self.qdrant else {
            tracing::debug!("session-summaries: skipped, no vector store");
            return Ok(Vec::new());
        };
        if !self.effective_embed_provider().supports_embeddings() {
            tracing::debug!("session-summaries: skipped, no embedding support");
            return Ok(Vec::new());
        }

        let vector = match tokio::time::timeout(
            self.embed_timeout,
            self.effective_embed_provider().embed(query),
        )
        .await
        {
            Ok(Ok(v)) => v,
            Ok(Err(e)) => return Err(e.into()),
            Err(_) => {
                tracing::warn!(
                    "search_session_summaries: embed timed out, returning empty results"
                );
                return Ok(Vec::new());
            }
        };
        qdrant
            .ensure_named_collection_for_vector(SESSION_SUMMARIES_COLLECTION, &vector)
            .await?;

        // Always scope to this database (#5742): without a db_instance_id condition, this
        // search would otherwise span every conversation in every database sharing this
        // Qdrant instance whenever exclude_conversation_id is None (the common case).
        let filter = Some(VectorFilter {
            must: vec![FieldCondition {
                field: "db_instance_id".into(),
                value: FieldValue::Text(qdrant.db_instance_id().to_owned()),
            }],
            must_not: exclude_conversation_id
                .map(|cid| {
                    vec![FieldCondition {
                        field: "conversation_id".into(),
                        value: FieldValue::Integer(cid.0),
                    }]
                })
                .unwrap_or_default(),
        });

        let points = qdrant
            .search_collection(SESSION_SUMMARIES_COLLECTION, &vector, limit, filter)
            .await?;

        tracing::debug!(
            results = points.len(),
            limit,
            exclude_conversation_id = exclude_conversation_id.map(|c| c.0),
            "session-summaries: search complete"
        );

        let results = points
            .into_iter()
            .filter_map(|point| {
                let summary_text = point.payload.get("summary_text")?.as_str()?.to_owned();
                let conversation_id =
                    ConversationId(point.payload.get("conversation_id")?.as_i64()?);
                Some(SessionSummaryResult {
                    summary_text,
                    score: point.score,
                    conversation_id,
                })
            })
            .collect();

        Ok(results)
    }
}

#[cfg(test)]
mod tests {
    use zeph_llm::any::AnyProvider;
    use zeph_llm::mock::MockProvider;

    use crate::types::MessageId;

    use super::*;

    async fn make_memory() -> SemanticMemory {
        SemanticMemory::new(
            ":memory:",
            "http://127.0.0.1:1",
            None,
            AnyProvider::Mock(MockProvider::default()),
            "test-model",
        )
        .await
        .unwrap()
    }

    /// Insert a real message into the conversation and return its `MessageId`.
    /// Required because the `summaries` table has FK constraints on `messages.id`.
    async fn insert_message(memory: &SemanticMemory, cid: ConversationId) -> MessageId {
        memory
            .sqlite()
            .save_message(cid, "user", "test message")
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn has_session_summary_returns_false_when_no_summaries() {
        let memory = make_memory().await;
        let cid = memory.sqlite().create_conversation().await.unwrap();

        let result = memory.has_session_summary(cid).await.unwrap();
        assert!(!result, "new conversation must have no summaries");
    }

    #[tokio::test]
    async fn has_session_summary_returns_true_after_summary_stored_via_sqlite() {
        let memory = make_memory().await;
        let cid = memory.sqlite().create_conversation().await.unwrap();
        let msg_id = insert_message(&memory, cid).await;

        // Use sqlite directly to insert a valid summary with real FK references.
        memory
            .sqlite()
            .save_summary(
                cid,
                "session about Rust and async",
                Some(msg_id),
                Some(msg_id),
                10,
            )
            .await
            .unwrap();

        let result = memory.has_session_summary(cid).await.unwrap();
        assert!(result, "must return true after a summary is persisted");
    }

    #[tokio::test]
    async fn has_session_summary_is_isolated_per_conversation() {
        let memory = make_memory().await;
        let cid_a = memory.sqlite().create_conversation().await.unwrap();
        let cid_b = memory.sqlite().create_conversation().await.unwrap();
        let msg_id = insert_message(&memory, cid_a).await;

        memory
            .sqlite()
            .save_summary(
                cid_a,
                "summary for conversation A",
                Some(msg_id),
                Some(msg_id),
                5,
            )
            .await
            .unwrap();

        assert!(
            memory.has_session_summary(cid_a).await.unwrap(),
            "cid_a must have a summary"
        );
        assert!(
            !memory.has_session_summary(cid_b).await.unwrap(),
            "cid_b must not be affected by cid_a summary"
        );
    }

    /// Mirrors the point-id computation in `store_session_summary` so the assertion stays
    /// meaningful if the production format string changes shape.
    fn point_id_name(db_instance_id: &str, cid: ConversationId) -> String {
        format!("{db_instance_id}:{}", cid.0)
    }

    #[test]
    fn store_session_summary_point_id_is_deterministic() {
        // Same (db_instance_id, conversation_id) must always produce the same UUID v5 point
        // ID, ensuring that repeated compaction calls upsert rather than insert a new point.
        const NS: uuid::Uuid = uuid::Uuid::NAMESPACE_OID;
        let cid = ConversationId(42);
        let id1 = uuid::Uuid::new_v5(&NS, point_id_name("", cid).as_bytes()).to_string();
        let id2 = uuid::Uuid::new_v5(&NS, point_id_name("", cid).as_bytes()).to_string();
        assert_eq!(
            id1, id2,
            "point_id must be deterministic for the same inputs"
        );

        let cid2 = ConversationId(43);
        let id3 = uuid::Uuid::new_v5(&NS, point_id_name("", cid2).as_bytes()).to_string();
        assert_ne!(
            id1, id3,
            "different conversation_ids must produce different point_ids"
        );
    }

    #[test]
    fn store_session_summary_point_id_boundary_ids() {
        // conversation_id = 0 and negative values are valid i64 variants — confirm they produce
        // valid, distinct, and stable UUIDs.
        const NS: uuid::Uuid = uuid::Uuid::NAMESPACE_OID;

        let id_zero_a = uuid::Uuid::new_v5(&NS, point_id_name("", ConversationId(0)).as_bytes());
        let id_zero_b = uuid::Uuid::new_v5(&NS, point_id_name("", ConversationId(0)).as_bytes());
        assert_eq!(id_zero_a, id_zero_b, "zero conversation_id must be stable");

        let id_neg = uuid::Uuid::new_v5(&NS, point_id_name("", ConversationId(-1)).as_bytes());
        assert_ne!(
            id_zero_a, id_neg,
            "zero and -1 conversation_ids must produce different point_ids"
        );

        // Confirm the UUID version is 5 (deterministic SHA-1 name-based).
        assert_eq!(
            id_zero_a.get_version_num(),
            5,
            "generated UUID must be version 5"
        );
    }

    /// Regression test for the write-time clobber finding in #5742: two databases whose
    /// `conversation_id` sequences both reach the same value must not collide on the same
    /// Qdrant point id.
    #[test]
    fn store_session_summary_point_id_differs_across_db_instances() {
        const NS: uuid::Uuid = uuid::Uuid::NAMESPACE_OID;
        let cid = ConversationId(1);
        let id_db_a = uuid::Uuid::new_v5(&NS, point_id_name("db-a", cid).as_bytes());
        let id_db_b = uuid::Uuid::new_v5(&NS, point_id_name("db-b", cid).as_bytes());
        assert_ne!(
            id_db_a, id_db_b,
            "same conversation_id in two different databases must produce different point_ids"
        );
    }

    /// Builds a memory whose `EmbeddingStore` is tagged with `db_instance_id` and backed by
    /// `shared_vector_pool` instead of its own private pool — simulating a second physical
    /// database sharing one Qdrant instance with whatever other store(s) were also built against
    /// `shared_vector_pool` (#5742). The memory keeps its own private `SqliteStore`, so two
    /// memories built this way independently start at `conversation_id = 1`.
    async fn make_embed_memory_with_db_instance(
        db_instance_id: &str,
        shared_vector_pool: zeph_db::DbPool,
    ) -> SemanticMemory {
        let mut mock = MockProvider::default();
        mock.supports_embeddings = true;
        mock.embedding = vec![1.0, 0.0, 0.0, 0.0];
        let provider = AnyProvider::Mock(mock);

        let sqlite = crate::store::SqliteStore::new(":memory:").await.unwrap();
        let pool = sqlite.pool().clone();
        let store = crate::embedding_store::EmbeddingStore::with_store(
            Box::new(crate::db_vector_store::DbVectorStore::new(
                shared_vector_pool,
            )),
            pool,
        )
        .with_db_instance_id(db_instance_id);

        SemanticMemory::base(
            sqlite,
            Some(std::sync::Arc::new(store)),
            provider,
            "test-model".into(),
            0.7,
            0.3,
            std::sync::Arc::new(crate::token_counter::TokenCounter::new()),
        )
    }

    /// #5742 regression: two independent databases (own `conversation_id` counters, each starting
    /// at 1) share one backing vector store. `search_session_summaries`'s unscoped branch
    /// (`exclude_conversation_id = None`, the common cross-session-recall case) must not surface
    /// a summary written by a different database, even when their `conversation_id` values match.
    #[tokio::test]
    async fn search_session_summaries_scoped_excludes_other_database_same_conversation_id() {
        let shared = crate::store::SqliteStore::new(":memory:").await.unwrap();
        let shared_pool = shared.pool().clone();

        let memory_a = make_embed_memory_with_db_instance("db-a", shared_pool.clone()).await;
        let memory_b = make_embed_memory_with_db_instance("db-b", shared_pool.clone()).await;

        let cid_a = memory_a.sqlite().create_conversation().await.unwrap();
        let cid_b = memory_b.sqlite().create_conversation().await.unwrap();
        assert_eq!(
            cid_a, cid_b,
            "both databases must independently start at conversation_id=1"
        );

        memory_a
            .store_session_summary(cid_a, "summary from database A")
            .await
            .unwrap();
        memory_b
            .store_session_summary(cid_b, "summary from database B")
            .await
            .unwrap();

        let results_a = memory_a
            .search_session_summaries("summary", 10, None)
            .await
            .unwrap();
        assert_eq!(
            results_a.len(),
            1,
            "db-a's unscoped cross-session recall must not see db-b's summary"
        );
        assert_eq!(results_a[0].summary_text, "summary from database A");

        let results_b = memory_b
            .search_session_summaries("summary", 10, None)
            .await
            .unwrap();
        assert_eq!(
            results_b.len(),
            1,
            "db-b's unscoped cross-session recall must not see db-a's summary"
        );
        assert_eq!(results_b[0].summary_text, "summary from database B");
    }

    #[tokio::test]
    async fn store_shutdown_summary_succeeds_with_null_message_ids() {
        let memory = make_memory().await;
        let cid = memory.sqlite().create_conversation().await.unwrap();

        let result = memory
            .store_shutdown_summary(cid, "summary text", &[])
            .await;

        assert!(
            result.is_ok(),
            "shutdown summary must succeed without messages"
        );
        assert!(
            memory.has_session_summary(cid).await.unwrap(),
            "SQLite must record the shutdown summary"
        );
    }
}