Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions crates/openfang-api/src/channel_bridge.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, String> {
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,
Expand Down
1 change: 1 addition & 0 deletions crates/openfang-api/src/routes.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down
34 changes: 31 additions & 3 deletions crates/openfang-channels/src/bridge.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, String>;

/// 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<String, String> {
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()`.
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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) => {
Expand Down
34 changes: 34 additions & 0 deletions crates/openfang-kernel/src/kernel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<String>,
sender_name: Option<String>,
) -> KernelResult<AgentLoopResult> {
let handle: Option<Arc<dyn KernelHandle>> = self
.self_handle
.get()
.and_then(|w| w.upgrade())
.map(|arc| arc as Arc<dyn KernelHandle>);
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,
Expand Down Expand Up @@ -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()))
})?;
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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),
Expand Down
4 changes: 4 additions & 0 deletions crates/openfang-runtime/src/agent_loop.rs
Original file line number Diff line number Diff line change
Expand Up @@ -298,6 +298,7 @@ pub async fn run_agent_loop(
driver: Arc<dyn LlmDriver>,
available_tools: &[ToolDefinition],
kernel: Option<Arc<dyn KernelHandle>>,
caller_sender_id: Option<&str>,
skill_registry: Option<&SkillRegistry>,
mcp_connections: Option<&tokio::sync::Mutex<Vec<McpConnection>>>,
web_ctx: Option<&WebToolsContext>,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -1525,6 +1527,7 @@ pub async fn run_agent_loop_streaming(
driver: Arc<dyn LlmDriver>,
available_tools: &[ToolDefinition],
kernel: Option<Arc<dyn KernelHandle>>,
caller_sender_id: Option<&str>,
stream_tx: mpsc::Sender<StreamEvent>,
skill_registry: Option<&SkillRegistry>,
mcp_connections: Option<&tokio::sync::Mutex<Vec<McpConnection>>>,
Expand Down Expand Up @@ -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,
Expand Down
25 changes: 21 additions & 4 deletions crates/openfang-runtime/src/tool_runner.rs
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,7 @@ pub async fn execute_tool(
kernel: Option<&Arc<dyn KernelHandle>>,
allowed_tools: Option<&[String]>,
caller_agent_id: Option<&str>,
caller_sender_id: Option<&str>,
skill_registry: Option<&SkillRegistry>,
mcp_connections: Option<&tokio::sync::Mutex<Vec<mcp::McpConnection>>>,
web_ctx: Option<&WebToolsContext>,
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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<dyn KernelHandle>>,
caller_sender_id: Option<&str>,
) -> Result<String, String> {
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<dyn KernelHandle>>,
caller_sender_id: Option<&str>,
) -> Result<String, String> {
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}'.")),
}
Expand Down