recall_echo/graph/
mod.rs

1//! recall-graph — Knowledge graph with semantic search for AI memory systems.
2//!
3//! Provides a structured graph layer (Layer 0) underneath flat-file memory systems.
4//! Used by recall-echo (pulse-null entities) and recall-claude (Claude Code users).
5
6pub mod confidence;
7pub mod crud;
8pub mod dedup;
9pub mod embed;
10pub mod error;
11pub mod extract;
12pub mod gc;
13pub mod ingest;
14pub mod llm;
15pub mod pipeline;
16pub mod pipeline_sync;
17pub mod query;
18pub mod search;
19pub mod store;
20pub mod traverse;
21pub mod types;
22pub mod util;
23pub mod utility;
24pub mod vigil_sync;
25
26use std::collections::HashMap;
27use std::path::{Path, PathBuf};
28
29pub use confidence::{Provenance, ProvenanceWeights};
30use embed::{FastEmbedder, LazyEmbedder};
31use error::GraphError;
32pub use ingest::{IngestContext, ProvenancePolicy};
33use store::Db;
34pub use store::ServerConfig;
35#[allow(unused_imports)] // Required in scope for SurrealValue derive macro expansion
36use surrealdb::types::SurrealValue;
37use surrealdb::Surreal;
38use types::*;
39
40/// Take serde_json::Value results from a SurrealDB response and deserialize to a Rust type.
41/// This avoids needing SurrealValue derive on complex types.
42pub(crate) fn deserialize_take<T: serde::de::DeserializeOwned>(
43    response: &mut surrealdb::IndexedResults,
44    index: usize,
45) -> Result<Vec<T>, GraphError> {
46    let values: Vec<serde_json::Value> = response.take(index)?;
47    values
48        .into_iter()
49        .map(|v| serde_json::from_value(v).map_err(GraphError::from))
50        .collect()
51}
52
53pub(crate) fn deserialize_take_opt<T: serde::de::DeserializeOwned>(
54    response: &mut surrealdb::IndexedResults,
55    index: usize,
56) -> Result<Option<T>, GraphError> {
57    let values: Vec<T> = deserialize_take(response, index)?;
58    Ok(values.into_iter().next())
59}
60
61/// The main entry point for graph memory operations.
62pub struct GraphMemory {
63    db: Surreal<Db>,
64    embedder: LazyEmbedder,
65    path: PathBuf,
66    scoring: crate::config::GraphScoringConfig,
67    provenance: confidence::ProvenanceWeights,
68}
69
70impl GraphMemory {
71    /// Open a graph store at the given path.
72    ///
73    /// The backend is chosen at runtime from the `[graph] mode` key of
74    /// `.recall-echo.toml` in the parent directory (memory_dir):
75    /// `embedded` (default) opens SurrealKV at `path/surreal/`; `server`
76    /// connects to a SurrealDB server via the configured URL.
77    /// The `path` is used for the FastEmbed models cache in both modes.
78    pub async fn open(path: &Path) -> Result<Self, GraphError> {
79        let memory_dir = path.parent().unwrap_or(path);
80        let config = crate::config::load_from_dir(memory_dir);
81        let mode = config
82            .graph
83            .as_ref()
84            .map(|g| g.mode.clone())
85            .unwrap_or_else(|| "embedded".to_string());
86
87        match mode.as_str() {
88            "server" => Self::open_server(path).await,
89            _ => Self::open_embedded(path).await,
90        }
91    }
92
93    /// Open the embedded SurrealKV store at `path/surreal/`.
94    pub async fn open_embedded(path: &Path) -> Result<Self, GraphError> {
95        std::fs::create_dir_all(path)?;
96
97        let db = store::open(path).await?;
98        store::init_schema(&db).await?;
99
100        let models_dir = path.join("models");
101        std::fs::create_dir_all(&models_dir)?;
102        let embedder = LazyEmbedder::new(&models_dir);
103
104        let graph_config = load_graph_section(path);
105
106        Ok(Self {
107            db,
108            embedder,
109            path: path.to_path_buf(),
110            scoring: graph_config.scoring,
111            provenance: graph_config.provenance,
112        })
113    }
114
115    /// Connect to a SurrealDB server using `[graph]` settings from
116    /// `.recall-echo.toml` in the parent directory (memory_dir).
117    /// The `path` is still used for the FastEmbed models cache.
118    pub async fn open_server(path: &Path) -> Result<Self, GraphError> {
119        let memory_dir = path.parent().unwrap_or(path);
120        let config = crate::config::load_from_dir(memory_dir);
121
122        let graph_section = config.graph.unwrap_or_default();
123        let password = if graph_section.password_file.is_empty() {
124            String::new()
125        } else {
126            let pw_path = if graph_section.password_file.starts_with('/') {
127                std::path::PathBuf::from(&graph_section.password_file)
128            } else {
129                // Relative to entity root (memory_dir's parent)
130                let entity_root = memory_dir.parent().unwrap_or(memory_dir);
131                entity_root.join(&graph_section.password_file)
132            };
133            std::fs::read_to_string(&pw_path)
134                .map(|s| s.trim().to_string())
135                .map_err(|e| {
136                    GraphError::Io(std::io::Error::new(
137                        e.kind(),
138                        format!(
139                            "failed to read graph password file {}: {e}",
140                            pw_path.display()
141                        ),
142                    ))
143                })?
144        };
145
146        let scoring = graph_section.scoring.clone();
147        let provenance = graph_section.provenance;
148        let server_config = store::ServerConfig {
149            url: graph_section.url,
150            username: graph_section.username,
151            password,
152            namespace: graph_section.namespace,
153            database: graph_section.database,
154        };
155
156        let models_dir = path.join("models");
157        let mut gm = Self::connect(&server_config, &models_dir).await?;
158        gm.scoring = scoring;
159        gm.provenance = provenance;
160        Ok(gm)
161    }
162
163    /// Connect to a SurrealDB server over WebSocket with explicit config.
164    pub async fn connect(
165        config: &store::ServerConfig,
166        models_dir: &Path,
167    ) -> Result<Self, GraphError> {
168        let db = store::connect(config).await?;
169        store::init_schema(&db).await?;
170
171        std::fs::create_dir_all(models_dir)?;
172        let embedder = LazyEmbedder::new(models_dir);
173
174        Ok(Self {
175            db,
176            embedder,
177            path: models_dir.to_path_buf(),
178            scoring: crate::config::GraphScoringConfig::default(),
179            provenance: confidence::ProvenanceWeights::default(),
180        })
181    }
182
183    /// Path to the graph store.
184    pub fn path(&self) -> &Path {
185        &self.path
186    }
187
188    /// Evidence weights this store applies to observations, by provenance
189    /// class (`[graph.provenance]`).
190    #[must_use]
191    pub fn provenance_weights(&self) -> &confidence::ProvenanceWeights {
192        &self.provenance
193    }
194
195    /// Internal access to the database handle.
196    #[allow(dead_code)]
197    pub(crate) fn db(&self) -> &Surreal<Db> {
198        &self.db
199    }
200
201    /// Internal access to the embedder (initializes it on first use).
202    #[allow(dead_code)]
203    pub(crate) fn embedder(&self) -> Result<&FastEmbedder, GraphError> {
204        self.embedder.get()
205    }
206
207    // --- Entity CRUD ---
208
209    /// Add a new entity to the graph.
210    pub async fn add_entity(&self, entity: NewEntity) -> Result<Entity, GraphError> {
211        crud::add_entity(&self.db, self.embedder.get()?, entity).await
212    }
213
214    /// Get an entity by name.
215    pub async fn get_entity(&self, name: &str) -> Result<Option<Entity>, GraphError> {
216        crud::get_entity_by_name(&self.db, name).await
217    }
218
219    /// Get an entity by its record ID.
220    pub async fn get_entity_by_id(&self, id: &str) -> Result<Option<Entity>, GraphError> {
221        crud::get_entity_by_id(&self.db, id).await
222    }
223
224    /// Update an entity's fields.
225    pub async fn update_entity(
226        &self,
227        id: &str,
228        updates: EntityUpdate,
229    ) -> Result<Entity, GraphError> {
230        crud::update_entity(&self.db, self.embedder.get()?, id, updates).await
231    }
232
233    /// Delete an entity and its relationships.
234    pub async fn delete_entity(&self, id: &str) -> Result<(), GraphError> {
235        crud::delete_entity(&self.db, id).await
236    }
237
238    /// List all entities, optionally filtered by type.
239    pub async fn list_entities(
240        &self,
241        entity_type: Option<&str>,
242    ) -> Result<Vec<Entity>, GraphError> {
243        crud::list_entities(&self.db, entity_type).await
244    }
245
246    // --- Relationships ---
247
248    /// Create a relationship between two named entities.
249    pub async fn add_relationship(&self, rel: NewRelationship) -> Result<Relationship, GraphError> {
250        crud::add_relationship(&self.db, rel).await
251    }
252
253    /// Get relationships for an entity.
254    pub async fn get_relationships(
255        &self,
256        entity_name: &str,
257        direction: Direction,
258    ) -> Result<Vec<Relationship>, GraphError> {
259        crud::get_relationships(&self.db, entity_name, direction).await
260    }
261
262    /// Supersede a relationship: close the old one, create a new one.
263    pub async fn supersede_relationship(
264        &self,
265        old_id: &str,
266        new: NewRelationship,
267    ) -> Result<Relationship, GraphError> {
268        crud::supersede_relationship(&self.db, old_id, new).await
269    }
270
271    /// Overwrite a relationship's confidence, resetting its evidence to the
272    /// prior around the new mean.
273    pub async fn update_relationship_confidence(
274        &self,
275        rel_id: &str,
276        confidence: f64,
277    ) -> Result<(), GraphError> {
278        crud::update_relationship_confidence(&self.db, rel_id, confidence).await
279    }
280
281    /// Persist updated evidence for a relationship and reset its decay clock.
282    ///
283    /// Called when a relationship is corroborated: the new posterior mean is
284    /// stored as `confidence`, the coherence tally is stored beside it, and
285    /// `last_reinforced` is set to now, preventing temporal decay from eroding
286    /// the edge.
287    pub async fn reinforce_relationship(
288        &self,
289        rel_id: &str,
290        evidence: confidence::EdgeEvidence,
291    ) -> Result<(), GraphError> {
292        crud::reinforce_relationship(&self.db, rel_id, evidence).await
293    }
294
295    // --- Episodes ---
296
297    /// Add a new episode authored by the agent itself.
298    ///
299    /// The conservative default: a caller that cannot say where the text came
300    /// from must not have it counted as independent evidence. Ingestion, which
301    /// does know, uses [`GraphMemory::add_episode_from`].
302    pub async fn add_episode(&self, episode: NewEpisode) -> Result<Episode, GraphError> {
303        crud::add_episode(&self.db, self.embedder.get()?, episode).await
304    }
305
306    /// Add a new episode stamped with the class of whoever authored it.
307    pub async fn add_episode_from(
308        &self,
309        episode: NewEpisode,
310        provenance: Provenance,
311    ) -> Result<Episode, GraphError> {
312        crud::add_episode_from(&self.db, self.embedder.get()?, episode, provenance).await
313    }
314
315    /// Get episodes by session ID.
316    pub async fn get_episodes_by_session(
317        &self,
318        session_id: &str,
319    ) -> Result<Vec<Episode>, GraphError> {
320        crud::get_episodes_by_session(&self.db, session_id).await
321    }
322
323    /// Get episode by log number.
324    pub async fn get_episode_by_log_number(
325        &self,
326        log_number: u32,
327    ) -> Result<Option<Episode>, GraphError> {
328        crud::get_episode_by_log_number(&self.db, log_number).await
329    }
330
331    // --- Ingestion ---
332
333    /// Ingest a conversation archive into the knowledge graph.
334    ///
335    /// The [`IngestContext`] carries the provenance policy: conversation
336    /// archives infer per chunk from turn roles, document ingestion forces a
337    /// class.
338    pub async fn ingest_archive(
339        &self,
340        archive_text: &str,
341        context: &IngestContext,
342        llm: Option<&dyn llm::LlmProvider>,
343    ) -> Result<IngestionReport, GraphError> {
344        ingest::ingest_archive(self, archive_text, context, llm).await
345    }
346
347    /// Run LLM extraction on an archive without creating episodes.
348    pub async fn extract_from_archive(
349        &self,
350        archive_text: &str,
351        context: &IngestContext,
352        llm: &dyn llm::LlmProvider,
353    ) -> Result<IngestionReport, GraphError> {
354        ingest::extract_from_archive(self, archive_text, context, llm).await
355    }
356
357    /// Mark all episodes with a given log_number as extracted.
358    pub async fn mark_extracted(&self, log_number: u32) -> Result<(), GraphError> {
359        crud::mark_episodes_extracted(&self.db, log_number).await
360    }
361
362    /// Get log numbers of episodes that have NOT been extracted.
363    pub async fn unextracted_log_numbers(&self) -> Result<Vec<i64>, GraphError> {
364        crud::get_unextracted_log_numbers(&self.db).await
365    }
366
367    // --- Search ---
368
369    /// Semantic search across entities (legacy — returns full Entity).
370    pub async fn search(&self, query: &str, limit: usize) -> Result<Vec<SearchResult>, GraphError> {
371        search::search(&self.db, self.embedder.get()?, &self.scoring, query, limit).await
372    }
373
374    /// Search with options — L1 projections, type/keyword filters.
375    pub async fn search_with_options(
376        &self,
377        query: &str,
378        options: &SearchOptions,
379    ) -> Result<Vec<ScoredEntity>, GraphError> {
380        search::search_with_options(
381            &self.db,
382            self.embedder.get()?,
383            &self.scoring,
384            query,
385            options,
386        )
387        .await
388    }
389
390    /// Semantic search across episodes.
391    pub async fn search_episodes(
392        &self,
393        query: &str,
394        limit: usize,
395    ) -> Result<Vec<EpisodeSearchResult>, GraphError> {
396        search::search_episodes(&self.db, self.embedder.get()?, query, limit).await
397    }
398
399    // --- Hybrid Query ---
400
401    /// Hybrid query: semantic + graph expansion + optional episode search.
402    pub async fn query(
403        &self,
404        query_text: &str,
405        options: &QueryOptions,
406    ) -> Result<QueryResult, GraphError> {
407        query::query(
408            &self.db,
409            self.embedder.get()?,
410            &self.scoring,
411            query_text,
412            options,
413        )
414        .await
415    }
416
417    // --- Traversal ---
418
419    /// Traverse the graph from a named entity.
420    pub async fn traverse(
421        &self,
422        entity_name: &str,
423        depth: u32,
424    ) -> Result<TraversalNode, GraphError> {
425        traverse::traverse(&self.db, entity_name, depth).await
426    }
427
428    /// Traverse with type filter.
429    pub async fn traverse_filtered(
430        &self,
431        entity_name: &str,
432        depth: u32,
433        type_filter: Option<&str>,
434    ) -> Result<TraversalNode, GraphError> {
435        traverse::traverse_filtered(&self.db, entity_name, depth, type_filter).await
436    }
437
438    // --- Pipeline ---
439
440    /// Sync pipeline documents into the graph.
441    pub async fn sync_pipeline(
442        &self,
443        docs: &PipelineDocuments,
444    ) -> Result<PipelineSyncReport, GraphError> {
445        pipeline_sync::sync_pipeline(self, docs).await
446    }
447
448    /// Get pipeline stats from the graph.
449    pub async fn pipeline_stats(
450        &self,
451        staleness_days: u32,
452    ) -> Result<PipelineGraphStats, GraphError> {
453        query::pipeline_stats(&self.db, staleness_days).await
454    }
455
456    /// Get pipeline entities by stage and optional status.
457    pub async fn pipeline_entities(
458        &self,
459        stage: &str,
460        status: Option<&str>,
461    ) -> Result<Vec<EntityDetail>, GraphError> {
462        query::pipeline_entities(&self.db, stage, status).await
463    }
464
465    /// Trace pipeline flow for an entity.
466    pub async fn pipeline_flow(
467        &self,
468        entity_name: &str,
469    ) -> Result<Vec<(EntityDetail, String, EntityDetail)>, GraphError> {
470        query::pipeline_flow(&self.db, entity_name).await
471    }
472
473    // --- Vigil Sync ---
474
475    /// Sync vigil signal vectors into the graph as Measurement entities.
476    pub async fn sync_vigil_signals(
477        &self,
478        signals_path: &std::path::Path,
479    ) -> Result<VigilSyncReport, GraphError> {
480        vigil_sync::sync_vigil_signals(self, signals_path).await
481    }
482
483    /// Sync outcome records into the graph as Outcome entities.
484    pub async fn sync_outcomes(
485        &self,
486        outcomes_path: &std::path::Path,
487    ) -> Result<VigilSyncReport, GraphError> {
488        vigil_sync::sync_outcomes(self, outcomes_path).await
489    }
490
491    /// Sync both vigil signals and outcomes in one call.
492    pub async fn sync_vigil(
493        &self,
494        signals_path: &std::path::Path,
495        outcomes_path: &std::path::Path,
496    ) -> Result<VigilSyncReport, GraphError> {
497        vigil_sync::sync_vigil(self, signals_path, outcomes_path).await
498    }
499
500    /// Record outcome feedback: link retrieved entities to a session outcome and
501    /// update their `utility_score` via EMA. `used_entity_ids` distinguishes the
502    /// entities the response actually leaned on (full alpha) from retrieved-but-
503    /// unused (muted alpha). Pass `None` to treat all retrieved as used.
504    pub async fn record_outcome_feedback(
505        &self,
506        session_id: &str,
507        outcome: utility::OutcomeKind,
508        retrieved_entity_ids: &[String],
509        used_entity_ids: Option<&[String]>,
510    ) -> Result<utility::FeedbackReport, GraphError> {
511        utility::record_outcome_feedback(
512            &self.db,
513            session_id,
514            outcome,
515            retrieved_entity_ids,
516            used_entity_ids,
517        )
518        .await
519    }
520
521    /// Apply an outcome to every entity a session touched.
522    ///
523    /// Resolves the session's entities from the `contributed_to` records
524    /// ingestion left behind (falling back to the entities the session
525    /// authored), then records the outcome and moves their utility scores.
526    /// The report says which entities moved and where they landed.
527    pub async fn record_session_outcome(
528        &self,
529        session_id: &str,
530        outcome: utility::OutcomeKind,
531    ) -> Result<utility::FeedbackReport, GraphError> {
532        let session = utility::session_entities(&self.db, session_id).await?;
533        if session.is_empty() {
534            return Ok(utility::FeedbackReport::default());
535        }
536
537        utility::record_outcome_feedback(
538            &self.db,
539            session_id,
540            outcome,
541            &session.retrieved,
542            Some(&session.used),
543        )
544        .await
545    }
546
547    /// Record that a session touched these entities, without judging it.
548    pub async fn record_session_use(
549        &self,
550        session_id: &str,
551        entity_ids: &[String],
552    ) -> Result<u32, GraphError> {
553        utility::record_session_use(&self.db, session_id, entity_ids).await
554    }
555
556    // --- Garbage Collection ---
557
558    /// Run garbage collection with the given config.
559    pub async fn run_gc(&self, config: &gc::GcConfig) -> Result<gc::GcReport, GraphError> {
560        gc::run_gc(&self.db, config).await
561    }
562
563    /// Get GC health stats without running collection.
564    pub async fn gc_stats(&self) -> Result<gc::GcStatsReport, GraphError> {
565        gc::stats_only(&self.db).await
566    }
567
568    /// Delete a single relationship by ID.
569    pub async fn delete_relationship(&self, id: &str) -> Result<(), GraphError> {
570        crud::delete_relationship(&self.db, id).await
571    }
572
573    // --- Stats ---
574
575    /// Get graph statistics.
576    pub async fn stats(&self) -> Result<GraphStats, GraphError> {
577        let entity_count = db_count(&self.db, "entity").await?;
578        let relationship_count = db_count(&self.db, "relates_to").await?;
579        let episode_count = db_count(&self.db, "episode").await?;
580
581        // Count by type
582        let mut type_response = self
583            .db
584            .query("SELECT entity_type, count() AS count FROM entity GROUP BY entity_type")
585            .await?;
586
587        let type_rows: Vec<TypeCount> = type_response.take(0)?;
588        let entity_type_counts: HashMap<String, u64> = type_rows
589            .into_iter()
590            .map(|r| (r.entity_type, r.count))
591            .collect();
592
593        Ok(GraphStats {
594            entity_count,
595            relationship_count,
596            episode_count,
597            entity_type_counts,
598        })
599    }
600}
601
602/// Load `[graph]` from `.recall-echo.toml` in the memory directory (the parent
603/// of the graph store path). Returns defaults if the config file or the
604/// section is absent, preserving legacy behavior.
605fn load_graph_section(graph_path: &Path) -> crate::config::GraphSection {
606    let memory_dir = graph_path.parent().unwrap_or(graph_path);
607    crate::config::load_from_dir(memory_dir)
608        .graph
609        .unwrap_or_default()
610}
611
612async fn db_count(db: &Surreal<Db>, table: &str) -> Result<u64, GraphError> {
613    let query = format!("SELECT count() AS count FROM {table} GROUP ALL");
614    let mut response = db.query(&query).await?;
615    let rows: Vec<CountRow> = response.take(0)?;
616    Ok(rows.first().map(|r| r.count).unwrap_or(0))
617}
618
619#[derive(serde::Deserialize, surrealdb::types::SurrealValue)]
620struct CountRow {
621    count: u64,
622}
623
624#[derive(serde::Deserialize, surrealdb::types::SurrealValue)]
625struct TypeCount {
626    entity_type: String,
627    count: u64,
628}
recall_echo/graph/mod.rs

recall_echo/graph/
mod.rs
