recall_echo/
graph_bridge.rs

1//! Bridge between recall-echo and recall-graph.
2//!
3//! Provides graph ingestion for archived conversations.
4//! When pulse-null feature is enabled, also bridges LmProvider → LlmProvider.
5
6/// Ingest a conversation archive into the knowledge graph.
7///
8/// Non-blocking: logs warnings on failure but never fails the caller.
9/// Returns the ingestion report on success.
10pub async fn ingest_into_graph(
11    memory_dir: &std::path::Path,
12    archive_content: &str,
13    session_id: &str,
14    log_number: Option<u32>,
15) -> Result<crate::graph::types::IngestionReport, crate::error::RecallError> {
16    let graph_dir = memory_dir.join("graph");
17    if !graph_dir.exists() {
18        return Err(crate::error::RecallError::NotInitialized(
19            "graph/ not initialized \u{2014} run `graph init` first".into(),
20        ));
21    }
22
23    // Hot path (SessionEnd hook): goes through the serve daemon so a
24    // concurrent session never collides on the embedded store lock.
25    // No LLM provider in standalone mode — episodes only, no entity extraction.
26    let request = crate::serve::Request::IngestArchive(crate::serve::IngestArchiveArgs {
27        content: archive_content.to_string(),
28        session_id: session_id.to_string(),
29        log_number,
30    });
31    let report: crate::graph::types::IngestionReport =
32        serde_json::from_value(crate::serve_client::execute(memory_dir, &request).await?)?;
33
34    eprintln!(
35        "recall-echo: graph ingested \u{2014} {} episodes, {} entities created, {} merged, {} skipped, {} relationships",
36        report.episodes_created,
37        report.entities_created,
38        report.entities_merged,
39        report.entities_skipped,
40        report.relationships_created,
41    );
42
43    if !report.errors.is_empty() {
44        eprintln!(
45            "recall-echo: graph ingestion had {} warnings",
46            report.errors.len()
47        );
48    }
49
50    Ok(report)
51}
52
53/// Sync the pipeline documents into the knowledge graph.
54///
55/// Pipeline sync needs no LLM provider, so it runs as an ordinary daemon
56/// request. That matters on the SessionEnd hook path: ingest and sync then
57/// share one warm daemon instead of the sync stopping the daemon the ingest
58/// just started and reloading the embedding model in-process.
59pub async fn sync_pipeline_into_graph(
60    memory_dir: &std::path::Path,
61    docs: crate::graph::types::PipelineDocuments,
62) -> Result<crate::graph::types::PipelineSyncReport, crate::error::RecallError> {
63    let request = crate::serve::Request::SyncPipeline(crate::serve::SyncPipelineArgs { docs });
64    Ok(serde_json::from_value(
65        crate::serve_client::execute(memory_dir, &request).await?,
66    )?)
67}
68
69/// Ingest with an LLM provider for entity extraction.
70///
71/// When pulse-null feature is enabled, this bridges the LmProvider
72/// to recall-graph's LlmProvider for full entity/relationship extraction.
73#[cfg(feature = "pulse-null")]
74pub async fn ingest_into_graph_with_llm(
75    memory_dir: &std::path::Path,
76    archive_content: &str,
77    session_id: &str,
78    log_number: Option<u32>,
79    provider: Option<&dyn pulse_system_types::llm::LmProvider>,
80) -> Result<crate::graph::types::IngestionReport, crate::error::RecallError> {
81    let graph_dir = memory_dir.join("graph");
82    if !graph_dir.exists() {
83        return Err(crate::error::RecallError::NotInitialized(
84            "graph/ not initialized \u{2014} run `graph init` first".into(),
85        ));
86    }
87
88    // LLM-backed extraction cannot cross the socket (the provider lives in
89    // this process), so this path takes the store exclusively instead.
90    let report = crate::serve_client::exclusive(memory_dir, |gm| async move {
91        let bridge = provider.map(GraphLlmBridge::new);
92        let llm_ref: Option<&dyn crate::graph::llm::LlmProvider> = bridge
93            .as_ref()
94            .map(|b| b as &dyn crate::graph::llm::LlmProvider);
95
96        Ok(gm
97            .ingest_archive(archive_content, session_id, log_number, llm_ref)
98            .await?)
99    })
100    .await?;
101
102    eprintln!(
103        "recall-echo: graph ingested \u{2014} {} episodes, {} entities created, {} merged, {} skipped, {} relationships",
104        report.episodes_created,
105        report.entities_created,
106        report.entities_merged,
107        report.entities_skipped,
108        report.relationships_created,
109    );
110
111    if !report.errors.is_empty() {
112        eprintln!(
113            "recall-echo: graph ingestion had {} warnings",
114            report.errors.len()
115        );
116    }
117
118    Ok(report)
119}
120
121/// Adapter that wraps an `pulse_system_types::LmProvider` to implement
122/// `crate::graph::LlmProvider`.
123#[cfg(feature = "pulse-null")]
124pub struct GraphLlmBridge<'a> {
125    provider: &'a dyn pulse_system_types::llm::LmProvider,
126}
127
128#[cfg(feature = "pulse-null")]
129impl<'a> GraphLlmBridge<'a> {
130    pub fn new(provider: &'a dyn pulse_system_types::llm::LmProvider) -> Self {
131        Self { provider }
132    }
133}
134
135#[cfg(feature = "pulse-null")]
136#[async_trait::async_trait]
137impl crate::graph::llm::LlmProvider for GraphLlmBridge<'_> {
138    async fn complete(
139        &self,
140        system_prompt: &str,
141        user_message: &str,
142        max_tokens: u32,
143    ) -> Result<String, crate::graph::error::GraphError> {
144        use pulse_system_types::llm::{Message, MessageContent, Role};
145
146        let messages = vec![Message {
147            role: Role::User,
148            content: MessageContent::Text(user_message.to_string()),
149            source: None,
150        }];
151
152        let response = self
153            .provider
154            .invoke(system_prompt, &messages, max_tokens, None)
155            .await
156            .map_err(|e| crate::graph::error::GraphError::Llm(e.to_string()))?;
157
158        Ok(response.text())
159    }
160}
recall_echo/graph_bridge.rs

recall_echo/
graph_bridge.rs
