Struct ConversationMemorySystem 

pub struct ConversationMemorySystem {

    pub session_manager: SessionManager,
    pub context_processor: IncrementalContextProcessor,
    pub graph_manager: SimpleGraphManager,
    pub workspace_manager: Arc<WorkspaceManager>,
    pub storage_actor: StorageActorHandle,
    pub vector_storage: Arc<dyn VectorStorage>,
    pub config: SystemConfig,
    pub performance_monitor: Arc<PerformanceMonitor>,
    pub circuit_breaker: Arc<CircuitBreaker>,
    pub system_metrics: Arc<SystemMetrics>,
    pub content_vectorizer: Arc<OnceCell<Arc<ContentVectorizer>>>,
    pub semantic_query_engine: Arc<OnceCell<Arc<SemanticQueryEngine>>>,
    pub embedding_config_holder: Arc<EmbeddingConfigHolder>,
}

Conversation memory system using actors and channels

Fields

session_manager: SessionManager

Session lifecycle manager with cache

context_processor: IncrementalContextProcessor

Incremental context processing metrics

graph_manager: SimpleGraphManager

Knowledge graph metric container

workspace_manager: Arc<WorkspaceManager>

Workspace management coordinator

storage_actor: StorageActorHandle

Handle to the async storage actor

vector_storage: Arc<dyn VectorStorage>

Vector storage backend for semantic search

config: SystemConfig

System configuration

performance_monitor: Arc<PerformanceMonitor>

Performance metrics monitor

circuit_breaker: Arc<CircuitBreaker>

Circuit breaker for fault tolerance

system_metrics: Arc<SystemMetrics>

Global system metrics

content_vectorizer: Arc<OnceCell<Arc<ContentVectorizer>>>

Lazily initialized content vectorizer

semantic_query_engine: Arc<OnceCell<Arc<SemanticQueryEngine>>>

Lazily initialized semantic query engine

embedding_config_holder: Arc<EmbeddingConfigHolder>

Configuration holder used for lazy embedding initialization

Implementations

impl ConversationMemorySystem

pub async fn spawn_background_vectorization( &self, session_id: Uuid, session_arc: Arc<ArcSwapAny<Arc<ActiveSession>>>, )

