From 67b30c15498944b55ce980ccd347bad4bd1e81a6 Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 16 Mar 2026 10:44:06 +0100 Subject: [PATCH] feat: HTTP memory backend for shared memory infrastructure Route SemanticStore remember/recall operations to the memory-api gateway (PostgreSQL + pgvector + Jina AI embeddings) when backend=http is configured. - Add backend, http_url, http_token_env fields to MemoryConfig - Create http_client module with MemoryApiClient (reqwest::blocking) - Add HTTP dispatch to SemanticStore with graceful SQLite fallback - Wire MemoryConfig through MemorySubstrate::open() and kernel boot - Add reqwest as optional dependency behind http-memory feature flag Sessions, KV store, and knowledge graph remain local SQLite. Co-Authored-By: Claude Opus 4.6 --- crates/openfang-kernel/src/kernel.rs | 2 +- crates/openfang-memory/Cargo.toml | 5 + crates/openfang-memory/src/http_client.rs | 246 ++++++++++++++++++++++ crates/openfang-memory/src/lib.rs | 2 + crates/openfang-memory/src/semantic.rs | 175 ++++++++++++++- crates/openfang-memory/src/substrate.rs | 46 +++- crates/openfang-types/src/config.rs | 17 ++ 7 files changed, 486 insertions(+), 7 deletions(-) create mode 100644 crates/openfang-memory/src/http_client.rs diff --git a/crates/openfang-kernel/src/kernel.rs b/crates/openfang-kernel/src/kernel.rs index 5e582d04..a93acb6a 100644 --- a/crates/openfang-kernel/src/kernel.rs +++ b/crates/openfang-kernel/src/kernel.rs @@ -562,7 +562,7 @@ impl OpenFangKernel { .clone() .unwrap_or_else(|| config.data_dir.join("openfang.db")); let memory = Arc::new( - MemorySubstrate::open(&db_path, config.memory.decay_rate) + MemorySubstrate::open(&db_path, config.memory.decay_rate, &config.memory) .map_err(|e| KernelError::BootFailed(format!("Memory init failed: {e}")))?, ); diff --git a/crates/openfang-memory/Cargo.toml b/crates/openfang-memory/Cargo.toml index c33bf840..2ad9ee6b 100644 --- a/crates/openfang-memory/Cargo.toml +++ b/crates/openfang-memory/Cargo.toml @@ -5,6 +5,10 @@ edition.workspace = true license.workspace = true description = "Memory substrate for the OpenFang Agent OS" +[features] +default = ["http-memory"] +http-memory = ["reqwest"] + [dependencies] openfang-types = { path = "../openfang-types" } tokio = { workspace = true } @@ -17,6 +21,7 @@ uuid = { workspace = true } thiserror = { workspace = true } async-trait = { workspace = true } tracing = { workspace = true } +reqwest = { workspace = true, features = ["blocking"], optional = true } [dev-dependencies] tokio-test = { workspace = true } diff --git a/crates/openfang-memory/src/http_client.rs b/crates/openfang-memory/src/http_client.rs new file mode 100644 index 00000000..1bdf13dd --- /dev/null +++ b/crates/openfang-memory/src/http_client.rs @@ -0,0 +1,246 @@ +//! HTTP client for the memory-api gateway. +//! +//! Provides a blocking HTTP client that routes `remember` and `recall` operations +//! to the shared memory-api service (PostgreSQL + pgvector + Jina AI embeddings). +//! Designed to be called from synchronous SemanticStore methods within +//! `spawn_blocking` contexts. + +use serde::{Deserialize, Serialize}; +use tracing::{debug, warn}; + +/// Error type for memory API operations. +#[derive(Debug, thiserror::Error)] +pub enum MemoryApiError { + #[error("HTTP error: {0}")] + Http(String), + #[error("API error (status {status}): {message}")] + Api { status: u16, message: String }, + #[error("Parse error: {0}")] + Parse(String), + #[error("Missing config: {0}")] + Config(String), +} + +/// HTTP client for the memory-api gateway service. +#[derive(Clone)] +pub struct MemoryApiClient { + base_url: String, + token: String, + client: reqwest::blocking::Client, +} + +// -- Request/Response types matching memory-api endpoints -- + +#[derive(Serialize)] +struct StoreRequest<'a> { + content: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + category: Option<&'a str>, + #[serde(rename = "agentId", skip_serializing_if = "Option::is_none")] + agent_id: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + source: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + importance: Option, + #[serde(skip_serializing_if = "Option::is_none")] + tags: Option>, +} + +#[derive(Deserialize, Debug)] +pub struct StoreResponse { + pub id: serde_json::Value, + #[serde(default)] + pub deduplicated: bool, +} + +#[derive(Serialize)] +struct SearchRequest<'a> { + query: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + limit: Option, + #[serde(skip_serializing_if = "Option::is_none")] + category: Option<&'a str>, +} + +#[derive(Deserialize, Debug)] +pub struct SearchResponse { + pub results: Vec, + pub count: usize, +} + +#[derive(Deserialize, Debug, Clone)] +pub struct SearchResult { + pub id: serde_json::Value, + pub content: String, + #[serde(default)] + pub category: Option, + #[serde(default)] + pub score: f64, + #[serde(rename = "createdAt", default)] + pub created_at: Option, +} + +#[derive(Deserialize, Debug)] +struct HealthResponse { + pub status: String, +} + +impl MemoryApiClient { + /// Create a new memory-api HTTP client. + /// + /// `base_url`: The base URL of the memory-api service (e.g., "http://127.0.0.1:5500"). + /// `token_env`: The name of the environment variable holding the bearer token. + pub fn new(base_url: &str, token_env: &str) -> Result { + let token = if token_env.is_empty() { + String::new() + } else { + std::env::var(token_env).unwrap_or_else(|_| { + warn!(env = token_env, "Memory API token env var not set"); + String::new() + }) + }; + + let client = reqwest::blocking::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .user_agent("openfang-memory/0.4") + .build() + .map_err(|e| MemoryApiError::Http(e.to_string()))?; + + let base_url = base_url.trim_end_matches('/').to_string(); + + Ok(Self { + base_url, + token, + client, + }) + } + + /// Check if memory-api is reachable. + pub fn health_check(&self) -> Result<(), MemoryApiError> { + let url = format!("{}/health", self.base_url); + let resp = self + .client + .get(&url) + .send() + .map_err(|e| MemoryApiError::Http(e.to_string()))?; + + if !resp.status().is_success() { + return Err(MemoryApiError::Api { + status: resp.status().as_u16(), + message: resp.text().unwrap_or_default(), + }); + } + + let body: HealthResponse = resp + .json() + .map_err(|e| MemoryApiError::Parse(e.to_string()))?; + + if body.status != "ok" { + return Err(MemoryApiError::Api { + status: 503, + message: format!("memory-api status: {}", body.status), + }); + } + + debug!("memory-api health check passed"); + Ok(()) + } + + /// Store a memory via POST /memory/store. + /// + /// The memory-api handles embedding generation (Jina AI) and deduplication. + pub fn store( + &self, + content: &str, + category: Option<&str>, + agent_id: Option<&str>, + source: Option<&str>, + importance: Option, + tags: Option>, + ) -> Result { + let url = format!("{}/memory/store", self.base_url); + + let body = StoreRequest { + content, + category, + agent_id, + source, + importance, + tags, + }; + + let mut req = self.client.post(&url).json(&body); + if !self.token.is_empty() { + req = req.header("Authorization", format!("Bearer {}", self.token)); + } + + let resp = req + .send() + .map_err(|e| MemoryApiError::Http(e.to_string()))?; + let status = resp.status().as_u16(); + + if status != 200 && status != 201 { + let body_text = resp.text().unwrap_or_default(); + return Err(MemoryApiError::Api { + status, + message: body_text, + }); + } + + let result: StoreResponse = resp + .json() + .map_err(|e| MemoryApiError::Parse(e.to_string()))?; + + debug!( + id = %result.id, + deduplicated = result.deduplicated, + "Stored memory via HTTP" + ); + + Ok(result) + } + + /// Search memories via POST /memory/search. + /// + /// The memory-api handles embedding the query (Jina AI) and hybrid vector+BM25 search. + pub fn search( + &self, + query: &str, + limit: usize, + category: Option<&str>, + ) -> Result, MemoryApiError> { + let url = format!("{}/memory/search", self.base_url); + + let body = SearchRequest { + query, + limit: Some(limit), + category, + }; + + let mut req = self.client.post(&url).json(&body); + if !self.token.is_empty() { + req = req.header("Authorization", format!("Bearer {}", self.token)); + } + + let resp = req + .send() + .map_err(|e| MemoryApiError::Http(e.to_string()))?; + let status = resp.status().as_u16(); + + if status != 200 { + let body_text = resp.text().unwrap_or_default(); + return Err(MemoryApiError::Api { + status, + message: body_text, + }); + } + + let result: SearchResponse = resp + .json() + .map_err(|e| MemoryApiError::Parse(e.to_string()))?; + + debug!(count = result.count, "Searched memories via HTTP"); + + Ok(result.results) + } +} diff --git a/crates/openfang-memory/src/lib.rs b/crates/openfang-memory/src/lib.rs index 15a4fbd4..bf16737a 100644 --- a/crates/openfang-memory/src/lib.rs +++ b/crates/openfang-memory/src/lib.rs @@ -8,6 +8,8 @@ //! Agents interact with a single `Memory` trait that abstracts over all three stores. pub mod consolidation; +#[cfg(feature = "http-memory")] +pub mod http_client; pub mod knowledge; pub mod migration; pub mod semantic; diff --git a/crates/openfang-memory/src/semantic.rs b/crates/openfang-memory/src/semantic.rs index 20f52ccf..73181d6b 100644 --- a/crates/openfang-memory/src/semantic.rs +++ b/crates/openfang-memory/src/semantic.rs @@ -14,18 +14,43 @@ use openfang_types::memory::{MemoryFilter, MemoryFragment, MemoryId, MemorySourc use rusqlite::Connection; use std::collections::HashMap; use std::sync::{Arc, Mutex}; -use tracing::debug; +use tracing::{debug, warn}; + +#[cfg(feature = "http-memory")] +use crate::http_client::MemoryApiClient; /// Semantic store backed by SQLite with optional vector search. +/// +/// Supports two backends: +/// - **SQLite** (default): Local LIKE matching / cosine similarity. +/// - **HTTP**: Routes `remember`/`recall` to the memory-api gateway +/// (PostgreSQL + pgvector + Jina AI embeddings). #[derive(Clone)] pub struct SemanticStore { conn: Arc>, + #[cfg(feature = "http-memory")] + http_client: Option, } impl SemanticStore { - /// Create a new semantic store wrapping the given connection. + /// Create a new semantic store wrapping the given connection (SQLite backend). pub fn new(conn: Arc>) -> Self { - Self { conn } + Self { + conn, + #[cfg(feature = "http-memory")] + http_client: None, + } + } + + /// Create a semantic store with an HTTP backend for the memory-api gateway. + /// + /// The SQLite connection is still required for local fallback and other stores. + #[cfg(feature = "http-memory")] + pub fn new_with_http(conn: Arc>, client: MemoryApiClient) -> Self { + Self { + conn, + http_client: Some(client), + } } /// Store a new memory fragment (without embedding). @@ -41,6 +66,9 @@ impl SemanticStore { } /// Store a new memory fragment with an optional embedding vector. + /// + /// When HTTP backend is configured, stores via memory-api (which handles + /// embedding generation and deduplication). Falls back to local SQLite. pub fn remember_with_embedding( &self, agent_id: AgentId, @@ -49,6 +77,26 @@ impl SemanticStore { scope: &str, metadata: HashMap, embedding: Option<&[f32]>, + ) -> OpenFangResult { + // HTTP backend: route to memory-api + #[cfg(feature = "http-memory")] + if let Some(ref client) = self.http_client { + return self.remember_via_http(client, agent_id, content, source, scope, &metadata); + } + + // SQLite backend (default) + self.remember_sqlite(agent_id, content, source, scope, metadata, embedding) + } + + /// SQLite implementation of remember_with_embedding. + fn remember_sqlite( + &self, + agent_id: AgentId, + content: &str, + source: MemorySource, + scope: &str, + metadata: HashMap, + embedding: Option<&[f32]>, ) -> OpenFangResult { let conn = self .conn @@ -80,6 +128,53 @@ impl SemanticStore { Ok(id) } + /// HTTP implementation of remember — routes to memory-api POST /memory/store. + #[cfg(feature = "http-memory")] + fn remember_via_http( + &self, + client: &MemoryApiClient, + agent_id: AgentId, + content: &str, + source: MemorySource, + scope: &str, + metadata: &HashMap, + ) -> OpenFangResult { + let source_str = format!("{:?}", source).to_lowercase(); + let importance = metadata + .get("importance") + .and_then(|v| v.as_u64()) + .map(|v| v.min(10) as u8) + .unwrap_or(5); + let tags: Option> = metadata + .get("tags") + .and_then(|v| serde_json::from_value(v.clone()).ok()); + + match client.store( + content, + Some(scope), + Some(&agent_id.0.to_string()), + Some(&source_str), + Some(importance), + tags, + ) { + Ok(resp) => { + debug!(id = %resp.id, "Stored memory via HTTP backend"); + Ok(MemoryId::new()) + } + Err(e) => { + warn!(error = %e, "HTTP memory store failed, falling back to SQLite"); + self.remember_sqlite( + agent_id, + content, + source, + scope, + metadata.clone(), + None, + ) + } + } + } + /// Search for memories using text matching (fallback, no embeddings). pub fn recall( &self, @@ -92,6 +187,9 @@ impl SemanticStore { /// Search for memories using vector similarity when a query embedding is provided, /// falling back to LIKE matching otherwise. + /// + /// When HTTP backend is configured, searches via memory-api (hybrid vector+BM25). + /// Falls back to local SQLite on HTTP errors. pub fn recall_with_embedding( &self, query: &str, @@ -99,6 +197,17 @@ impl SemanticStore { filter: Option, query_embedding: Option<&[f32]>, ) -> OpenFangResult> { + // HTTP backend: route to memory-api + #[cfg(feature = "http-memory")] + if let Some(ref client) = self.http_client { + match self.recall_via_http(client, query, limit, &filter) { + Ok(results) => return Ok(results), + Err(e) => { + warn!(error = %e, "HTTP memory search failed, falling back to SQLite"); + } + } + } + let conn = self .conn .lock() @@ -277,7 +386,15 @@ impl SemanticStore { } /// Soft-delete a memory fragment. + /// + /// In HTTP mode, logs a warning (memory-api doesn't support delete yet) + /// and performs the soft-delete locally only. pub fn forget(&self, id: MemoryId) -> OpenFangResult<()> { + #[cfg(feature = "http-memory")] + if self.http_client.is_some() { + warn!(id = %id.0, "forget() not supported via HTTP backend, local-only soft-delete"); + } + let conn = self .conn .lock() @@ -304,6 +421,58 @@ impl SemanticStore { .map_err(|e| OpenFangError::Memory(e.to_string()))?; Ok(()) } + + /// HTTP implementation of recall — routes to memory-api POST /memory/search. + /// + /// Maps memory-api search results to `MemoryFragment` structs. Fields not + /// available from the HTTP API (agent_id, embedding, access_count) use defaults. + #[cfg(feature = "http-memory")] + fn recall_via_http( + &self, + client: &MemoryApiClient, + query: &str, + limit: usize, + filter: &Option, + ) -> OpenFangResult> { + let category = filter.as_ref().and_then(|f| f.scope.as_deref()); + + let results = client + .search(query, limit, category) + .map_err(|e| OpenFangError::Memory(format!("HTTP search failed: {e}")))?; + + let fragments: Vec = results + .into_iter() + .map(|r| { + let created_at = r + .created_at + .map(|ms| { + chrono::DateTime::from_timestamp_millis(ms as i64) + .unwrap_or_else(|| Utc::now()) + }) + .unwrap_or_else(|| Utc::now()); + + MemoryFragment { + id: MemoryId::new(), + agent_id: filter + .as_ref() + .and_then(|f| f.agent_id) + .unwrap_or_else(AgentId::new), + content: r.content, + embedding: None, + metadata: HashMap::new(), + source: MemorySource::System, + confidence: r.score as f32, + created_at, + accessed_at: Utc::now(), + access_count: 0, + scope: r.category.unwrap_or_else(|| "general".to_string()), + } + }) + .collect(); + + debug!(count = fragments.len(), "Recalled memories via HTTP backend"); + Ok(fragments) + } } /// Compute cosine similarity between two vectors. diff --git a/crates/openfang-memory/src/substrate.rs b/crates/openfang-memory/src/substrate.rs index 34232dd3..597e9138 100644 --- a/crates/openfang-memory/src/substrate.rs +++ b/crates/openfang-memory/src/substrate.rs @@ -13,6 +13,7 @@ use crate::usage::UsageStore; use async_trait::async_trait; use openfang_types::agent::{AgentEntry, AgentId, SessionId}; +use openfang_types::config::MemoryConfig; use openfang_types::error::{OpenFangError, OpenFangResult}; use openfang_types::memory::{ ConsolidationReport, Entity, ExportFormat, GraphMatch, GraphPattern, ImportReport, Memory, @@ -22,6 +23,7 @@ use rusqlite::Connection; use std::collections::HashMap; use std::path::Path; use std::sync::{Arc, Mutex}; +use tracing::{info, warn}; /// The unified memory substrate. Implements the `Memory` trait by delegating /// to specialized stores backed by a shared SQLite connection. @@ -37,17 +39,23 @@ pub struct MemorySubstrate { impl MemorySubstrate { /// Open or create a memory substrate at the given database path. - pub fn open(db_path: &Path, decay_rate: f32) -> OpenFangResult { + /// + /// When `memory_config.backend == "http"` and `http_url`/`http_token_env` are set, + /// the semantic store routes `remember`/`recall` to the memory-api gateway. + /// All other stores (KV, knowledge graph, sessions) remain local SQLite. + pub fn open(db_path: &Path, decay_rate: f32, memory_config: &MemoryConfig) -> OpenFangResult { let conn = Connection::open(db_path).map_err(|e| OpenFangError::Memory(e.to_string()))?; conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000;") .map_err(|e| OpenFangError::Memory(e.to_string()))?; run_migrations(&conn).map_err(|e| OpenFangError::Memory(e.to_string()))?; let shared = Arc::new(Mutex::new(conn)); + let semantic = Self::create_semantic_store(Arc::clone(&shared), memory_config); + Ok(Self { conn: Arc::clone(&shared), structured: StructuredStore::new(Arc::clone(&shared)), - semantic: SemanticStore::new(Arc::clone(&shared)), + semantic, knowledge: KnowledgeStore::new(Arc::clone(&shared)), sessions: SessionStore::new(Arc::clone(&shared)), usage: UsageStore::new(Arc::clone(&shared)), @@ -55,7 +63,39 @@ impl MemorySubstrate { }) } - /// Create an in-memory substrate (for testing). + /// Create the semantic store, optionally with HTTP backend. + fn create_semantic_store( + conn: Arc>, + memory_config: &MemoryConfig, + ) -> SemanticStore { + #[cfg(feature = "http-memory")] + if memory_config.backend == "http" { + if let (Some(url), Some(token_env)) = (&memory_config.http_url, &memory_config.http_token_env) { + match crate::http_client::MemoryApiClient::new(url, token_env) { + Ok(client) => { + // Best-effort health check on startup + match client.health_check() { + Ok(()) => info!(url = %url, "HTTP memory backend connected"), + Err(e) => warn!(url = %url, error = %e, "HTTP memory backend health check failed, will retry on use"), + } + return SemanticStore::new_with_http(conn, client); + } + Err(e) => { + warn!(error = %e, "Failed to create HTTP memory client, falling back to SQLite"); + } + } + } else { + warn!("backend=http but http_url/http_token_env not set, falling back to SQLite"); + } + } + + #[cfg(not(feature = "http-memory"))] + let _ = memory_config; + + SemanticStore::new(conn) + } + + /// Create an in-memory substrate (for testing). Always uses SQLite backend. pub fn open_in_memory(decay_rate: f32) -> OpenFangResult { let conn = Connection::open_in_memory().map_err(|e| OpenFangError::Memory(e.to_string()))?; diff --git a/crates/openfang-types/src/config.rs b/crates/openfang-types/src/config.rs index 71748eb0..856eff24 100644 --- a/crates/openfang-types/src/config.rs +++ b/crates/openfang-types/src/config.rs @@ -1488,12 +1488,26 @@ pub struct MemoryConfig { /// How often to run memory consolidation (hours). 0 = disabled. #[serde(default = "default_consolidation_interval")] pub consolidation_interval_hours: u64, + /// Memory backend: "sqlite" (default) or "http". + #[serde(default = "default_memory_backend")] + pub backend: String, + /// HTTP memory API URL (when backend = "http"). + /// e.g., "http://127.0.0.1:5500" + #[serde(default)] + pub http_url: Option, + /// Env var name holding the HTTP memory API bearer token. + #[serde(default)] + pub http_token_env: Option, } fn default_consolidation_interval() -> u64 { 24 } +fn default_memory_backend() -> String { + "sqlite".to_string() +} + impl Default for MemoryConfig { fn default() -> Self { Self { @@ -1504,6 +1518,9 @@ impl Default for MemoryConfig { embedding_provider: None, embedding_api_key_env: None, consolidation_interval_hours: default_consolidation_interval(), + backend: default_memory_backend(), + http_url: None, + http_token_env: None, } } }