diff --git a/crates/openfang-api/src/channel_bridge.rs b/crates/openfang-api/src/channel_bridge.rs index 37ad72921f..83cf8e6fc8 100644 --- a/crates/openfang-api/src/channel_bridge.rs +++ b/crates/openfang-api/src/channel_bridge.rs @@ -84,6 +84,23 @@ impl ChannelBridgeHandle for KernelBridgeAdapter { Ok(result.response) } + async fn send_message_from( + &self, + agent_id: AgentId, + message: &str, + sender_id: Option<&str>, + ) -> Result { + let result = self + .kernel + .send_message_from(agent_id, message, sender_id.map(String::from), None) + .await + .map_err(|e| format!("{e}"))?; + if result.silent { + return Ok(String::new()); + } + Ok(result.response) + } + async fn send_message_with_blocks( &self, agent_id: AgentId, diff --git a/crates/openfang-api/src/routes.rs b/crates/openfang-api/src/routes.rs index cebb1f599a..4ac0a7d68f 100644 --- a/crates/openfang-api/src/routes.rs +++ b/crates/openfang-api/src/routes.rs @@ -7074,6 +7074,7 @@ pub async fn mcp_http( Some(&kernel_handle), None, None, + None, Some(&skill_snapshot), Some(&state.kernel.mcp_connections), Some(&state.kernel.web_ctx), diff --git a/crates/openfang-channels/src/bridge.rs b/crates/openfang-channels/src/bridge.rs index 2043aeaa76..03317feace 100644 --- a/crates/openfang-channels/src/bridge.rs +++ b/crates/openfang-channels/src/bridge.rs @@ -102,6 +102,21 @@ pub trait ChannelBridgeHandle: Send + Sync { /// Send a message to an agent and get the text response. async fn send_message(&self, agent_id: AgentId, message: &str) -> Result; + /// Send a message on behalf of a specific end-user identity (e.g. the + /// Nextcloud Talk / Telegram / Discord actor who sent it), so tools like + /// memory_store/memory_recall can scope data per real person instead of + /// sharing one namespace across every channel user. Default falls back + /// to `send_message()` (no per-user scoping) for handles that don't + /// implement this. + async fn send_message_from( + &self, + agent_id: AgentId, + message: &str, + _sender_id: Option<&str>, + ) -> Result { + self.send_message(agent_id, message).await + } + /// Send a message with structured content blocks (text + images) to an agent. /// /// Default implementation extracts text from blocks and falls back to `send_message()`. @@ -1330,8 +1345,19 @@ async fn dispatch_message( text.clone() }; - // Send to agent and relay response - let result = handle.send_message(agent_id, &prefixed_text).await; + // Send to agent and relay response. Scope by the channel actor's own + // identity (falls back to platform_id, e.g. the room/chat id, for + // adapters that don't set a distinct actor_id) so per-user tools like + // memory_store/memory_recall don't leak across different real people. + let sender_scope = message + .metadata + .get("actor_id") + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + .unwrap_or(message.sender.platform_id.as_str()); + let result = handle + .send_message_from(agent_id, &prefixed_text, Some(sender_scope)) + .await; // Stop the typing refresh now that we have a response typing_task.abort(); @@ -1359,7 +1385,9 @@ async fn dispatch_message( // Try re-resolution before reporting error if let Some(new_id) = try_reresolution(&e, &channel_key, handle, router).await { let typing_task2 = spawn_typing_loop(adapter_arc.clone(), message.sender.clone()); - let retry = handle.send_message(new_id, &text).await; + let retry = handle + .send_message_from(new_id, &text, Some(sender_scope)) + .await; typing_task2.abort(); match retry { Ok(response) => { diff --git a/crates/openfang-kernel/src/kernel.rs b/crates/openfang-kernel/src/kernel.rs index 8f59414c97..b47c6eadb7 100644 --- a/crates/openfang-kernel/src/kernel.rs +++ b/crates/openfang-kernel/src/kernel.rs @@ -1831,6 +1831,28 @@ impl OpenFangKernel { .await } + /// Like `send_message`, but on behalf of a specific end-user identity + /// (e.g. the real person behind a Nextcloud Talk / Telegram / Discord + /// message), so per-user tools such as memory_store/memory_recall can + /// scope data correctly instead of sharing one namespace across every + /// channel user. Resolves the kernel handle the same way `send_message` + /// does. + pub async fn send_message_from( + &self, + agent_id: AgentId, + message: &str, + sender_id: Option, + sender_name: Option, + ) -> KernelResult { + let handle: Option> = self + .self_handle + .get() + .and_then(|w| w.upgrade()) + .map(|arc| arc as Arc); + self.send_message_with_handle(agent_id, message, handle, sender_id, sender_name) + .await + } + /// Send a multimodal message (text + images) to an agent and get a response. /// /// Used by channel bridges when a user sends a photo — the image is downloaded, @@ -1998,6 +2020,11 @@ impl OpenFangKernel { .check_quota(agent_id) .map_err(KernelError::OpenFang)?; + // Captured before `sender_id` is moved into the prompt context below, + // so tool calls (e.g. memory_store/memory_recall) can scope data + // per end-user identity instead of sharing one global namespace. + let sender_id_for_tools = sender_id.clone(); + let entry = self.registry.get(agent_id).ok_or_else(|| { KernelError::OpenFang(OpenFangError::AgentNotFound(agent_id.to_string())) })?; @@ -2355,6 +2382,7 @@ impl OpenFangKernel { driver, &tools, kernel_handle, + sender_id_for_tools.as_deref(), tx, Some(&skill_snapshot), Some(&kernel_clone.mcp_connections), @@ -2643,6 +2671,11 @@ impl OpenFangKernel { .check_quota(agent_id, &entry.manifest.resources) .map_err(KernelError::OpenFang)?; + // Captured before `sender_id` is moved into the prompt context below, + // so tool calls (e.g. memory_store/memory_recall) can scope data + // per end-user identity instead of sharing one global namespace. + let sender_id_for_tools = sender_id.clone(); + let mut session = self .memory .get_session(entry.session_id) @@ -2947,6 +2980,7 @@ impl OpenFangKernel { driver, &tools, kernel_handle, + sender_id_for_tools.as_deref(), Some(&skill_snapshot), Some(&self.mcp_connections), Some(&self.web_ctx), diff --git a/crates/openfang-runtime/src/agent_loop.rs b/crates/openfang-runtime/src/agent_loop.rs index 615fbfdb73..2a7d6119aa 100644 --- a/crates/openfang-runtime/src/agent_loop.rs +++ b/crates/openfang-runtime/src/agent_loop.rs @@ -298,6 +298,7 @@ pub async fn run_agent_loop( driver: Arc, available_tools: &[ToolDefinition], kernel: Option>, + caller_sender_id: Option<&str>, skill_registry: Option<&SkillRegistry>, mcp_connections: Option<&tokio::sync::Mutex>>, web_ctx: Option<&WebToolsContext>, @@ -933,6 +934,7 @@ pub async fn run_agent_loop( kernel.as_ref(), Some(&allowed_tool_names), Some(&caller_id_str), + caller_sender_id, skill_registry, mcp_connections, web_ctx, @@ -1525,6 +1527,7 @@ pub async fn run_agent_loop_streaming( driver: Arc, available_tools: &[ToolDefinition], kernel: Option>, + caller_sender_id: Option<&str>, stream_tx: mpsc::Sender, skill_registry: Option<&SkillRegistry>, mcp_connections: Option<&tokio::sync::Mutex>>, @@ -2142,6 +2145,7 @@ pub async fn run_agent_loop_streaming( kernel.as_ref(), Some(&allowed_tool_names), Some(&caller_id_str), + caller_sender_id, skill_registry, mcp_connections, web_ctx, diff --git a/crates/openfang-runtime/src/tool_runner.rs b/crates/openfang-runtime/src/tool_runner.rs index 16695616ee..cd189093ba 100644 --- a/crates/openfang-runtime/src/tool_runner.rs +++ b/crates/openfang-runtime/src/tool_runner.rs @@ -113,6 +113,7 @@ pub async fn execute_tool( kernel: Option<&Arc>, allowed_tools: Option<&[String]>, caller_agent_id: Option<&str>, + caller_sender_id: Option<&str>, skill_registry: Option<&SkillRegistry>, mcp_connections: Option<&tokio::sync::Mutex>>, web_ctx: Option<&WebToolsContext>, @@ -303,8 +304,8 @@ pub async fn execute_tool( "agent_activate" => tool_agent_activate(input, kernel), // Shared memory tools - "memory_store" => tool_memory_store(input, kernel), - "memory_recall" => tool_memory_recall(input, kernel), + "memory_store" => tool_memory_store(input, kernel, caller_sender_id), + "memory_recall" => tool_memory_recall(input, kernel, caller_sender_id), // Collaboration tools "agent_find" => tool_agent_find(input, kernel), @@ -1873,24 +1874,40 @@ fn tool_agent_activate( // Shared memory tools // --------------------------------------------------------------------------- +/// Namespace a memory key by the caller's channel identity, so different +/// end users (e.g. distinct Nextcloud Talk / Telegram / Discord senders) +/// never read or write each other's memory entries. Falls back to the +/// unprefixed key when no sender identity is available (dashboard/API/cron +/// callers, which are already owner-trusted). +fn scoped_memory_key(key: &str, caller_sender_id: Option<&str>) -> String { + match caller_sender_id { + Some(sender) if !sender.is_empty() => format!("user:{sender}:{key}"), + _ => key.to_string(), + } +} + fn tool_memory_store( input: &serde_json::Value, kernel: Option<&Arc>, + caller_sender_id: Option<&str>, ) -> Result { let kh = require_kernel(kernel)?; let key = input["key"].as_str().ok_or("Missing 'key' parameter")?; let value = input.get("value").ok_or("Missing 'value' parameter")?; - kh.memory_store(key, value.clone())?; + let scoped_key = scoped_memory_key(key, caller_sender_id); + kh.memory_store(&scoped_key, value.clone())?; Ok(format!("Stored value under key '{key}'.")) } fn tool_memory_recall( input: &serde_json::Value, kernel: Option<&Arc>, + caller_sender_id: Option<&str>, ) -> Result { let kh = require_kernel(kernel)?; let key = input["key"].as_str().ok_or("Missing 'key' parameter")?; - match kh.memory_recall(key)? { + let scoped_key = scoped_memory_key(key, caller_sender_id); + match kh.memory_recall(&scoped_key)? { Some(val) => Ok(serde_json::to_string_pretty(&val).unwrap_or_else(|_| val.to_string())), None => Ok(format!("No value found for key '{key}'.")), }