Vectorize the latest update in the background. Fire-and-forget; failures are logged. Returns immediately after the vectorizer has been initialised — on first call this still waits for the model download (~50 MB for potion-multilingual-128M). The new canonical write path goes through MemoryServiceImpl (its update_context impl), which hands the work to the bounded background crate::pipeline::Pipeline — the pipeline worker runs the same init inside its own task, so callers of update_context never see the model-load cost on the hot path. This legacy method stays as the safety net for direct callers of ConversationMemorySystem::add_incremental_update until they migrate (TODO.md item #4 follow-up, slated for 0.4.0).

pub async fn ensure_semantic_engine_initialized( &self, ) -> Result<Arc<SemanticQueryEngine>, String>

Lazy-initialize semantic query engine on first use

pub async fn vectorize_session(&self, session_id: Uuid) -> Result<usize, String>

Vectorize a session’s content (requires embeddings feature)

pub async fn auto_vectorize_if_enabled( &self, session_id: Uuid, ) -> Result<(), String>

Auto-vectorize only the latest update (incremental vectorization) This is much more efficient than re-vectorizing the entire session Includes retry mechanism for transient failures

pub async fn vectorize_all_sessions( &self, ) -> Result<(usize, usize, usize), String>

Vectorize all sessions in the system with parallel processing Returns total number of vectorized items across all sessions and statistics Uses a semaphore to limit concurrent vectorization tasks

pub async fn semantic_search_global( &self, query: &str, limit: Option<usize>, date_range: Option<(DateTime<Utc>, DateTime<Utc>)>, recency_bias: Option<f32>, ) -> Result<Vec<SemanticSearchResult>, String>

Perform semantic search across all sessions

pub async fn semantic_search_session( &self, session_id: Uuid, query: &str, limit: Option<usize>, date_range: Option<(DateTime<Utc>, DateTime<Utc>)>, recency_bias: Option<f32>, ) -> Result<Vec<SemanticSearchResult>, String>

Perform semantic search within a specific session

pub async fn find_related_content( &self, session_id: Uuid, topic: &str, limit: Option<usize>, ) -> Result<Vec<SemanticSearchResult>, String>

Find related content across sessions

pub async fn semantic_search_multisession( &self, session_ids: &[Uuid], query: &str, limit: Option<usize>, date_range: Option<(DateTime<Utc>, DateTime<Utc>)>, recency_bias: Option<f32>, ) -> Result<Vec<SemanticSearchResult>, String>

Perform semantic search across multiple sessions

pub fn get_vectorization_stats(&self) -> Result<HashMap<String, usize>, String>

Get vectorization statistics

pub fn embeddings_enabled(&self) -> bool

Check if embeddings are enabled and initialized

pub async fn enable_embeddings_config(&mut self) -> Result<(), String>

Enable embeddings at runtime (requires restart to initialize components)

pub async fn set_embedding_model( &mut self, model_type: String, ) -> Result<(), String>

Configure embedding model type

pub async fn invalidate_and_rebuild_entity_graph( &self, session_id: Uuid, file_path: &str, ) -> Result<(u32, usize), String>

Invalidate a source file and rebuild the entity graph for the given session.

1. Removes SourceReference entries for the file (storage layer)
2. Removes incremental updates referencing the file from the session
3. Rebuilds entity graph from remaining updates
4. Persists the updated session

Returns (entries_invalidated, entities_after_rebuild).

pub async fn clear_query_cache(&self) -> Result<(), String>

Clear query cache to prevent stale vector IDs after restart

This should be called on daemon startup to ensure cached query results don’t reference vector IDs from before the restart, which would cause incorrect similarity calculations.

impl ConversationMemorySystem

pub async fn new( config: SystemConfig, ) -> Result<ConversationMemorySystem, String>

Create system from config

Errors

Returns an error if storage initialization fails or if system setup encounters any issues

pub async fn new_with_storage( storage: RealRocksDBStorage, config: SystemConfig, ) -> Result<ConversationMemorySystem, String>

Create system with RocksDB storage (backward compatibility)

pub async fn create_session( &self, name: Option<String>, description: Option<String>, ) -> Result<Uuid, String>

Create a new session with name and description

pub async fn get_session( &self, session_id: Uuid, ) -> Result<Arc<ArcSwapAny<Arc<ActiveSession>>>, String>

Get existing session by ID

pub fn get_storage(&self) -> &StorageActorHandle

Get storage actor handle for compatibility (replaces storage.read().await)

pub async fn update_session_metadata( &self, session_id: Uuid, name: Option<String>, description: Option<String>, ) -> Result<(), String>

Update session metadata compatibility method

pub async fn delete_session(&self, session_id: Uuid) -> Result<bool, String>

Delete a session: evicts both in-memory caches before tearing down the persisted row. Without the cache evict step, get_session (used by session action=load) would still hand out the cached ActiveSession after the underlying storage row was already gone.

pub async fn delete_entity( &self, session_id: Uuid, entity_name: &str, ) -> Result<bool, String>

Delete a single entity from a session: removes the node + incident edges from the in-memory graph (so cached writes don’t resurrect it) and cascades the same delete to storage (entity row + all 8 edge tables). Returns whether the entity existed.

pub async fn delete_update( &self, session_id: Uuid, entry_id: Uuid, ) -> Result<bool, String>

Delete a single incremental update (context entry) by its entry_id.

Used by MCP manage_entity action=delete_update to clean up ghost/bad rows produced when a client posts content the dispatcher can’t map to a title. Persists the updated session via the storage actor. Returns true if the entry existed.

pub async fn find_sessions_by_name_or_description( &self, query: &str, ) -> Result<Vec<Uuid>, String>

Find sessions by name or description compatibility method

pub async fn list_sessions(&self) -> Result<Vec<Uuid>, String>

List all sessions

pub async fn apply_entity_graph_update_now( &self, session_id: Uuid, update: ContextUpdate, ) -> Result<(), String>

Apply an entity-graph update for a session now (no spawn).

Encapsulates the load → mutate → CAS-swap → persist pattern that add_incremental_update_internal previously inlined inside a tokio::spawn. Exposed publicly so the non-blocking crate::pipeline workers can drive entity-graph maintenance off the hot write path.

Idempotent w.r.t. concurrent updates: a failed CAS-swap is treated as “another writer won” and silently dropped (the next update catches up). Errors propagate so the caller — typically a pipeline worker — can log them and continue.

pub async fn vectorize_latest_update_now( &self, session_id: Uuid, ) -> Result<usize, String>

Vectorize the latest update for a session now (no spawn).

Equivalent body of Self::spawn_background_vectorization without the surrounding tokio::spawn — lets the crate::pipeline::EmbeddingQueue worker run vectorization from inside its own task while the write path stays in single-digit ms.

pub async fn add_incremental_update( &self, session_id: Uuid, description: String, metadata: Option<Value>, ) -> Result<String, String>

Add incremental update - compatibility wrapper

pub async fn get_conversation_context( &self, session_id: Uuid, ) -> Result<String, String>

Get conversation context

pub fn get_system_health(&self) -> SystemHealth

Get system health - all atomic reads

pub async fn cleanup_expired_sessions(&self)

Cleanup expired sessions - background task Also enforces memory limits on active_sessions DashMap

pub async fn force_memory_cleanup(&self, target_size: usize)

Force cleanup to reduce memory usage immediately Useful when the system is under memory pressure

pub async fn save_workspace_metadata( &self, workspace_id: Uuid, name: &str, description: &str, session_ids: &[Uuid], ) -> Result<(), String>

Save workspace metadata (proxy to storage actor)