Consolidate memory RPC in tools/memory.rs

- Move memory_rpc(), socket_path(), SocketConn from mcp_server.rs
- Convert remaining callers to typed async API:
  - defs.rs: organize placeholder, run_agent query
  - cli/agent.rs: query resolution (now async)
  - mind/identity.rs: Store context loading
- Re-export socket_path/memory_rpc from mcp_server for compatibility

All external memory access now goes through tools/memory.rs typed API.

Co-Authored-By: Proof of Concept <poc@bcachefs.org>
This commit is contained in:
Kent Overstreet 2026-04-13 13:39:59 -04:00
parent 5b07a81aa7
commit fb46ab095d
6 changed files with 146 additions and 134 deletions

View file

@ -31,6 +31,115 @@ pub fn is_daemon() -> bool {
STORE_HANDLE.get().is_some() || LOCAL_STORE.with(|s| s.borrow().is_some())
}
// ── Socket RPC ─────────────────────────────────────────────────
use std::sync::Mutex;
use std::path::PathBuf;
pub fn socket_path() -> PathBuf {
dirs::home_dir()
.unwrap_or_default()
.join(".consciousness/mcp.sock")
}
// Cached socket connection for RPC forwarding
static SOCKET_CONN: OnceLock<Mutex<Option<SocketConn>>> = OnceLock::new();
struct SocketConn {
reader: std::io::BufReader<std::os::unix::net::UnixStream>,
writer: std::io::BufWriter<std::os::unix::net::UnixStream>,
next_id: u64,
}
impl SocketConn {
fn connect() -> Result<Self> {
use std::os::unix::net::UnixStream;
use std::io::{BufRead, BufReader, BufWriter, Write};
let path = socket_path();
let stream = UnixStream::connect(&path)?;
let mut reader = BufReader::new(stream.try_clone()?);
let mut writer = BufWriter::new(stream);
// Initialize MCP connection
let init = serde_json::json!({"jsonrpc": "2.0", "id": 1, "method": "initialize",
"params": {"protocolVersion": "2024-11-05", "capabilities": {},
"clientInfo": {"name": "forward", "version": "0.1"}}});
writeln!(writer, "{}", init)?;
writer.flush()?;
let mut buf = String::new();
reader.read_line(&mut buf)?;
Ok(Self { reader, writer, next_id: 1 })
}
fn call(&mut self, tool_name: &str, args: &serde_json::Value) -> Result<String> {
use std::io::{BufRead, Write};
self.next_id += 1;
let call = serde_json::json!({"jsonrpc": "2.0", "id": self.next_id, "method": "tools/call",
"params": {"name": tool_name, "arguments": args}});
writeln!(self.writer, "{}", call)?;
self.writer.flush()?;
let mut buf = String::new();
self.reader.read_line(&mut buf)?;
let resp: serde_json::Value = serde_json::from_str(&buf)?;
if let Some(err) = resp.get("error") {
anyhow::bail!("daemon error: {}", err);
}
let result = resp.get("result").cloned().unwrap_or(serde_json::json!({}));
let text = result.get("content")
.and_then(|c| c.as_array())
.and_then(|arr| arr.first())
.and_then(|c| c.get("text"))
.and_then(|t| t.as_str())
.unwrap_or("");
Ok(text.to_string())
}
}
/// Forward a tool call to the daemon socket, or execute locally if daemon is down.
/// Used by external processes that don't have direct store access.
pub fn memory_rpc(tool_name: &str, args: serde_json::Value) -> Result<String> {
let conn_lock = SOCKET_CONN.get_or_init(|| Mutex::new(None));
let mut guard = conn_lock.lock().unwrap();
// Try cached connection first
if let Some(conn) = guard.as_mut() {
match conn.call(tool_name, &args) {
Ok(result) => return Ok(result),
Err(_) => {
// Connection broken, clear cache and retry
*guard = None;
}
}
}
// Try to establish new connection
match SocketConn::connect() {
Ok(mut conn) => {
let result = conn.call(tool_name, &args);
*guard = Some(conn);
result
}
Err(_) => {
// Socket unavailable - fall back to local store
drop(guard); // Release lock before blocking
tokio::task::block_in_place(|| {
tokio::runtime::Handle::current()
.block_on(rpc_local(tool_name, &args))
})
}
}
}
/// Execute a tool locally when daemon isn't running.
async fn rpc_local(tool_name: &str, args: &serde_json::Value) -> Result<String> {
run_with_local_store(tool_name, args.clone()).await
}
// ── Helpers ────────────────────────────────────────────────────
fn get_str<'a>(args: &'a serde_json::Value, name: &'a str) -> Result<&'a str> {
@ -179,7 +288,7 @@ macro_rules! memory_tool {
#[allow(unused_mut)]
let mut map = serde_json::Map::new();
$($(memory_tool!(@insert_json map, $arg, $($typ)+);)*)?
return crate::mcp_server::memory_rpc(stringify!($name), serde_json::Value::Object(map));
return memory_rpc(stringify!($name), serde_json::Value::Object(map));
}
let prov = match agent {
Some(a) => a.state.lock().await.provenance.clone(),
@ -208,7 +317,7 @@ macro_rules! memory_tool {
#[allow(unused_mut)]
let mut map = serde_json::Map::new();
$($(memory_tool!(@insert_json map, $arg, $($typ)+);)*)?
return crate::mcp_server::memory_rpc(stringify!($name), serde_json::Value::Object(map));
return memory_rpc(stringify!($name), serde_json::Value::Object(map));
}
let prov = match agent {
Some(a) => a.state.lock().await.provenance.clone(),
@ -270,7 +379,7 @@ async fn dispatch(
// Forward to daemon
let name = tool_name.to_string();
return tokio::task::spawn_blocking(move || {
crate::mcp_server::memory_rpc(&name, args)
memory_rpc(&name, args)
}).await.map_err(|e| anyhow::anyhow!("spawn_blocking: {}", e))?;
}