From 5e660f2fe83b7c2cea70d1e298687ca3c9d43d12 Mon Sep 17 00:00:00 2001 From: Jwalin Shah Date: Thu, 23 Apr 2026 09:57:26 -0700 Subject: [PATCH] feat(core): memory namespaces, recall citations, provider_surfaces RPC (#803) Co-authored-by: Jwalin Shah Co-authored-by: Steven Enamakel --- src/core/all.rs | 8 + src/core/socketio.rs | 3 + src/openhuman/agent/harness/memory_context.rs | 25 +- .../agent/harness/session/builder.rs | 1 + .../agent/harness/session/runtime.rs | 7 + src/openhuman/agent/harness/session/turn.rs | 41 ++- src/openhuman/agent/harness/session/types.rs | 3 + .../agent/harness/subagent_runner/ops.rs | 12 +- src/openhuman/agent/memory_loader.rs | 175 +++++++++++- src/openhuman/channels/context.rs | 19 +- src/openhuman/channels/proactive.rs | 1 + .../channels/providers/presentation.rs | 16 ++ src/openhuman/channels/providers/web.rs | 34 ++- src/openhuman/channels/routes.rs | 14 +- src/openhuman/channels/runtime/dispatch.rs | 1 + src/openhuman/channels/tests/common.rs | 14 +- src/openhuman/channels/tests/memory.rs | 19 +- src/openhuman/cost/tracker.rs | 2 +- src/openhuman/learning/prompt_sections.rs | 14 +- src/openhuman/learning/reflection.rs | 19 +- src/openhuman/learning/tool_tracker.rs | 32 ++- src/openhuman/learning/user_profile.rs | 20 +- src/openhuman/memory/mod.rs | 2 +- src/openhuman/memory/store/memory_trait.rs | 257 ++++++++++++++---- src/openhuman/memory/store/unified/init.rs | 52 ++++ src/openhuman/memory/traits.rs | 52 ++-- src/openhuman/migration/core.rs | 6 +- src/openhuman/mod.rs | 1 + src/openhuman/provider_surfaces/mod.rs | 18 ++ src/openhuman/provider_surfaces/ops.rs | 136 +++++++++ src/openhuman/provider_surfaces/rpc.rs | 4 + src/openhuman/provider_surfaces/schemas.rs | 151 ++++++++++ src/openhuman/provider_surfaces/store.rs | 70 +++++ src/openhuman/provider_surfaces/types.rs | 66 +++++ src/openhuman/tools/impl/memory/forget.rs | 40 ++- src/openhuman/tools/impl/memory/recall.rs | 32 ++- src/openhuman/tools/impl/memory/store.rs | 24 +- src/openhuman/tools/impl/system/tool_stats.rs | 18 +- tests/agent_builder_public.rs | 14 +- tests/agent_harness_public.rs | 14 +- tests/agent_memory_loader_public.rs | 14 +- 41 files changed, 1287 insertions(+), 164 deletions(-) create mode 100644 src/openhuman/provider_surfaces/mod.rs create mode 100644 src/openhuman/provider_surfaces/ops.rs create mode 100644 src/openhuman/provider_surfaces/rpc.rs create mode 100644 src/openhuman/provider_surfaces/schemas.rs create mode 100644 src/openhuman/provider_surfaces/store.rs create mode 100644 src/openhuman/provider_surfaces/types.rs diff --git a/src/core/all.rs b/src/core/all.rs index 11d268ad2..2745919d1 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -126,6 +126,10 @@ fn build_registered_controllers() -> Vec { controllers.extend(crate::openhuman::billing::all_billing_registered_controllers()); // Team and role management controllers.extend(crate::openhuman::team::all_team_registered_controllers()); + // Local assistive surfaces over third-party provider apps + controllers.extend( + crate::openhuman::provider_surfaces::all_provider_surfaces_registered_controllers(), + ); // OS-level text input interactions controllers.extend(crate::openhuman::text_input::all_text_input_registered_controllers()); // Voice transcription and synthesis @@ -189,6 +193,7 @@ fn build_declared_controller_schemas() -> Vec { schemas.extend(crate::openhuman::referral::all_referral_controller_schemas()); schemas.extend(crate::openhuman::billing::all_billing_controller_schemas()); schemas.extend(crate::openhuman::team::all_team_controller_schemas()); + schemas.extend(crate::openhuman::provider_surfaces::all_provider_surfaces_controller_schemas()); schemas.extend(crate::openhuman::text_input::all_text_input_controller_schemas()); schemas.extend(crate::openhuman::voice::all_voice_controller_schemas()); schemas.extend(crate::openhuman::subconscious::all_subconscious_controller_schemas()); @@ -255,6 +260,9 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> { "referral" => Some("Referral codes, stats, and apply flows via the hosted backend API."), "billing" => Some("Subscription plan, payment links, and credit top-up via the backend."), "team" => Some("Team member management, invites, and role changes via the backend."), + "provider_surfaces" => Some( + "Local-first assistive surfaces for provider events, respond queues, and drafts.", + ), "voice" => Some("Speech-to-text and text-to-speech using local models."), "subconscious" => Some("Periodic local-model background awareness loop."), "text_input" => Some("Read, insert, and preview text in the OS-focused input field."), diff --git a/src/core/socketio.rs b/src/core/socketio.rs index bd431d622..f2012ae30 100644 --- a/src/core/socketio.rs +++ b/src/core/socketio.rs @@ -69,6 +69,9 @@ pub struct WebChannelEvent { /// `tool_result` events. #[serde(skip_serializing_if = "Option::is_none")] pub tool_call_id: Option, + /// Optional citations attached to `chat_done` payloads. + #[serde(skip_serializing_if = "Option::is_none")] + pub citations: Option, } #[derive(Debug, Deserialize)] diff --git a/src/openhuman/agent/harness/memory_context.rs b/src/openhuman/agent/harness/memory_context.rs index bc2ef98c4..88bf413c5 100644 --- a/src/openhuman/agent/harness/memory_context.rs +++ b/src/openhuman/agent/harness/memory_context.rs @@ -17,7 +17,10 @@ pub(crate) async fn build_context( let mut seen_keys = HashSet::new(); // Pull relevant memories for this message - if let Ok(entries) = mem.recall(user_msg, 5, None).await { + if let Ok(entries) = mem + .recall(user_msg, 5, crate::openhuman::memory::RecallOpts::default()) + .await + { let relevant: Vec<_> = entries .iter() .filter(|e| match e.score { @@ -40,7 +43,11 @@ pub(crate) async fn build_context( // facts can influence the turn in a controlled way. let working_query = format!("working.user {user_msg}"); if let Ok(entries) = mem - .recall(&working_query, WORKING_MEMORY_LIMIT + 2, None) + .recall( + &working_query, + WORKING_MEMORY_LIMIT + 2, + crate::openhuman::memory::RecallOpts::default(), + ) .await { let working: Vec<_> = entries @@ -86,6 +93,7 @@ mod tests { async fn store( &self, + _namespace: &str, _key: &str, _content: &str, _category: MemoryCategory, @@ -98,7 +106,7 @@ mod tests { &self, query: &str, _limit: usize, - _session_id: Option<&str>, + _opts: crate::openhuman::memory::RecallOpts<'_>, ) -> anyhow::Result> { if query.starts_with("working.user ") { return Ok(self.working.clone()); @@ -109,22 +117,29 @@ mod tests { Ok(self.primary.clone()) } - async fn get(&self, _key: &str) -> anyhow::Result> { + async fn get(&self, _namespace: &str, _key: &str) -> anyhow::Result> { Ok(None) } async fn list( &self, + _namespace: Option<&str>, _category: Option<&MemoryCategory>, _session_id: Option<&str>, ) -> anyhow::Result> { Ok(Vec::new()) } - async fn forget(&self, _key: &str) -> anyhow::Result { + async fn forget(&self, _namespace: &str, _key: &str) -> anyhow::Result { Ok(false) } + async fn namespace_summaries( + &self, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn count(&self) -> anyhow::Result { Ok(0) } diff --git a/src/openhuman/agent/harness/session/builder.rs b/src/openhuman/agent/harness/session/builder.rs index d9ef8dde9..505221077 100644 --- a/src/openhuman/agent/harness/session/builder.rs +++ b/src/openhuman/agent/harness/session/builder.rs @@ -367,6 +367,7 @@ impl AgentBuilder { skills: self.skills.unwrap_or_default(), auto_save: self.auto_save.unwrap_or(false), last_memory_context: None, + last_turn_citations: Vec::new(), history: Vec::new(), post_turn_hooks: self.post_turn_hooks, learning_enabled: self.learning_enabled, diff --git a/src/openhuman/agent/harness/session/runtime.rs b/src/openhuman/agent/harness/session/runtime.rs index 6be3d5fc0..32a8cd1b4 100644 --- a/src/openhuman/agent/harness/session/runtime.rs +++ b/src/openhuman/agent/harness/session/runtime.rs @@ -187,6 +187,13 @@ impl Agent { self.history.clear(); } + /// Drain and return memory citations collected for the latest completed turn. + pub fn take_last_turn_citations( + &mut self, + ) -> Vec { + std::mem::take(&mut self.last_turn_citations) + } + // ───────────────────────────────────────────────────────────────── // Static helpers for turn parsing + telemetry // ───────────────────────────────────────────────────────────────── diff --git a/src/openhuman/agent/harness/session/turn.rs b/src/openhuman/agent/harness/session/turn.rs index 3bab430df..7d74ec257 100644 --- a/src/openhuman/agent/harness/session/turn.rs +++ b/src/openhuman/agent/harness/session/turn.rs @@ -25,6 +25,7 @@ use crate::core::event_bus::{publish_global, DomainEvent}; use crate::openhuman::agent::dispatcher::{ParsedToolCall, ToolExecutionResult}; use crate::openhuman::agent::harness; use crate::openhuman::agent::hooks::{self, ToolCallRecord, TurnContext}; +use crate::openhuman::agent::memory_loader::collect_recall_citations; use crate::openhuman::agent::progress::AgentProgress; use crate::openhuman::context::prompt::{LearnedContextData, PromptContext, PromptTool}; use crate::openhuman::context::{ReductionOutcome, ARCHIVIST_EXTRACTION_PROMPT}; @@ -130,11 +131,39 @@ impl Agent { if self.auto_save { let _ = self .memory - .store("user_msg", user_message, MemoryCategory::Conversation, None) + .store( + "", + "user_msg", + user_message, + MemoryCategory::Conversation, + None, + ) .await; } log::info!("[agent] loading memory context for user message"); + const MEMORY_CITATION_LIMIT: usize = 5; + const MEMORY_CITATION_MIN_RELEVANCE: f64 = 0.4; + match collect_recall_citations( + self.memory.as_ref(), + user_message, + MEMORY_CITATION_LIMIT, + MEMORY_CITATION_MIN_RELEVANCE, + ) + .await + { + Ok(citations) => { + log::debug!( + "[agent_loop] memory citations collected count={}", + citations.len() + ); + self.last_turn_citations = citations; + } + Err(err) => { + log::warn!("[agent_loop] memory citation collection failed: {err}"); + self.last_turn_citations.clear(); + } + } let context = self .memory_loader .load_context(self.memory.as_ref(), user_message) @@ -505,7 +534,7 @@ impl Agent { let summary = truncate_with_ellipsis(&final_text, 100); let _ = self .memory - .store("assistant_resp", &summary, MemoryCategory::Daily, None) + .store("", "assistant_resp", &summary, MemoryCategory::Daily, None) .await; } @@ -1053,6 +1082,7 @@ impl Agent { let obs_entries = self .memory .list( + Some("learning_observations"), Some(&MemoryCategory::Custom("learning_observations".into())), None, ) @@ -1062,6 +1092,7 @@ impl Agent { let pat_entries = self .memory .list( + Some("learning_patterns"), Some(&MemoryCategory::Custom("learning_patterns".into())), None, ) @@ -1070,7 +1101,11 @@ impl Agent { let profile_entries = self .memory - .list(Some(&MemoryCategory::Custom("user_profile".into())), None) + .list( + Some("user_profile"), + Some(&MemoryCategory::Custom("user_profile".into())), + None, + ) .await .unwrap_or_default(); diff --git a/src/openhuman/agent/harness/session/types.rs b/src/openhuman/agent/harness/session/types.rs index 859df87a2..3ffda41fe 100644 --- a/src/openhuman/agent/harness/session/types.rs +++ b/src/openhuman/agent/harness/session/types.rs @@ -52,6 +52,9 @@ pub struct Agent { /// Last memory context loaded for the current turn. Stored so it can /// be forwarded to subagents via `ParentExecutionContext`. pub(super) last_memory_context: Option, + /// Citation metadata collected from memory recall for the most recent turn. + /// Consumed by web-channel delivery to render source chips in the UI. + pub(super) last_turn_citations: Vec, pub(super) history: Vec, pub(super) post_turn_hooks: Vec>, pub(super) learning_enabled: bool, diff --git a/src/openhuman/agent/harness/subagent_runner/ops.rs b/src/openhuman/agent/harness/subagent_runner/ops.rs index 2acd20c8c..ef22edb3f 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops.rs +++ b/src/openhuman/agent/harness/subagent_runner/ops.rs @@ -1385,6 +1385,7 @@ mod tests { impl crate::openhuman::memory::Memory for NoopMemory { async fn store( &self, + _namespace: &str, _key: &str, _content: &str, _category: crate::openhuman::memory::MemoryCategory, @@ -1396,26 +1397,33 @@ mod tests { &self, _query: &str, _limit: usize, - _session_id: Option<&str>, + _opts: crate::openhuman::memory::RecallOpts<'_>, ) -> anyhow::Result> { Ok(vec![]) } async fn get( &self, + _namespace: &str, _key: &str, ) -> anyhow::Result> { Ok(None) } async fn list( &self, + _namespace: Option<&str>, _category: Option<&crate::openhuman::memory::MemoryCategory>, _session_id: Option<&str>, ) -> anyhow::Result> { Ok(vec![]) } - async fn forget(&self, _key: &str) -> anyhow::Result { + async fn forget(&self, _namespace: &str, _key: &str) -> anyhow::Result { Ok(true) } + async fn namespace_summaries( + &self, + ) -> anyhow::Result> { + Ok(vec![]) + } async fn count(&self) -> anyhow::Result { Ok(0) } diff --git a/src/openhuman/agent/memory_loader.rs b/src/openhuman/agent/memory_loader.rs index 0d10e1840..a6c99effd 100644 --- a/src/openhuman/agent/memory_loader.rs +++ b/src/openhuman/agent/memory_loader.rs @@ -1,5 +1,6 @@ use crate::openhuman::memory::Memory; use async_trait::async_trait; +use serde::{Deserialize, Serialize}; use std::collections::HashSet; use super::harness::memory_context::{WORKING_MEMORY_KEY_PREFIX, WORKING_MEMORY_LIMIT}; @@ -17,6 +18,22 @@ pub struct DefaultMemoryLoader { max_context_chars: usize, } +/// Lightweight citation object derived from recalled memory entries. +/// +/// These citations are attached to agent responses so the UI can show +/// provenance for memory-informed answers without exposing full raw memory. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct MemoryCitation { + pub id: String, + pub key: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub namespace: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub score: Option, + pub timestamp: String, + pub snippet: String, +} + impl Default for DefaultMemoryLoader { fn default() -> Self { Self { @@ -42,6 +59,50 @@ impl DefaultMemoryLoader { } } +/// Collect citation metadata from semantic memory recall for a user turn. +/// +/// This mirrors the primary recall path used by `DefaultMemoryLoader` so the +/// UI can display trusted sources whenever memory context influenced a reply. +pub async fn collect_recall_citations( + memory: &dyn Memory, + user_message: &str, + limit: usize, + min_relevance_score: f64, +) -> anyhow::Result> { + let entries = memory + .recall( + user_message, + limit.max(1), + crate::openhuman::memory::RecallOpts::default(), + ) + .await?; + + let citations = entries + .into_iter() + .filter(|entry| match entry.score { + Some(score) => score >= min_relevance_score, + None => true, + }) + .map(|entry| { + let snippet = if entry.content.chars().count() > 280 { + crate::openhuman::util::truncate_with_ellipsis(&entry.content, 280) + } else { + entry.content + }; + MemoryCitation { + id: entry.id, + key: entry.key, + namespace: entry.namespace, + score: entry.score, + timestamp: entry.timestamp, + snippet, + } + }) + .collect(); + + Ok(citations) +} + #[async_trait] impl MemoryLoader for DefaultMemoryLoader { async fn load_context( @@ -49,7 +110,13 @@ impl MemoryLoader for DefaultMemoryLoader { memory: &dyn Memory, user_message: &str, ) -> anyhow::Result { - let entries = memory.recall(user_message, self.limit, None).await?; + let entries = memory + .recall( + user_message, + self.limit, + crate::openhuman::memory::RecallOpts::default(), + ) + .await?; let mut context = String::new(); let budget = if self.max_context_chars > 0 { self.max_context_chars @@ -88,7 +155,11 @@ impl MemoryLoader for DefaultMemoryLoader { // Explicit bounded recall for sync-derived user working memory. let working_query = format!("working.user {user_message}"); let working_entries = memory - .recall(&working_query, WORKING_MEMORY_LIMIT + 2, None) + .recall( + &working_query, + WORKING_MEMORY_LIMIT + 2, + crate::openhuman::memory::RecallOpts::default(), + ) .await .unwrap_or_default(); let mut appended_working_header = false; @@ -130,3 +201,103 @@ impl MemoryLoader for DefaultMemoryLoader { Ok(context) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::memory::{Memory, MemoryCategory, MemoryEntry}; + + struct MockMemory { + entries: Vec, + } + + #[async_trait] + impl Memory for MockMemory { + fn name(&self) -> &str { + "mock" + } + + async fn store( + &self, + _namespace: &str, + _key: &str, + _content: &str, + _category: MemoryCategory, + _session_id: Option<&str>, + ) -> anyhow::Result<()> { + Ok(()) + } + + async fn recall( + &self, + _query: &str, + _limit: usize, + _opts: crate::openhuman::memory::RecallOpts<'_>, + ) -> anyhow::Result> { + Ok(self.entries.clone()) + } + + async fn get(&self, _namespace: &str, _key: &str) -> anyhow::Result> { + Ok(None) + } + + async fn list( + &self, + _namespace: Option<&str>, + _category: Option<&MemoryCategory>, + _session_id: Option<&str>, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + + async fn forget(&self, _namespace: &str, _key: &str) -> anyhow::Result { + Ok(false) + } + + async fn namespace_summaries( + &self, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + + async fn count(&self) -> anyhow::Result { + Ok(self.entries.len()) + } + + async fn health_check(&self) -> bool { + true + } + } + + fn entry(key: &str, content: &str, score: Option) -> MemoryEntry { + MemoryEntry { + id: format!("id-{key}"), + key: key.to_string(), + content: content.to_string(), + namespace: Some("test".to_string()), + category: MemoryCategory::Conversation, + timestamp: "2026-04-22T00:00:00Z".to_string(), + session_id: None, + score, + } + } + + #[tokio::test] + async fn collect_recall_citations_filters_and_truncates_entries() { + let mem = MockMemory { + entries: vec![ + entry("keep", "useful context", Some(0.9)), + entry("drop", "too weak", Some(0.1)), + entry("long", &"x".repeat(600), Some(0.8)), + ], + }; + + let citations = collect_recall_citations(&mem, "hello", 5, 0.4) + .await + .expect("citation collection should succeed"); + assert_eq!(citations.len(), 2); + assert_eq!(citations[0].key, "keep"); + assert_eq!(citations[1].key, "long"); + assert!(citations[1].snippet.ends_with("...")); + } +} diff --git a/src/openhuman/channels/context.rs b/src/openhuman/channels/context.rs index ef2915b8f..c6c5a8e2a 100644 --- a/src/openhuman/channels/context.rs +++ b/src/openhuman/channels/context.rs @@ -154,7 +154,10 @@ pub(crate) async fn build_memory_context( ) -> String { let mut context = String::new(); - if let Ok(entries) = mem.recall(user_msg, 5, None).await { + if let Ok(entries) = mem + .recall(user_msg, 5, crate::openhuman::memory::RecallOpts::default()) + .await + { let mut included = 0usize; let mut used_chars = 0usize; @@ -256,6 +259,7 @@ mod tests { async fn store( &self, + _namespace: &str, _key: &str, _content: &str, _category: MemoryCategory, @@ -268,27 +272,34 @@ mod tests { &self, _query: &str, _limit: usize, - _session_id: Option<&str>, + _opts: crate::openhuman::memory::RecallOpts<'_>, ) -> anyhow::Result> { Ok(self.entries.clone()) } - async fn get(&self, _key: &str) -> anyhow::Result> { + async fn get(&self, _namespace: &str, _key: &str) -> anyhow::Result> { Ok(None) } async fn list( &self, + _namespace: Option<&str>, _category: Option<&MemoryCategory>, _session_id: Option<&str>, ) -> anyhow::Result> { Ok(Vec::new()) } - async fn forget(&self, _key: &str) -> anyhow::Result { + async fn forget(&self, _namespace: &str, _key: &str) -> anyhow::Result { Ok(false) } + async fn namespace_summaries( + &self, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn count(&self) -> anyhow::Result { Ok(self.entries.len()) } diff --git a/src/openhuman/channels/proactive.rs b/src/openhuman/channels/proactive.rs index 320ec81f8..b6c2872d3 100644 --- a/src/openhuman/channels/proactive.rs +++ b/src/openhuman/channels/proactive.rs @@ -138,6 +138,7 @@ impl EventHandler for ProactiveMessageSubscriber { delta: None, delta_kind: None, tool_call_id: None, + citations: None, }); // 2. If an active external channel is configured, deliver there too. diff --git a/src/openhuman/channels/providers/presentation.rs b/src/openhuman/channels/providers/presentation.rs index 9abfdbb4a..a6e7dba99 100644 --- a/src/openhuman/channels/providers/presentation.rs +++ b/src/openhuman/channels/providers/presentation.rs @@ -29,6 +29,7 @@ pub async fn deliver_response( request_id: &str, full_response: &str, user_message: &str, + citations: &[crate::openhuman::agent::memory_loader::MemoryCitation], ) { // Spawn reaction decision in parallel — it runs on the local model and // shouldn't block segmentation or delivery. @@ -63,6 +64,11 @@ pub async fn deliver_response( delta: None, delta_kind: None, tool_call_id: None, + citations: if citations.is_empty() { + None + } else { + Some(serde_json::json!(citations)) + }, }); return; } @@ -97,6 +103,11 @@ pub async fn deliver_response( delta: None, delta_kind: None, tool_call_id: None, + citations: if i == 0 && !citations.is_empty() { + Some(serde_json::json!(citations)) + } else { + None + }, }); } @@ -121,6 +132,11 @@ pub async fn deliver_response( delta: None, delta_kind: None, tool_call_id: None, + citations: if citations.is_empty() { + None + } else { + Some(serde_json::json!(citations)) + }, }); } diff --git a/src/openhuman/channels/providers/web.rs b/src/openhuman/channels/providers/web.rs index 480c1f09f..974f8bae9 100644 --- a/src/openhuman/channels/providers/web.rs +++ b/src/openhuman/channels/providers/web.rs @@ -71,6 +71,12 @@ struct InFlightEntry { handle: tokio::task::JoinHandle<()>, } +#[derive(Debug, Clone)] +struct WebChatTaskResult { + full_response: String, + citations: Vec, +} + static THREAD_SESSIONS: Lazy>> = Lazy::new(|| Mutex::new(HashMap::new())); @@ -161,6 +167,7 @@ pub async fn start_chat( delta: None, delta_kind: None, tool_call_id: None, + citations: None, }); } } @@ -183,7 +190,7 @@ pub async fn start_chat( .await; match result { - Ok(full_response) => { + Ok(chat_result) => { // ── Presentation layer (local model, fire-and-forget) ───── // Segment the response into human-readable bubbles and // decide whether to react — both run via local Ollama if @@ -192,8 +199,9 @@ pub async fn start_chat( &client_id_task, &thread_id_task, &request_id_task, - &full_response, + &chat_result.full_response, &user_message, + &chat_result.citations, ) .await; } @@ -225,6 +233,7 @@ pub async fn start_chat( delta: None, delta_kind: None, tool_call_id: None, + citations: None, }); } } @@ -316,6 +325,7 @@ pub async fn cancel_chat(client_id: &str, thread_id: &str) -> Result, temperature: Option, -) -> Result { +) -> Result { let config = config_rpc::load_config_with_timeout().await?; let map_key = key_for(client_id, thread_id); let model_override = normalize_model_override(model_override); @@ -399,7 +409,13 @@ async fn run_chat_task( ); let result = match agent.run_single(message).await { - Ok(response) => Ok(response), + Ok(response) => { + let citations = agent.take_last_turn_citations(); + Ok(WebChatTaskResult { + full_response: response, + citations, + }) + } Err(err) => { let err_message = err.to_string(); if is_inference_budget_exceeded_error(&err_message) { @@ -409,7 +425,10 @@ async fn run_chat_task( thread_id, request_id ); - Ok(inference_budget_exceeded_user_message().to_string()) + Ok(WebChatTaskResult { + full_response: inference_budget_exceeded_user_message().to_string(), + citations: Vec::new(), + }) } else { Err(err_message) } @@ -566,6 +585,7 @@ fn spawn_progress_bridge( delta: None, delta_kind: None, tool_call_id: None, + citations: None, }); } AgentProgress::IterationStarted { @@ -593,6 +613,7 @@ fn spawn_progress_bridge( delta: None, delta_kind: None, tool_call_id: None, + citations: None, }); } AgentProgress::ToolCallStarted { @@ -660,6 +681,7 @@ fn spawn_progress_bridge( delta: None, delta_kind: None, tool_call_id: None, + citations: None, }); } AgentProgress::SubagentCompleted { @@ -689,6 +711,7 @@ fn spawn_progress_bridge( delta: None, delta_kind: None, tool_call_id: None, + citations: None, }); } AgentProgress::SubagentFailed { @@ -716,6 +739,7 @@ fn spawn_progress_bridge( delta: None, delta_kind: None, tool_call_id: None, + citations: None, }); } AgentProgress::TextDelta { delta, iteration } => { diff --git a/src/openhuman/channels/routes.rs b/src/openhuman/channels/routes.rs index c8117a306..4ed684ee6 100644 --- a/src/openhuman/channels/routes.rs +++ b/src/openhuman/channels/routes.rs @@ -368,6 +368,7 @@ mod tests { async fn store( &self, + _namespace: &str, _key: &str, _content: &str, _category: MemoryCategory, @@ -380,27 +381,34 @@ mod tests { &self, _query: &str, _limit: usize, - _session_id: Option<&str>, + _opts: crate::openhuman::memory::RecallOpts<'_>, ) -> anyhow::Result> { Ok(Vec::new()) } - async fn get(&self, _key: &str) -> anyhow::Result> { + async fn get(&self, _namespace: &str, _key: &str) -> anyhow::Result> { Ok(None) } async fn list( &self, + _namespace: Option<&str>, _category: Option<&MemoryCategory>, _session_id: Option<&str>, ) -> anyhow::Result> { Ok(Vec::new()) } - async fn forget(&self, _key: &str) -> anyhow::Result { + async fn forget(&self, _namespace: &str, _key: &str) -> anyhow::Result { Ok(false) } + async fn namespace_summaries( + &self, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn count(&self) -> anyhow::Result { Ok(0) } diff --git a/src/openhuman/channels/runtime/dispatch.rs b/src/openhuman/channels/runtime/dispatch.rs index 26b7eba7a..b60d91b9b 100644 --- a/src/openhuman/channels/runtime/dispatch.rs +++ b/src/openhuman/channels/runtime/dispatch.rs @@ -752,6 +752,7 @@ pub(crate) async fn process_channel_message( let _ = ctx .memory .store( + "", &autosave_key, &msg.content, crate::openhuman::memory::MemoryCategory::Conversation, diff --git a/src/openhuman/channels/tests/common.rs b/src/openhuman/channels/tests/common.rs index 144d7e64d..2adb0b71d 100644 --- a/src/openhuman/channels/tests/common.rs +++ b/src/openhuman/channels/tests/common.rs @@ -376,6 +376,7 @@ impl Memory for NoopMemory { async fn store( &self, + _namespace: &str, _key: &str, _content: &str, _category: MemoryCategory, @@ -388,27 +389,34 @@ impl Memory for NoopMemory { &self, _query: &str, _limit: usize, - _session_id: Option<&str>, + _opts: crate::openhuman::memory::RecallOpts<'_>, ) -> anyhow::Result> { Ok(Vec::new()) } - async fn get(&self, _key: &str) -> anyhow::Result> { + async fn get(&self, _namespace: &str, _key: &str) -> anyhow::Result> { Ok(None) } async fn list( &self, + _namespace: Option<&str>, _category: Option<&MemoryCategory>, _session_id: Option<&str>, ) -> anyhow::Result> { Ok(Vec::new()) } - async fn forget(&self, _key: &str) -> anyhow::Result { + async fn forget(&self, _namespace: &str, _key: &str) -> anyhow::Result { Ok(false) } + async fn namespace_summaries( + &self, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn count(&self) -> anyhow::Result { Ok(0) } diff --git a/src/openhuman/channels/tests/memory.rs b/src/openhuman/channels/tests/memory.rs index d7ee8a68c..ec17c57bb 100644 --- a/src/openhuman/channels/tests/memory.rs +++ b/src/openhuman/channels/tests/memory.rs @@ -77,6 +77,7 @@ async fn autosave_keys_preserve_multiple_conversation_facts() { }; mem.store( + "", &conversation_memory_key(&msg1), &msg1.content, MemoryCategory::Conversation, @@ -85,6 +86,7 @@ async fn autosave_keys_preserve_multiple_conversation_facts() { .await .unwrap(); mem.store( + "", &conversation_memory_key(&msg2), &msg2.content, MemoryCategory::Conversation, @@ -95,7 +97,10 @@ async fn autosave_keys_preserve_multiple_conversation_facts() { assert_eq!(mem.count().await.unwrap(), 2); - let recalled = mem.recall("45", 5, None).await.unwrap(); + let recalled = mem + .recall("45", 5, crate::openhuman::memory::RecallOpts::default()) + .await + .unwrap(); assert!(recalled.iter().any(|entry| entry.content.contains("45"))); } @@ -103,9 +108,15 @@ async fn autosave_keys_preserve_multiple_conversation_facts() { async fn build_memory_context_includes_recalled_entries() { let tmp = TempDir::new().unwrap(); let mem = UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap(); - mem.store("age_fact", "Age is 45", MemoryCategory::Conversation, None) - .await - .unwrap(); + mem.store( + "", + "age_fact", + "Age is 45", + MemoryCategory::Conversation, + None, + ) + .await + .unwrap(); let context = build_memory_context(&mem, "age", 0.0).await; assert!(context.contains("[Memory context]")); diff --git a/src/openhuman/cost/tracker.rs b/src/openhuman/cost/tracker.rs index 191642bad..44ca0f66d 100644 --- a/src/openhuman/cost/tracker.rs +++ b/src/openhuman/cost/tracker.rs @@ -593,7 +593,7 @@ mod tests { let tracker = CostTracker::new(config, tmp.path()).unwrap(); // Record usage just under warning threshold (80% of 10 = 8.0) - let usage = TokenUsage::new("test/model", 100000, 50000, 1.0, 2.0); + let _usage = TokenUsage::new("test/model", 100000, 50000, 1.0, 2.0); // This has a cost, so let's just check the budget with a projected amount let check = tracker.check_budget(8.5).unwrap(); assert!( diff --git a/src/openhuman/learning/prompt_sections.rs b/src/openhuman/learning/prompt_sections.rs index 8a7fa673f..8789a4c5e 100644 --- a/src/openhuman/learning/prompt_sections.rs +++ b/src/openhuman/learning/prompt_sections.rs @@ -102,6 +102,7 @@ mod tests { async fn store( &self, + _namespace: &str, _key: &str, _content: &str, _category: MemoryCategory, @@ -114,27 +115,34 @@ mod tests { &self, _query: &str, _limit: usize, - _session_id: Option<&str>, + _opts: crate::openhuman::memory::RecallOpts<'_>, ) -> anyhow::Result> { Ok(Vec::new()) } - async fn get(&self, _key: &str) -> anyhow::Result> { + async fn get(&self, _namespace: &str, _key: &str) -> anyhow::Result> { Ok(None) } async fn list( &self, + _namespace: Option<&str>, _category: Option<&MemoryCategory>, _session_id: Option<&str>, ) -> anyhow::Result> { Ok(Vec::new()) } - async fn forget(&self, _key: &str) -> anyhow::Result { + async fn forget(&self, _namespace: &str, _key: &str) -> anyhow::Result { Ok(false) } + async fn namespace_summaries( + &self, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn count(&self) -> anyhow::Result { Ok(0) } diff --git a/src/openhuman/learning/reflection.rs b/src/openhuman/learning/reflection.rs index 9d8c1ad97..7deb76ff3 100644 --- a/src/openhuman/learning/reflection.rs +++ b/src/openhuman/learning/reflection.rs @@ -188,6 +188,7 @@ impl ReflectionHook { let key = format!("obs/{date}/{hash}"); self.memory .store( + "learning_observations", &key, &content, MemoryCategory::Custom("learning_observations".into()), @@ -205,6 +206,7 @@ impl ReflectionHook { let key = format!("pat/{slug}"); self.memory .store( + "learning_patterns", &key, pattern, MemoryCategory::Custom("learning_patterns".into()), @@ -219,6 +221,7 @@ impl ReflectionHook { let key = format!("pref/{slug}"); self.memory .store( + "user_profile", &key, pref, MemoryCategory::Custom("user_profile".into()), @@ -326,6 +329,7 @@ mod tests { async fn store( &self, + namespace: &str, key: &str, content: &str, category: MemoryCategory, @@ -337,7 +341,7 @@ mod tests { id: key.to_string(), key: key.to_string(), content: content.to_string(), - namespace: None, + namespace: Some(namespace.to_string()), category, timestamp: "now".into(), session_id: session_id.map(str::to_string), @@ -351,27 +355,34 @@ mod tests { &self, _query: &str, _limit: usize, - _session_id: Option<&str>, + _opts: crate::openhuman::memory::RecallOpts<'_>, ) -> anyhow::Result> { Ok(Vec::new()) } - async fn get(&self, key: &str) -> anyhow::Result> { + async fn get(&self, _namespace: &str, key: &str) -> anyhow::Result> { Ok(self.entries.lock().get(key).cloned()) } async fn list( &self, + _namespace: Option<&str>, _category: Option<&MemoryCategory>, _session_id: Option<&str>, ) -> anyhow::Result> { Ok(self.entries.lock().values().cloned().collect()) } - async fn forget(&self, key: &str) -> anyhow::Result { + async fn forget(&self, _namespace: &str, key: &str) -> anyhow::Result { Ok(self.entries.lock().remove(key).is_some()) } + async fn namespace_summaries( + &self, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn count(&self) -> anyhow::Result { Ok(self.entries.lock().len()) } diff --git a/src/openhuman/learning/tool_tracker.rs b/src/openhuman/learning/tool_tracker.rs index 460076781..48bf4a00d 100644 --- a/src/openhuman/learning/tool_tracker.rs +++ b/src/openhuman/learning/tool_tracker.rs @@ -113,7 +113,7 @@ impl ToolTrackerHook { let _guard = lock.lock().await; let key = format!("tool/{tool_name}"); - let mut stats: ToolStats = match self.memory.get(&key).await { + let mut stats: ToolStats = match self.memory.get("tool_effectiveness", &key).await { Ok(Some(entry)) => serde_json::from_str(&entry.content).unwrap_or_default(), _ => ToolStats::default(), }; @@ -123,6 +123,7 @@ impl ToolTrackerHook { let content = serde_json::to_string(&stats)?; self.memory .store( + "tool_effectiveness", &key, &content, MemoryCategory::Custom("tool_effectiveness".into()), @@ -198,6 +199,7 @@ mod tests { async fn store( &self, + namespace: &str, key: &str, content: &str, category: MemoryCategory, @@ -209,7 +211,7 @@ mod tests { id: key.to_string(), key: key.to_string(), content: content.to_string(), - namespace: None, + namespace: Some(namespace.to_string()), category, timestamp: "now".into(), session_id: session_id.map(str::to_string), @@ -223,27 +225,34 @@ mod tests { &self, _query: &str, _limit: usize, - _session_id: Option<&str>, + _opts: crate::openhuman::memory::RecallOpts<'_>, ) -> anyhow::Result> { Ok(Vec::new()) } - async fn get(&self, key: &str) -> anyhow::Result> { + async fn get(&self, _namespace: &str, key: &str) -> anyhow::Result> { Ok(self.entries.lock().get(key).cloned()) } async fn list( &self, + _namespace: Option<&str>, _category: Option<&MemoryCategory>, _session_id: Option<&str>, ) -> anyhow::Result> { Ok(self.entries.lock().values().cloned().collect()) } - async fn forget(&self, key: &str) -> anyhow::Result { + async fn forget(&self, _namespace: &str, key: &str) -> anyhow::Result { Ok(self.entries.lock().remove(key).is_some()) } + async fn namespace_summaries( + &self, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn count(&self) -> anyhow::Result { Ok(self.entries.lock().len()) } @@ -306,6 +315,7 @@ mod tests { let memory_impl = Arc::new(MockMemory::default()); memory_impl .store( + "tool_effectiveness", "tool/shell", &serde_json::to_string(&ToolStats { total_calls: 2, @@ -333,7 +343,11 @@ mod tests { hook.update_stats("shell", true, 250, None).await.unwrap(); - let stored = memory_impl.get("tool/shell").await.unwrap().unwrap(); + let stored = memory_impl + .get("tool_effectiveness", "tool/shell") + .await + .unwrap() + .unwrap(); let parsed: ToolStats = serde_json::from_str(&stored.content).unwrap(); assert_eq!(parsed.total_calls, 3); assert_eq!(parsed.successes, 2); @@ -397,7 +411,11 @@ mod tests { hook.on_turn_complete(&ctx).await.unwrap(); - let stored = memory_impl.get("tool/shell").await.unwrap().unwrap(); + let stored = memory_impl + .get("tool_effectiveness", "tool/shell") + .await + .unwrap() + .unwrap(); let parsed: ToolStats = serde_json::from_str(&stored.content).unwrap(); assert_eq!(parsed.total_calls, 2); assert_eq!(parsed.successes, 1); diff --git a/src/openhuman/learning/user_profile.rs b/src/openhuman/learning/user_profile.rs index 8f5ca2e0d..28dee2a14 100644 --- a/src/openhuman/learning/user_profile.rs +++ b/src/openhuman/learning/user_profile.rs @@ -87,13 +87,14 @@ impl UserProfileHook { let key = format!("pref/{slug}"); // Check for existing entry to avoid duplicates - if let Ok(Some(_)) = self.memory.get(&key).await { + if let Ok(Some(_)) = self.memory.get("user_profile", &key).await { log::debug!("[learning] user preference already stored: {key}"); continue; } self.memory .store( + "user_profile", &key, pref, MemoryCategory::Custom("user_profile".into()), @@ -168,6 +169,7 @@ mod tests { async fn store( &self, + namespace: &str, key: &str, content: &str, category: MemoryCategory, @@ -179,7 +181,7 @@ mod tests { id: key.to_string(), key: key.to_string(), content: content.to_string(), - namespace: None, + namespace: Some(namespace.to_string()), category, timestamp: "now".into(), session_id: session_id.map(str::to_string), @@ -193,27 +195,34 @@ mod tests { &self, _query: &str, _limit: usize, - _session_id: Option<&str>, + _opts: crate::openhuman::memory::RecallOpts<'_>, ) -> anyhow::Result> { Ok(Vec::new()) } - async fn get(&self, key: &str) -> anyhow::Result> { + async fn get(&self, _namespace: &str, key: &str) -> anyhow::Result> { Ok(self.entries.lock().get(key).cloned()) } async fn list( &self, + _namespace: Option<&str>, _category: Option<&MemoryCategory>, _session_id: Option<&str>, ) -> anyhow::Result> { Ok(self.entries.lock().values().cloned().collect()) } - async fn forget(&self, key: &str) -> anyhow::Result { + async fn forget(&self, _namespace: &str, key: &str) -> anyhow::Result { Ok(self.entries.lock().remove(key).is_some()) } + async fn namespace_summaries( + &self, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + async fn count(&self) -> anyhow::Result { Ok(self.entries.lock().len()) } @@ -265,6 +274,7 @@ mod tests { let memory_impl = Arc::new(MockMemory::default()); memory_impl .store( + "user_profile", "pref/i_prefer_rust", "I prefer Rust", MemoryCategory::Custom("user_profile".into()), diff --git a/src/openhuman/memory/mod.rs b/src/openhuman/memory/mod.rs index 52bab6705..08869ea5f 100644 --- a/src/openhuman/memory/mod.rs +++ b/src/openhuman/memory/mod.rs @@ -36,5 +36,5 @@ pub use store::{ MemoryClientRef, MemoryItemKind, MemoryState, NamespaceDocumentInput, NamespaceMemoryHit, NamespaceQueryResult, NamespaceRetrievalContext, RetrievalScoreBreakdown, UnifiedMemory, }; -pub use traits::{Memory, MemoryCategory, MemoryEntry}; +pub use traits::{Memory, MemoryCategory, MemoryEntry, NamespaceSummary, RecallOpts}; pub use tree::{all_memory_tree_controller_schemas, all_memory_tree_registered_controllers}; diff --git a/src/openhuman/memory/store/memory_trait.rs b/src/openhuman/memory/store/memory_trait.rs index 1ad88fa97..cf139798a 100644 --- a/src/openhuman/memory/store/memory_trait.rs +++ b/src/openhuman/memory/store/memory_trait.rs @@ -4,10 +4,10 @@ //! struct. This allows `UnifiedMemory` to be used as a generic memory backend //! within the OpenHuman system. //! -//! It handles the mapping between the generic memory interface (which uses -//! keys and categories) and the unified namespace-based storage (which uses -//! documents and namespaces). It primarily uses the `GLOBAL_NAMESPACE` for -//! these operations. +//! Callers pass an explicit `namespace` on `store`/`get`/`forget` and via +//! `RecallOpts` on `recall`. When a `namespace` is omitted on `recall`/`list`, +//! the implementation falls back to `GLOBAL_NAMESPACE` (legacy behavior), which +//! Phase B/C will tighten once the memory tools pass namespace explicitly. use async_trait::async_trait; use chrono::{TimeZone, Utc}; @@ -16,15 +16,14 @@ use serde_json::json; use crate::openhuman::memory::store::types::{NamespaceDocumentInput, GLOBAL_NAMESPACE}; use crate::openhuman::memory::store::unified::fts5; -use crate::openhuman::memory::traits::{Memory, MemoryCategory, MemoryEntry}; +use crate::openhuman::memory::traits::{ + Memory, MemoryCategory, MemoryEntry, NamespaceSummary, RecallOpts, +}; use anyhow::Context; use super::unified::UnifiedMemory; /// Convert a UNIX timestamp (f64) to RFC3339 string. -/// -/// Handles fractional seconds and ensures the result is a valid timestamp. -/// If conversion fails, it returns a string representation of the raw number. fn timestamp_to_rfc3339(ts: f64) -> String { let secs = ts.trunc() as i64; let nanos = ((ts.fract()) * 1_000_000_000.0).round() as u32; @@ -34,6 +33,17 @@ fn timestamp_to_rfc3339(ts: f64) -> String { .unwrap_or_else(|| format!("{ts}")) } +/// Normalize a namespace value: trim whitespace and fall back to +/// `GLOBAL_NAMESPACE` for `None` or blank/whitespace-only inputs. This ensures +/// that `recall`/`list` calls derived from user or RPC input never silently +/// receive an empty string that misses the global namespace. +fn normalize_namespace(namespace: Option<&str>) -> &str { + namespace + .map(str::trim) + .filter(|ns| !ns.is_empty()) + .unwrap_or(GLOBAL_NAMESPACE) +} + /// Helper to convert a raw string category from the database into a `MemoryCategory`. fn memory_category_from_stored(raw: &str) -> MemoryCategory { match raw { @@ -46,24 +56,25 @@ fn memory_category_from_stored(raw: &str) -> MemoryCategory { #[async_trait] impl Memory for UnifiedMemory { - /// Returns the name of this memory implementation ("namespace"). fn name(&self) -> &str { "namespace" } - /// Store a piece of information in the global namespace. - /// - /// Maps the provided key and content to a standard document in the - /// `GLOBAL_NAMESPACE`. async fn store( &self, + namespace: &str, key: &str, content: &str, category: MemoryCategory, session_id: Option<&str>, ) -> anyhow::Result<()> { + let ns = if namespace.trim().is_empty() { + GLOBAL_NAMESPACE.to_string() + } else { + namespace.to_string() + }; self.upsert_document(NamespaceDocumentInput { - namespace: GLOBAL_NAMESPACE.to_string(), + namespace: ns, key: key.to_string(), title: key.to_string(), content: content.to_string(), @@ -80,30 +91,29 @@ impl Memory for UnifiedMemory { .map_err(anyhow::Error::msg) } - /// Recall relevant information based on a query string. - /// - /// Performs a ranked search in the global namespace. If a `session_id` is - /// provided, it also searches for episodic entries (conversation history) - /// and merges them with the results. async fn recall( &self, query: &str, limit: usize, - session_id: Option<&str>, + opts: RecallOpts<'_>, ) -> anyhow::Result> { - // 1. Query the global document namespace. + let namespace = normalize_namespace(opts.namespace); + let ranked = self - .query_namespace_ranked(GLOBAL_NAMESPACE, query, limit as u32) + .query_namespace_ranked(namespace, query, limit as u32) .await .map_err(anyhow::Error::msg)?; + + let min_score = opts.min_score.unwrap_or(f64::NEG_INFINITY); let mut out: Vec = ranked .into_iter() .enumerate() + .filter(|(_, r)| r.score >= min_score) .map(|(idx, r)| MemoryEntry { - id: format!("global:{idx}"), + id: format!("{namespace}:{idx}"), key: r.key, content: r.content, - namespace: Some(GLOBAL_NAMESPACE.to_string()), + namespace: Some(namespace.to_string()), category: memory_category_from_stored(&r.category), timestamp: Utc::now().to_rfc3339(), session_id: None, @@ -111,8 +121,12 @@ impl Memory for UnifiedMemory { }) .collect(); - // 2. Search episodic (chat) history if a session_id is present. - if let Some(sid) = session_id { + if let Some(ref cat) = opts.category { + let want = cat.to_string(); + out.retain(|e| e.category.to_string() == want); + } + + if let Some(sid) = opts.session_id { let episodic_entries = match fts5::episodic_session_entries(&self.conn, sid) { Ok(entries) => { tracing::debug!( @@ -129,7 +143,6 @@ impl Memory for UnifiedMemory { } }; - // Simple keyword-based filtering for episodic matches. let query_lower = query.to_lowercase(); let query_terms: Vec<&str> = query_lower.split_whitespace().collect(); for entry in episodic_entries { @@ -141,15 +154,17 @@ impl Memory for UnifiedMemory { if matched_count == 0 { continue; } - // Score based on proportion of query terms matched. let match_score = matched_count as f64 / query_terms.len().max(1) as f64; + if match_score < min_score { + continue; + } let ts_rfc3339 = timestamp_to_rfc3339(entry.timestamp); out.push(MemoryEntry { id: format!("episodic:{}", entry.id.unwrap_or(0)), key: format!("{}:{}", entry.session_id, entry.role), content: entry.content, - namespace: Some(GLOBAL_NAMESPACE.to_string()), + namespace: Some(namespace.to_string()), category: MemoryCategory::Conversation, timestamp: ts_rfc3339, session_id: Some(entry.session_id), @@ -157,7 +172,6 @@ impl Memory for UnifiedMemory { }); } - // 3. Re-sort the combined results by score. out.sort_by(|a, b| { b.score .unwrap_or(0.0) @@ -170,14 +184,18 @@ impl Memory for UnifiedMemory { Ok(out) } - /// Retrieve a specific memory entry by its key from the global namespace. - async fn get(&self, key: &str) -> anyhow::Result> { + async fn get(&self, namespace: &str, key: &str) -> anyhow::Result> { + let ns = if namespace.trim().is_empty() { + GLOBAL_NAMESPACE.to_string() + } else { + namespace.to_string() + }; let conn = self.conn.lock(); let row: Option<(String, String, String, f64, String)> = conn .query_row( "SELECT document_id, key, content, updated_at, category FROM memory_docs WHERE namespace = ?1 AND key = ?2 LIMIT 1", - params![GLOBAL_NAMESPACE, key], + params![ns, key], |row| { Ok(( row.get(0)?, @@ -194,23 +212,24 @@ impl Memory for UnifiedMemory { id, key, content, - namespace: Some(GLOBAL_NAMESPACE.to_string()), + namespace: Some(ns.clone()), category: memory_category_from_stored(&category), - timestamp: format!("{updated_at}"), + timestamp: timestamp_to_rfc3339(updated_at), session_id: None, score: None, }), ) } - /// List all memories in the global namespace, optionally filtered by category. async fn list( &self, + namespace: Option<&str>, category: Option<&MemoryCategory>, _session_id: Option<&str>, ) -> anyhow::Result> { + let ns = normalize_namespace(namespace); let docs = self - .list_documents(Some(GLOBAL_NAMESPACE)) + .list_documents(Some(ns)) .await .map_err(anyhow::Error::msg)?; let mut out = Vec::new(); @@ -237,7 +256,7 @@ impl Memory for UnifiedMemory { .and_then(serde_json::Value::as_str) .unwrap_or_default() .to_string(), - namespace: Some(GLOBAL_NAMESPACE.to_string()), + namespace: Some(ns.to_string()), category: cat, timestamp: format!("idx-{idx}"), session_id: None, @@ -247,13 +266,17 @@ impl Memory for UnifiedMemory { Ok(out) } - /// Delete a memory entry by its key from the global namespace. - async fn forget(&self, key: &str) -> anyhow::Result { + async fn forget(&self, namespace: &str, key: &str) -> anyhow::Result { + let ns = if namespace.trim().is_empty() { + GLOBAL_NAMESPACE.to_string() + } else { + namespace.to_string() + }; let row: Option = { let conn = self.conn.lock(); conn.query_row( "SELECT document_id FROM memory_docs WHERE namespace = ?1 AND key = ?2 LIMIT 1", - params![GLOBAL_NAMESPACE, key], + params![ns, key], |row| row.get(0), ) .optional()? @@ -261,25 +284,161 @@ impl Memory for UnifiedMemory { let Some(document_id) = row else { return Ok(false); }; - self.delete_document(GLOBAL_NAMESPACE, &document_id) + self.delete_document(&ns, &document_id) .await .map_err(anyhow::Error::msg)?; Ok(true) } - /// Count the total number of entries in the global namespace. + async fn namespace_summaries(&self) -> anyhow::Result> { + let conn = self.conn.lock(); + let mut stmt = conn.prepare( + "SELECT namespace, COUNT(*) AS n, MAX(updated_at) AS last + FROM memory_docs + GROUP BY namespace + ORDER BY namespace", + )?; + let rows = stmt.query_map([], |row| { + let ns: String = row.get(0)?; + let count: i64 = row.get(1)?; + let last: Option = row.get(2)?; + Ok((ns, count, last)) + })?; + let mut out = Vec::new(); + for r in rows { + let (ns, count, last) = r?; + out.push(NamespaceSummary { + namespace: ns, + count: usize::try_from(count).unwrap_or(0), + last_updated: last.map(timestamp_to_rfc3339), + }); + } + Ok(out) + } + async fn count(&self) -> anyhow::Result { let conn = self.conn.lock(); - let count: i64 = conn.query_row( - "SELECT COUNT(*) FROM memory_docs WHERE namespace = ?1", - params![GLOBAL_NAMESPACE], - |row| row.get(0), - )?; + let count: i64 = + conn.query_row("SELECT COUNT(*) FROM memory_docs", [], |row| row.get(0))?; usize::try_from(count).context("negative count") } - /// Verify the health of the memory store by checking file existence. async fn health_check(&self) -> bool { self.workspace_dir.exists() && self.db_path.exists() } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::memory::embeddings::NoopEmbedding; + use std::sync::Arc; + use tempfile::TempDir; + + fn fresh_mem() -> (TempDir, UnifiedMemory) { + let tmp = TempDir::new().unwrap(); + let mem = UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap(); + (tmp, mem) + } + + #[tokio::test] + async fn store_and_get_are_namespace_scoped() { + let (_tmp, mem) = fresh_mem(); + mem.store("ns_a", "k1", "value in a", MemoryCategory::Core, None) + .await + .unwrap(); + + let hit = mem.get("ns_a", "k1").await.unwrap(); + assert!(hit.is_some(), "same-namespace get should return entry"); + assert_eq!(hit.unwrap().content, "value in a"); + + let miss = mem.get("ns_b", "k1").await.unwrap(); + assert!(miss.is_none(), "cross-namespace get must not leak"); + } + + #[tokio::test] + async fn list_and_forget_are_namespace_scoped() { + let (_tmp, mem) = fresh_mem(); + mem.store("ns_a", "k1", "a", MemoryCategory::Core, None) + .await + .unwrap(); + mem.store("ns_b", "k1", "b", MemoryCategory::Core, None) + .await + .unwrap(); + + let in_b = mem.list(Some("ns_b"), None, None).await.unwrap(); + assert_eq!(in_b.len(), 1); + // `list` currently maps title → content (pre-Phase-A quirk preserved). + // What matters here is namespace isolation: ns_a rows must not appear. + assert!(in_b.iter().all(|e| e.namespace.as_deref() == Some("ns_b"))); + + // Forget in ns_a must not delete ns_b's row + assert!(mem.forget("ns_a", "k1").await.unwrap()); + assert!(mem.get("ns_b", "k1").await.unwrap().is_some()); + assert!(mem.get("ns_a", "k1").await.unwrap().is_none()); + } + + #[tokio::test] + async fn namespace_summaries_counts_per_namespace() { + let (_tmp, mem) = fresh_mem(); + mem.store("alpha", "k1", "x", MemoryCategory::Core, None) + .await + .unwrap(); + mem.store("alpha", "k2", "y", MemoryCategory::Core, None) + .await + .unwrap(); + mem.store("beta", "k1", "z", MemoryCategory::Core, None) + .await + .unwrap(); + + let summaries = mem.namespace_summaries().await.unwrap(); + let alpha = summaries.iter().find(|s| s.namespace == "alpha").unwrap(); + let beta = summaries.iter().find(|s| s.namespace == "beta").unwrap(); + assert_eq!(alpha.count, 2); + assert_eq!(beta.count, 1); + assert!(alpha.last_updated.is_some()); + } + + #[tokio::test] + async fn legacy_namespace_migration_splits_and_is_idempotent() { + use rusqlite::params; + + let tmp = TempDir::new().unwrap(); + let mem = UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap(); + + // Seed a legacy-shape row: GLOBAL namespace, key="ns_x/real_key". + { + let conn = mem.conn.lock(); + conn.execute( + "INSERT INTO memory_docs ( + document_id, namespace, key, title, content, source_type, + priority, tags_json, metadata_json, category, session_id, + created_at, updated_at, markdown_rel_path + ) VALUES (?1, ?2, ?3, ?4, ?5, 'chat', 'medium', '[]', '{}', 'core', NULL, 0.0, 0.0, '')", + params![ + "legacy-doc-1", + GLOBAL_NAMESPACE, + "ns_x/real_key", + "ns_x/real_key", + "legacy value" + ], + ) + .unwrap(); + } + + drop(mem); + + // Re-open so the startup migration runs again. + let mem = UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap(); + let hit = mem.get("ns_x", "real_key").await.unwrap(); + assert!(hit.is_some(), "migration should promote ns_x"); + assert_eq!(hit.unwrap().content, "legacy value"); + + // Re-open again — migration must be a no-op (no duplicate / crash). + drop(mem); + let mem = UnifiedMemory::new(tmp.path(), Arc::new(NoopEmbedding), None).unwrap(); + let still = mem.get("ns_x", "real_key").await.unwrap(); + assert!(still.is_some()); + assert_eq!(mem.count().await.unwrap(), 1); + } +} diff --git a/src/openhuman/memory/store/unified/init.rs b/src/openhuman/memory/store/unified/init.rs index e09ec012e..0d2de2986 100644 --- a/src/openhuman/memory/store/unified/init.rs +++ b/src/openhuman/memory/store/unified/init.rs @@ -118,6 +118,58 @@ impl UnifiedMemory { // User profile accumulation table. conn.execute_batch(super::profile::PROFILE_INIT_SQL)?; + // Idempotent legacy-namespace migration. + // + // Older writes via MemoryStoreTool packed the intended namespace into + // the key as `"{namespace}/{actual_key}"` and stored the row under the + // GLOBAL_NAMESPACE. Split those rows now so the new trait surface can + // rely on the `namespace` column. + // + // The anti-join guard prevents duplicate-split collisions if a + // post-split row already exists (UNIQUE(namespace, key) would otherwise + // fail). Safe to run on every boot. + let migrated = conn.execute( + "UPDATE memory_docs + SET namespace = substr(key, 1, instr(key, '/') - 1), + key = substr(key, instr(key, '/') + 1) + WHERE namespace = ?1 + AND instr(key, '/') > 0 + AND NOT EXISTS ( + SELECT 1 FROM memory_docs m2 + WHERE m2.namespace = substr(memory_docs.key, 1, instr(memory_docs.key, '/') - 1) + AND m2.key = substr(memory_docs.key, instr(memory_docs.key, '/') + 1) + )", + rusqlite::params![GLOBAL_NAMESPACE], + )?; + if migrated > 0 { + log::info!( + "[memory] migrated {migrated} legacy `ns/key` rows out of the `{GLOBAL_NAMESPACE}` namespace" + ); + } + + // Companion migration: `vector_chunks` rows keyed by `document_id` still + // point at `GLOBAL_NAMESPACE` after the `memory_docs` split above, so + // namespace-scoped recall would miss them. Re-home each chunk to its + // document's new namespace. Idempotent: after both migrations run, no + // chunk under GLOBAL_NAMESPACE maps to a document in another namespace. + let chunks_migrated = conn.execute( + "UPDATE vector_chunks + SET namespace = ( + SELECT namespace FROM memory_docs + WHERE memory_docs.document_id = vector_chunks.document_id + ) + WHERE namespace = ?1 + AND document_id IN ( + SELECT document_id FROM memory_docs WHERE namespace != ?1 + )", + rusqlite::params![GLOBAL_NAMESPACE], + )?; + if chunks_migrated > 0 { + log::info!( + "[memory] migrated {chunks_migrated} vector_chunks rows out of the `{GLOBAL_NAMESPACE}` namespace" + ); + } + Ok(Self { workspace_dir: workspace_dir.to_path_buf(), db_path, diff --git a/src/openhuman/memory/traits.rs b/src/openhuman/memory/traits.rs index fb515346a..036c17e72 100644 --- a/src/openhuman/memory/traits.rs +++ b/src/openhuman/memory/traits.rs @@ -54,6 +54,29 @@ impl std::fmt::Display for MemoryCategory { } } +/// Optional filters for `Memory::recall`. +/// +/// All fields default to `None`. `namespace = None` uses the backend's legacy +/// default namespace (`GLOBAL_NAMESPACE`). Pass `Some("namespace")` to scope +/// the semantic query to a specific namespace. +#[derive(Debug, Default, Clone)] +pub struct RecallOpts<'a> { + pub namespace: Option<&'a str>, + pub category: Option, + pub session_id: Option<&'a str>, + pub min_score: Option, +} + +/// Summary row returned by `Memory::namespace_summaries`, used for +/// agent-side namespace discovery. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct NamespaceSummary { + pub namespace: String, + pub count: usize, + /// RFC3339 timestamp of most recent `updated_at` in the namespace, if any. + pub last_updated: Option, +} + /// The core trait for memory storage and retrieval. /// /// Any persistence backend (SQLite, Postgres, Vector DB, etc.) should implement @@ -64,14 +87,9 @@ pub trait Memory: Send + Sync { fn name(&self) -> &str; /// Stores a new memory entry or updates an existing one. - /// - /// # Arguments - /// * `key` - The lookup key for the memory. - /// * `content` - The actual data to store. - /// * `category` - The organizational category. - /// * `session_id` - Optional session scope. async fn store( &self, + namespace: &str, key: &str, content: &str, category: MemoryCategory, @@ -80,31 +98,33 @@ pub trait Memory: Send + Sync { /// Recalls memories matching a query string using keyword or semantic search. /// - /// # Arguments - /// * `query` - The search term or natural language query. - /// * `limit` - Maximum number of results to return. - /// * `session_id` - Optional filter to scope search to a specific session. + /// Namespace is passed via `opts.namespace`; `None` uses the backend's + /// legacy default namespace (`GLOBAL_NAMESPACE`). async fn recall( &self, query: &str, limit: usize, - session_id: Option<&str>, + opts: RecallOpts<'_>, ) -> anyhow::Result>; - /// Retrieves a specific memory entry by its exact key. - async fn get(&self, key: &str) -> anyhow::Result>; + /// Retrieves a specific memory entry by exact (namespace, key). + async fn get(&self, namespace: &str, key: &str) -> anyhow::Result>; - /// Lists memory entries, optionally filtered by category and/or session. + /// Lists memory entries, optionally scoped by namespace, category, session. async fn list( &self, + namespace: Option<&str>, category: Option<&MemoryCategory>, session_id: Option<&str>, ) -> anyhow::Result>; - /// Deletes a memory entry associated with the given key. + /// Deletes a memory entry associated with the given (namespace, key). /// /// Returns `Ok(true)` if the entry was found and deleted, `Ok(false)` if not found. - async fn forget(&self, key: &str) -> anyhow::Result; + async fn forget(&self, namespace: &str, key: &str) -> anyhow::Result; + + /// Lists all namespaces with aggregate stats, for agent-side discovery. + async fn namespace_summaries(&self) -> anyhow::Result>; /// Returns the total count of all memory entries in the backend. async fn count(&self) -> anyhow::Result; diff --git a/src/openhuman/migration/core.rs b/src/openhuman/migration/core.rs index d21425cf1..f4ebfa1ce 100644 --- a/src/openhuman/migration/core.rs +++ b/src/openhuman/migration/core.rs @@ -91,7 +91,7 @@ pub async fn migrate_openclaw_memory( key = format!("openclaw_{idx}"); } - if let Some(existing) = memory.get(&key).await? { + if let Some(existing) = memory.get("", &key).await? { if existing.content.trim() == entry.content.trim() { stats.skipped_unchanged += 1; continue; @@ -103,7 +103,7 @@ pub async fn migrate_openclaw_memory( } memory - .store(&key, &entry.content, entry.category, None) + .store("", &key, &entry.content, entry.category, None) .await?; stats.imported += 1; } @@ -395,7 +395,7 @@ async fn next_available_key(memory: &dyn Memory, key: &str) -> Result { let mut idx = 1u32; loop { let candidate = format!("{key}_{idx}"); - if memory.get(&candidate).await?.is_none() { + if memory.get("", &candidate).await?.is_none() { return Ok(candidate); } idx += 1; diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index 53f2adf31..1973ed7d9 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -42,6 +42,7 @@ pub mod migration; pub mod node_runtime; pub mod notifications; pub mod overlay; +pub mod provider_surfaces; pub mod providers; pub mod referral; pub mod routing; diff --git a/src/openhuman/provider_surfaces/mod.rs b/src/openhuman/provider_surfaces/mod.rs new file mode 100644 index 000000000..e446ad3d1 --- /dev/null +++ b/src/openhuman/provider_surfaces/mod.rs @@ -0,0 +1,18 @@ +//! Local assistive surfaces for third-party provider apps. +//! +//! This domain will own the normalized event model, respond queue, local +//! draft shelf, and provider-specific assistive actions that sit above +//! embedded webviews and future API-first integrations. +//! +//! The initial scaffold is intentionally minimal so the namespace can be +//! wired into the controller registry before behavioral work begins. + +pub mod ops; +pub mod rpc; +pub mod schemas; +pub mod store; +pub mod types; + +pub use schemas::{ + all_provider_surfaces_controller_schemas, all_provider_surfaces_registered_controllers, +}; diff --git a/src/openhuman/provider_surfaces/ops.rs b/src/openhuman/provider_surfaces/ops.rs new file mode 100644 index 000000000..f0b2c3ec8 --- /dev/null +++ b/src/openhuman/provider_surfaces/ops.rs @@ -0,0 +1,136 @@ +//! Core operations for provider assistive surfaces. +//! +//! This initial cut keeps state in-memory so the RPC contract and UI wiring +//! can land before the SQLite-backed store arrives. + +use crate::openhuman::memory::{ApiEnvelope, ApiMeta, EmptyRequest}; +use crate::rpc::RpcOutcome; +use serde::Serialize; +use std::collections::BTreeMap; + +use super::store; +use super::types::{ProviderEvent, RespondQueueItem, RespondQueueListResponse}; + +fn request_id() -> String { + uuid::Uuid::new_v4().to_string() +} + +fn counts(entries: impl IntoIterator) -> BTreeMap { + entries + .into_iter() + .map(|(k, v)| (k.to_string(), v)) + .collect() +} + +fn envelope( + data: T, + counts: Option>, +) -> RpcOutcome> { + RpcOutcome::new( + ApiEnvelope { + data: Some(data), + error: None, + meta: ApiMeta { + request_id: request_id(), + latency_seconds: None, + cached: None, + counts, + pagination: None, + }, + }, + vec![], + ) +} + +pub async fn ingest_event( + request: ProviderEvent, +) -> Result>, String> { + tracing::debug!( + provider = %request.provider, + account_id = %request.account_id, + event_kind = %request.event_kind, + entity_id = %request.entity_id, + requires_attention = request.requires_attention, + "[provider-surfaces] ingest_event" + ); + let item = store::upsert_queue_item(request); + Ok(envelope(item, Some(counts([("queue_items", 1)])))) +} + +pub async fn list_queue( + _request: EmptyRequest, +) -> Result>, String> { + let items = store::list_queue_items(); + let count = items.len(); + tracing::debug!(count, "[provider-surfaces] list_queue"); + Ok(envelope( + RespondQueueListResponse { items, count }, + Some(counts([("queue_items", count)])), + )) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Mutex; + + /// Serializes tests that mutate the process-global RESPOND_QUEUE so cargo's + /// default parallel test runner cannot interleave clear/insert/assert cycles. + static TEST_MUTEX: Mutex<()> = Mutex::new(()); + + fn sample_event(entity_id: &str) -> ProviderEvent { + ProviderEvent { + provider: "linkedin".into(), + account_id: "acct-1".into(), + event_kind: "message".into(), + entity_id: entity_id.into(), + thread_id: Some("thread-1".into()), + title: Some("New message".into()), + snippet: Some("Can we talk tomorrow?".into()), + sender_name: Some("Taylor".into()), + sender_handle: Some("taylor".into()), + timestamp: "2026-04-22T16:55:00Z".into(), + deep_link: Some("https://www.linkedin.com/messaging/thread-1".into()), + requires_attention: true, + raw_payload: None, + } + } + + #[tokio::test] + async fn ingest_event_upserts_queue_item() { + let _lock = TEST_MUTEX.lock().unwrap_or_else(|p| p.into_inner()); + store::clear_queue(); + let first = ingest_event(sample_event("entity-1")).await.unwrap(); + let second = ingest_event(sample_event("entity-1")).await.unwrap(); + + let first_value = first.into_cli_compatible_json().unwrap(); + let second_value = second.into_cli_compatible_json().unwrap(); + let first_result = first_value.get("data").unwrap_or(&first_value); + let second_result = second_value.get("data").unwrap_or(&second_value); + + assert_eq!(first_result["provider"], "linkedin"); + assert_eq!(second_result["entity_id"], "entity-1"); + + let queue = list_queue(EmptyRequest {}).await.unwrap(); + let queue_json = queue.into_cli_compatible_json().unwrap(); + let data = queue_json.get("data").unwrap_or(&queue_json); + assert_eq!(data["count"], 1); + } + + #[tokio::test] + async fn list_queue_returns_newest_first() { + let _lock = TEST_MUTEX.lock().unwrap_or_else(|p| p.into_inner()); + store::clear_queue(); + ingest_event(sample_event("entity-1")).await.unwrap(); + ingest_event(sample_event("entity-2")).await.unwrap(); + + let queue = list_queue(EmptyRequest {}).await.unwrap(); + let queue_json = queue.into_cli_compatible_json().unwrap(); + let data = queue_json.get("data").unwrap_or(&queue_json); + let items = data["items"].as_array().unwrap(); + + assert_eq!(items.len(), 2); + assert_eq!(items[0]["entity_id"], "entity-2"); + assert_eq!(items[1]["entity_id"], "entity-1"); + } +} diff --git a/src/openhuman/provider_surfaces/rpc.rs b/src/openhuman/provider_surfaces/rpc.rs new file mode 100644 index 000000000..78df81940 --- /dev/null +++ b/src/openhuman/provider_surfaces/rpc.rs @@ -0,0 +1,4 @@ +//! RPC entry points for provider assistive surfaces. +//! +//! The first cut exposes normalized provider event ingestion plus a queue +//! listing endpoint for the local respond queue. diff --git a/src/openhuman/provider_surfaces/schemas.rs b/src/openhuman/provider_surfaces/schemas.rs new file mode 100644 index 000000000..db4d0a3b8 --- /dev/null +++ b/src/openhuman/provider_surfaces/schemas.rs @@ -0,0 +1,151 @@ +//! Controller registry for `provider_surfaces`. +//! +//! The first cut exposes normalized provider event ingestion plus a queue +//! listing endpoint suitable for local-first assistive UI surfaces. + +use serde::de::DeserializeOwned; +use serde_json::{Map, Value}; + +use crate::core::all::RegisteredController; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; +use crate::openhuman::memory::EmptyRequest; + +use super::ops; +use super::types::ProviderEvent; + +pub fn all_provider_surfaces_controller_schemas() -> Vec { + vec![schemas("ingest_event"), schemas("list_queue")] +} + +pub fn all_provider_surfaces_registered_controllers() -> Vec { + vec![ + RegisteredController { + schema: schemas("ingest_event"), + handler: handle_ingest_event, + }, + RegisteredController { + schema: schemas("list_queue"), + handler: handle_list_queue, + }, + ] +} + +pub fn schemas(function: &str) -> ControllerSchema { + match function { + "ingest_event" => ControllerSchema { + namespace: "provider_surfaces", + function: "ingest_event", + description: "Ingest a normalized provider event into the local respond queue.", + inputs: vec![ + field("provider", TypeSchema::String, "Provider slug (e.g. linkedin, gmail)."), + field("account_id", TypeSchema::String, "Provider account identifier."), + field("event_kind", TypeSchema::String, "Normalized event kind (e.g. message, mention)."), + field("entity_id", TypeSchema::String, "Stable provider entity identifier."), + optional("thread_id", TypeSchema::String, "Optional thread or conversation id."), + optional("title", TypeSchema::String, "Short human-readable title."), + optional("snippet", TypeSchema::String, "Preview snippet for queue rendering."), + optional("sender_name", TypeSchema::String, "Human-readable sender name."), + optional("sender_handle", TypeSchema::String, "Stable sender handle."), + field("timestamp", TypeSchema::String, "RFC3339 timestamp for the event."), + optional("deep_link", TypeSchema::String, "Provider deep link used to open the source surface."), + FieldSchema { + name: "requires_attention", + // ProviderEvent::requires_attention is #[serde(default)] so + // the deserializer accepts absence. Mark required: false here + // so the registry's validate_params agrees with the struct. + ty: TypeSchema::Bool, + comment: "Whether the event should enter the respond queue as actionable. Defaults to false when omitted.", + required: false, + }, + FieldSchema { + name: "raw_payload", + ty: TypeSchema::Option(Box::new(TypeSchema::Json)), + comment: "Optional provider-specific raw payload for future debugging and enrichment.", + required: false, + }, + ], + outputs: vec![json_output("result", "Envelope containing the upserted queue item.")], + }, + "list_queue" => ControllerSchema { + namespace: "provider_surfaces", + function: "list_queue", + description: "List the local respond queue derived from provider events.", + inputs: vec![], + outputs: vec![json_output("result", "Envelope containing queue items and count.")], + }, + _ => ControllerSchema { + namespace: "provider_surfaces", + function: "unknown", + description: "Unknown provider_surfaces controller.", + inputs: vec![], + outputs: vec![field("error", TypeSchema::String, "Lookup error details.")], + }, + } +} + +fn handle_ingest_event(params: Map) -> crate::core::all::ControllerFuture { + Box::pin(async move { + let payload: ProviderEvent = parse_params(params)?; + ops::ingest_event(payload).await?.into_cli_compatible_json() + }) +} + +fn handle_list_queue(params: Map) -> crate::core::all::ControllerFuture { + Box::pin(async move { + let payload: EmptyRequest = parse_params(params)?; + ops::list_queue(payload).await?.into_cli_compatible_json() + }) +} + +fn parse_params(params: Map) -> Result { + serde_json::from_value(Value::Object(params)).map_err(|e| format!("invalid params: {e}")) +} + +fn field(name: &'static str, ty: TypeSchema, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty, + comment, + required: true, + } +} + +fn optional(name: &'static str, ty: TypeSchema, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Option(Box::new(ty)), + comment, + required: false, + } +} + +fn json_output(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Json, + comment, + required: true, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn all_schemas_returns_two() { + assert_eq!(all_provider_surfaces_controller_schemas().len(), 2); + } + + #[test] + fn all_controllers_returns_two() { + assert_eq!(all_provider_surfaces_registered_controllers().len(), 2); + } + + #[test] + fn list_queue_schema_has_no_inputs() { + let schema = schemas("list_queue"); + assert!(schema.inputs.is_empty()); + assert_eq!(schema.namespace, "provider_surfaces"); + } +} diff --git a/src/openhuman/provider_surfaces/store.rs b/src/openhuman/provider_surfaces/store.rs new file mode 100644 index 000000000..6155788b6 --- /dev/null +++ b/src/openhuman/provider_surfaces/store.rs @@ -0,0 +1,70 @@ +//! Persistence for provider assistive surfaces. +//! +//! Follow-up work will add a SQLite-backed store for normalized provider +//! events, respond queue state, and local drafts. + +use std::sync::{Mutex, OnceLock}; + +use crate::openhuman::provider_surfaces::types::{ProviderEvent, RespondQueueItem}; + +/// Soft cap on the in-memory respond queue to bound growth under provider +/// firehose volume before the SQLite-backed store lands. The queue is +/// prepend-ordered, so oldest entries are dropped from the tail. +const MAX_QUEUE_ITEMS: usize = 500; + +static RESPOND_QUEUE: OnceLock>> = OnceLock::new(); + +fn queue() -> &'static Mutex> { + RESPOND_QUEUE.get_or_init(|| Mutex::new(Vec::new())) +} + +fn queue_lock() -> std::sync::MutexGuard<'static, Vec> { + queue() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn queue_item_id(event: &ProviderEvent) -> String { + format!( + "{}:{}:{}:{}", + event.provider, event.account_id, event.event_kind, event.entity_id + ) +} + +pub fn upsert_queue_item(event: ProviderEvent) -> RespondQueueItem { + let item = RespondQueueItem { + id: queue_item_id(&event), + provider: event.provider, + account_id: event.account_id, + event_kind: event.event_kind, + entity_id: event.entity_id, + thread_id: event.thread_id, + title: event.title, + snippet: event.snippet, + sender_name: event.sender_name, + sender_handle: event.sender_handle, + timestamp: event.timestamp, + deep_link: event.deep_link, + requires_attention: event.requires_attention, + status: "pending".to_string(), + }; + + let mut queue = queue_lock(); + if let Some(existing_idx) = queue.iter().position(|entry| entry.id == item.id) { + queue.remove(existing_idx); + } + queue.insert(0, item.clone()); + if queue.len() > MAX_QUEUE_ITEMS { + queue.truncate(MAX_QUEUE_ITEMS); + } + item +} + +pub fn list_queue_items() -> Vec { + queue_lock().clone() +} + +#[cfg(test)] +pub fn clear_queue() { + queue_lock().clear(); +} diff --git a/src/openhuman/provider_surfaces/types.rs b/src/openhuman/provider_surfaces/types.rs new file mode 100644 index 000000000..b521ce3c6 --- /dev/null +++ b/src/openhuman/provider_surfaces/types.rs @@ -0,0 +1,66 @@ +//! Shared types for provider assistive surfaces. + +use serde::{Deserialize, Serialize}; + +/// Inbound normalized provider event suitable for local assistive handling. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct ProviderEvent { + pub provider: String, + pub account_id: String, + pub event_kind: String, + pub entity_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub thread_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub title: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub snippet: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sender_name: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sender_handle: Option, + pub timestamp: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub deep_link: Option, + #[serde(default)] + pub requires_attention: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub raw_payload: Option, +} + +/// Queue item shown in the local respond queue. +/// +/// Field naming mirrors `ProviderEvent` and the declared controller schema +/// (`provider_surfaces::ingest_event` inputs), so callers see a single +/// snake_case contract on both request and response. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct RespondQueueItem { + pub id: String, + pub provider: String, + pub account_id: String, + pub event_kind: String, + pub entity_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub thread_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub title: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub snippet: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sender_name: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sender_handle: Option, + pub timestamp: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub deep_link: Option, + pub requires_attention: bool, + #[serde(default)] + pub status: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RespondQueueListResponse { + pub items: Vec, + pub count: usize, +} diff --git a/src/openhuman/tools/impl/memory/forget.rs b/src/openhuman/tools/impl/memory/forget.rs index dc4307cdd..6aabcc717 100644 --- a/src/openhuman/tools/impl/memory/forget.rs +++ b/src/openhuman/tools/impl/memory/forget.rs @@ -62,15 +62,28 @@ impl Tool for MemoryForgetTool { return Ok(ToolResult::error(error)); } - let namespaced_key = format!("{}/{}", namespace.trim(), key); - match self.memory.forget(&namespaced_key).await { - Ok(true) => Ok(ToolResult::success(format!( - "Forgot memory: {namespaced_key}" - ))), - Ok(false) => Ok(ToolResult::success(format!( - "No memory found with key: {key}" - ))), - Err(e) => Ok(ToolResult::error(format!("Failed to forget memory: {e}"))), + let namespace = namespace.trim(); + let legacy_key = format!("{namespace}/{key}"); + let display_key = format!("{namespace}/{key}"); + + // Try the new split namespace/key first (covers post-migration rows), + // then fall back to the legacy packed-key shape for rows that were + // stored before the boot migration ran (Phase A compatibility). + let deleted = match self.memory.forget(namespace, key).await { + Ok(true) => true, + Ok(false) => match self.memory.forget("", &legacy_key).await { + Ok(deleted) => deleted, + Err(e) => return Ok(ToolResult::error(format!("Failed to forget memory: {e}"))), + }, + Err(e) => return Ok(ToolResult::error(format!("Failed to forget memory: {e}"))), + }; + + if deleted { + Ok(ToolResult::success(format!("Forgot memory: {display_key}"))) + } else { + Ok(ToolResult::success(format!( + "No memory found with key: {display_key}" + ))) } } } @@ -104,6 +117,7 @@ mod tests { async fn forget_existing() { let (_tmp, mem) = test_mem(); mem.store( + "", "global/temp", "temporary", MemoryCategory::Conversation, @@ -120,7 +134,7 @@ mod tests { assert!(!result.is_error); assert!(result.output().contains("Forgot")); - assert!(mem.get("global/temp").await.unwrap().is_none()); + assert!(mem.get("", "global/temp").await.unwrap().is_none()); } #[tokio::test] @@ -147,6 +161,7 @@ mod tests { async fn forget_blocked_in_readonly_mode() { let (_tmp, mem) = test_mem(); mem.store( + "", "global/temp", "temporary", MemoryCategory::Conversation, @@ -165,13 +180,14 @@ mod tests { .unwrap(); assert!(result.is_error); assert!(result.output().contains("read-only mode")); - assert!(mem.get("global/temp").await.unwrap().is_some()); + assert!(mem.get("", "global/temp").await.unwrap().is_some()); } #[tokio::test] async fn forget_blocked_when_rate_limited() { let (_tmp, mem) = test_mem(); mem.store( + "", "global/temp", "temporary", MemoryCategory::Conversation, @@ -190,6 +206,6 @@ mod tests { .unwrap(); assert!(result.is_error); assert!(result.output().contains("Rate limit exceeded")); - assert!(mem.get("global/temp").await.unwrap().is_some()); + assert!(mem.get("", "global/temp").await.unwrap().is_some()); } } diff --git a/src/openhuman/tools/impl/memory/recall.rs b/src/openhuman/tools/impl/memory/recall.rs index 95a620439..2fa038d9a 100644 --- a/src/openhuman/tools/impl/memory/recall.rs +++ b/src/openhuman/tools/impl/memory/recall.rs @@ -71,11 +71,15 @@ impl Tool for MemoryRecallTool { .and_then(serde_json::Value::as_u64) .map_or(5, |v| v as usize); - // Search with the user query only. Prefixing `namespace` into the recall string adds a - // redundant token that matches almost every row (e.g. all `global/...` keys contain "global"). - // `namespace` is still required by the tool contract; unified memory scopes recall to the - // global document namespace until multi-namespace recall is wired through the `Memory` trait. - match self.memory.recall(query, limit, None).await { + // Search with the user query only. Prefixing `namespace` into the query + // string would add a redundant token matching almost every row. Instead, + // namespace scoping belongs in RecallOpts so the backend restricts the + // search to the correct namespace column. + let recall_opts = crate::openhuman::memory::RecallOpts { + namespace: Some(namespace), + ..crate::openhuman::memory::RecallOpts::default() + }; + match self.memory.recall(query, limit, recall_opts).await { Ok(entries) if entries.is_empty() => Ok(ToolResult::success( "No memories found matching that query.", )), @@ -126,16 +130,23 @@ mod tests { async fn recall_finds_match() { let (_tmp, mem) = seeded_mem(); mem.store( - "global/lang", + "global", + "lang", "User prefers Rust", MemoryCategory::Core, None, ) .await .unwrap(); - mem.store("global/tz", "Timezone is EST", MemoryCategory::Core, None) - .await - .unwrap(); + mem.store( + "global", + "tz", + "Timezone is EST", + MemoryCategory::Core, + None, + ) + .await + .unwrap(); let tool = MemoryRecallTool::new(mem); let result = tool @@ -152,7 +163,8 @@ mod tests { let (_tmp, mem) = seeded_mem(); for i in 0..10 { mem.store( - &format!("global/k{i}"), + "global", + &format!("k{i}"), &format!("Rust fact {i}"), MemoryCategory::Core, None, diff --git a/src/openhuman/tools/impl/memory/store.rs b/src/openhuman/tools/impl/memory/store.rs index 5e2a9155c..b70341ed9 100644 --- a/src/openhuman/tools/impl/memory/store.rs +++ b/src/openhuman/tools/impl/memory/store.rs @@ -82,15 +82,21 @@ impl Tool for MemoryStoreTool { return Ok(ToolResult::error(error)); } - let namespaced_key = format!("{}/{}", namespace.trim(), key); + let namespace = namespace.trim(); + if namespace.is_empty() { + return Ok(ToolResult::error("namespace cannot be empty".to_string())); + } + let key = key.trim(); + if key.is_empty() { + return Ok(ToolResult::error("key cannot be empty".to_string())); + } + let display_key = format!("{namespace}/{key}"); match self .memory - .store(&namespaced_key, content, category, None) + .store(namespace, key, content, category, None) .await { - Ok(()) => Ok(ToolResult::success(format!( - "Stored memory: {namespaced_key}" - ))), + Ok(()) => Ok(ToolResult::success(format!("Stored memory: {display_key}"))), Err(e) => Ok(ToolResult::error(format!("Failed to store memory: {e}"))), } } @@ -134,7 +140,7 @@ mod tests { assert!(!result.is_error); assert!(result.output().contains("lang")); - let entry = mem.get("global/lang").await.unwrap(); + let entry = mem.get("global", "lang").await.unwrap(); assert!(entry.is_some()); assert_eq!(entry.unwrap().content, "Prefers Rust"); } @@ -164,7 +170,7 @@ mod tests { .unwrap(); assert!(!result.is_error); - let entry = mem.get("global/proj_note").await.unwrap().unwrap(); + let entry = mem.get("global", "proj_note").await.unwrap().unwrap(); assert_eq!(entry.content, "Uses async runtime"); assert_eq!(entry.category, MemoryCategory::Custom("project".into())); } @@ -199,7 +205,7 @@ mod tests { .unwrap(); assert!(result.is_error); assert!(result.output().contains("read-only mode")); - assert!(mem.get("global/lang").await.unwrap().is_none()); + assert!(mem.get("global", "lang").await.unwrap().is_none()); } #[tokio::test] @@ -216,6 +222,6 @@ mod tests { .unwrap(); assert!(result.is_error); assert!(result.output().contains("Rate limit exceeded")); - assert!(mem.get("global/lang").await.unwrap().is_none()); + assert!(mem.get("global", "lang").await.unwrap().is_none()); } } diff --git a/src/openhuman/tools/impl/system/tool_stats.rs b/src/openhuman/tools/impl/system/tool_stats.rs index 86cabede1..5a8711f36 100644 --- a/src/openhuman/tools/impl/system/tool_stats.rs +++ b/src/openhuman/tools/impl/system/tool_stats.rs @@ -52,6 +52,7 @@ impl Tool for ToolStatsTool { let entries = self .memory .list( + Some("tool_effectiveness"), Some(&MemoryCategory::Custom("tool_effectiveness".into())), None, ) @@ -149,6 +150,7 @@ mod tests { } async fn store( &self, + namespace: &str, key: &str, content: &str, category: MemoryCategory, @@ -160,7 +162,7 @@ mod tests { id: key.to_string(), key: key.to_string(), content: content.to_string(), - namespace: None, + namespace: Some(namespace.to_string()), category, timestamp: "now".into(), session_id: session_id.map(str::to_string), @@ -173,23 +175,29 @@ mod tests { &self, _q: &str, _l: usize, - _s: Option<&str>, + _opts: crate::openhuman::memory::RecallOpts<'_>, ) -> anyhow::Result> { Ok(vec![]) } - async fn get(&self, key: &str) -> anyhow::Result> { + async fn get(&self, _namespace: &str, key: &str) -> anyhow::Result> { Ok(self.entries.lock().get(key).cloned()) } async fn list( &self, + _namespace: Option<&str>, _cat: Option<&MemoryCategory>, _s: Option<&str>, ) -> anyhow::Result> { Ok(self.entries.lock().values().cloned().collect()) } - async fn forget(&self, key: &str) -> anyhow::Result { + async fn forget(&self, _namespace: &str, key: &str) -> anyhow::Result { Ok(self.entries.lock().remove(key).is_some()) } + async fn namespace_summaries( + &self, + ) -> anyhow::Result> { + Ok(vec![]) + } async fn count(&self) -> anyhow::Result { Ok(self.entries.lock().len()) } @@ -237,6 +245,7 @@ mod tests { common_error_patterns: vec![], }; mem.store( + "tool_effectiveness", "tool/shell", &serde_json::to_string(&stats).unwrap(), MemoryCategory::Custom("tool_effectiveness".into()), @@ -264,6 +273,7 @@ mod tests { common_error_patterns: vec![], }; mem.store( + "tool_effectiveness", "tool/shell", &serde_json::to_string(&stats).unwrap(), MemoryCategory::Custom("tool_effectiveness".into()), diff --git a/tests/agent_builder_public.rs b/tests/agent_builder_public.rs index 8bb704fef..bb16e0206 100644 --- a/tests/agent_builder_public.rs +++ b/tests/agent_builder_public.rs @@ -69,6 +69,7 @@ struct StubMemory; impl Memory for StubMemory { async fn store( &self, + _namespace: &str, _key: &str, _content: &str, _category: MemoryCategory, @@ -81,27 +82,34 @@ impl Memory for StubMemory { &self, _query: &str, _limit: usize, - _session_id: Option<&str>, + _opts: openhuman_core::openhuman::memory::RecallOpts<'_>, ) -> Result> { Ok(Vec::new()) } - async fn get(&self, _key: &str) -> Result> { + async fn get(&self, _namespace: &str, _key: &str) -> Result> { Ok(None) } async fn list( &self, + _namespace: Option<&str>, _category: Option<&MemoryCategory>, _session_id: Option<&str>, ) -> Result> { Ok(Vec::new()) } - async fn forget(&self, _key: &str) -> Result { + async fn forget(&self, _namespace: &str, _key: &str) -> Result { Ok(false) } + async fn namespace_summaries( + &self, + ) -> Result> { + Ok(Vec::new()) + } + async fn count(&self) -> Result { Ok(0) } diff --git a/tests/agent_harness_public.rs b/tests/agent_harness_public.rs index c8319232c..96005e78f 100644 --- a/tests/agent_harness_public.rs +++ b/tests/agent_harness_public.rs @@ -49,6 +49,7 @@ struct StubMemory; impl Memory for StubMemory { async fn store( &self, + _namespace: &str, _key: &str, _content: &str, _category: MemoryCategory, @@ -61,27 +62,34 @@ impl Memory for StubMemory { &self, _query: &str, _limit: usize, - _session_id: Option<&str>, + _opts: openhuman_core::openhuman::memory::RecallOpts<'_>, ) -> Result> { Ok(Vec::new()) } - async fn get(&self, _key: &str) -> Result> { + async fn get(&self, _namespace: &str, _key: &str) -> Result> { Ok(None) } async fn list( &self, + _namespace: Option<&str>, _category: Option<&MemoryCategory>, _session_id: Option<&str>, ) -> Result> { Ok(Vec::new()) } - async fn forget(&self, _key: &str) -> Result { + async fn forget(&self, _namespace: &str, _key: &str) -> Result { Ok(false) } + async fn namespace_summaries( + &self, + ) -> Result> { + Ok(Vec::new()) + } + async fn count(&self) -> Result { Ok(0) } diff --git a/tests/agent_memory_loader_public.rs b/tests/agent_memory_loader_public.rs index 4706c5476..6d924cfa9 100644 --- a/tests/agent_memory_loader_public.rs +++ b/tests/agent_memory_loader_public.rs @@ -13,6 +13,7 @@ struct ScriptedMemory { impl Memory for ScriptedMemory { async fn store( &self, + _namespace: &str, _key: &str, _content: &str, _category: MemoryCategory, @@ -25,7 +26,7 @@ impl Memory for ScriptedMemory { &self, query: &str, _limit: usize, - _session_id: Option<&str>, + _opts: openhuman_core::openhuman::memory::RecallOpts<'_>, ) -> Result> { if query.contains("working.user") { Ok(self.working.clone()) @@ -34,22 +35,29 @@ impl Memory for ScriptedMemory { } } - async fn get(&self, _key: &str) -> Result> { + async fn get(&self, _namespace: &str, _key: &str) -> Result> { Ok(None) } async fn list( &self, + _namespace: Option<&str>, _category: Option<&MemoryCategory>, _session_id: Option<&str>, ) -> Result> { Ok(Vec::new()) } - async fn forget(&self, _key: &str) -> Result { + async fn forget(&self, _namespace: &str, _key: &str) -> Result { Ok(false) } + async fn namespace_summaries( + &self, + ) -> Result> { + Ok(Vec::new()) + } + async fn count(&self) -> Result { Ok(0) }