harness_cortexdb/

lib.rs

//! # harness-cortexdb — CortexDB as a harness `Memory`
//!
//! Implements [`harness_core::Memory`] over [CortexDB](https://github.com/liliang-cn/cortexdb)'s
//! MCP server: `recall` calls `memory_search` (semantic / lexical vector
//! retrieval) and `write` calls `memory_save`. Drop it in anywhere a
//! `Memory` is expected — the `MemoryGuide`, the experience layer
//! (`harness-experience`), a scheduler, etc. — to get **semantic recall** and a
//! brain that can be *shared* with other tools (Claude Code / Codex all use
//! `~/.cortexdb` by default).
//!
//! ```ignore
//! use harness_cortexdb::CortexdbMemory;
//! use std::sync::Arc;
//!
//! // Spawn CortexDB's MCP server; share the global brain (~/.cortexdb).
//! let mem = Arc::new(CortexdbMemory::connect_stdio("cortexdb-mcp-stdio", &[]).await?);
//! let recorder = harness_loop::experience::ExperienceRecorder::new(mem); // semantic!
//! ```
//!
//! The memory does not have to be a process this agent spawns. Point it at a
//! CortexDB that is already running — one brain shared by a fleet of agents, an
//! editor and a chat client, and highly available in the ordinary way. Which
//! constructor depends on what that CortexDB serves, and the two look alike
//! written down:
//!
//! ```ignore
//! // An MCP server reachable over Streamable HTTP:
//! let mem = Arc::new(CortexdbMemory::connect_http("https://cortex.internal/mcp").await?);
//!
//! // CortexDB's own gRPC API — what the openclaw sidecar runs, on :47821.
//! // Requires the `grpc` feature; see `grpc::CortexdbGrpcMemory`.
//! let mem = Arc::new(CortexdbGrpcMemory::connect("http://127.0.0.1:47821").await?);
//! ```
//!
//! Harness `MemoryEntry` fields round-trip through CortexDB: `content` maps to
//! the memory content; `tags` + `source` are stored under CortexDB's
//! `metadata` and read back on recall.
//!
//! ## Record conversations, then distill them into the graph
//!
//! Pair the [`TranscriptRecorder`](harness_loop::experience::TranscriptRecorder) hook
//! (turns → CortexDB) with periodic [`CortexdbMemory::consolidate`] (memories →
//! knowledge graph, server-side):
//!
//! ```ignore
//! let mem = Arc::new(CortexdbMemory::connect_stdio("cortexdb-mcp-stdio", &[]).await?
//!     .with_scope("session").with_namespace("myapp-chat"));
//!
//! // 1. capture every turn as it happens
//! let (recorder, rx) = harness_loop::experience::TranscriptRecorder::new("sess-1");
//! harness_loop::experience::spawn_transcript_writer(rx, mem.clone());
//! let loop_ = AgentLoop::new(model).with_hook(Arc::new(recorder));
//!
//! // 2. every 10 min, let CortexDB distill accumulated turns into the graph
//! let m = mem.clone();
//! tokio::spawn(async move {
//!     let mut tick = tokio::time::interval(std::time::Duration::from_secs(600));
//!     loop { tick.tick().await; let _ = m.consolidate(serde_json::Value::Null).await; }
//! });
//! ```

use async_trait::async_trait;
/// `Memory` over CortexDB's own gRPC API — what a deployment already serves.
/// See [`grpc::CortexdbGrpcMemory`]. Requires the `grpc` feature.
#[cfg(feature = "grpc")]
pub mod grpc;
#[cfg(feature = "grpc")]
pub use grpc::CortexdbGrpcMemory;

use harness_core::{Memory, MemoryEntry, MemoryError, Tool};
use harness_mcp_client::McpClient;
use serde_json::{Value, json};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

/// A [`Memory`] backed by CortexDB's MCP `memory_save` / `memory_search` tools.
pub struct CortexdbMemory {
    // Owns the MCP session so the CortexDB child process stays alive.
    _client: McpClient,
    save: Arc<dyn Tool>,
    search: Arc<dyn Tool>,
    /// Optional server-side distillation tool (`knowledge_memory_consolidate`),
    /// present only if the connected CortexDB exposes it. See [`consolidate`].
    consolidate: Option<Arc<dyn Tool>>,
    scope: String,
    namespace: String,
    user_id: Option<String>,
    seq: AtomicU64,
}

impl CortexdbMemory {
    /// Spawn `program args...` as CortexDB's MCP stdio server and adapt it.
    /// (e.g. `connect_stdio("cortexdb-mcp-stdio", &[])`.)
    pub async fn connect_stdio(program: &str, args: &[&str]) -> anyhow::Result<Self> {
        Self::from_client(McpClient::connect_stdio(program, args).await?)
    }

    /// Connect to a remote **MCP** server over Streamable HTTP.
    ///
    /// This is an MCP endpoint, not CortexDB's own API. The distinction is easy
    /// to lose because both are written as a URL: a CortexDB started the way the
    /// openclaw sidecar starts one serves **gRPC** on `host:47821` and answers
    /// nothing at all to HTTP, so pointing this at it fails at the transport.
    /// For that deployment use [`CortexdbGrpcMemory`](crate::grpc::CortexdbGrpcMemory)
    /// (`grpc` feature), which speaks the protocol the server actually serves.
    ///
    /// Use this one when the CortexDB in front of you exposes MCP over HTTP.
    ///
    /// Requires the `http` feature of `harness-rs-mcp-client` (on by default,
    /// with rustls). For an untrusted URL use
    /// [`McpClient::connect_http_with_client`] and [`from_client`](Self::from_client),
    /// which leaves the redirect and DNS policy with the caller.
    ///
    /// ```ignore
    /// let mem = CortexdbMemory::connect_http("https://cortex.internal/mcp").await?;
    /// ```
    #[cfg(feature = "http")]
    pub async fn connect_http(url: &str) -> anyhow::Result<Self> {
        Self::from_client(McpClient::connect_http(url).await?)
    }

    /// Adapt an already-connected CortexDB MCP client — stdio, HTTP, or any
    /// transport [`McpClient`] grows later.
    pub fn from_client(client: McpClient) -> anyhow::Result<Self> {
        let tools = client.tools();
        let find = |name: &str| tools.iter().find(|t| t.name() == name).cloned();
        let save = find("memory_save")
            .ok_or_else(|| anyhow::anyhow!("CortexDB MCP server exposes no `memory_save` tool"))?;
        let search = find("memory_search").ok_or_else(|| {
            anyhow::anyhow!("CortexDB MCP server exposes no `memory_search` tool")
        })?;
        // Optional — older/lighter CortexDB builds may not expose it.
        let consolidate = find("knowledge_memory_consolidate");
        Ok(Self {
            _client: client,
            save,
            search,
            consolidate,
            scope: "global".into(),
            namespace: "harness".into(),
            user_id: None,
            seq: AtomicU64::new(0),
        })
    }

    /// CortexDB memory scope: `global` (default, shared), `user`, or `session`.
    pub fn with_scope(mut self, scope: impl Into<String>) -> Self {
        self.scope = scope.into();
        self
    }
    /// CortexDB namespace (default `harness`). Isolates this app's memories.
    pub fn with_namespace(mut self, ns: impl Into<String>) -> Self {
        self.namespace = ns.into();
        self
    }
    /// Scope memories to a user id (for `scope = user`).
    pub fn with_user_id(mut self, user: impl Into<String>) -> Self {
        self.user_id = Some(user.into());
        self
    }

    /// Whether the connected CortexDB exposes the graph-distillation tool.
    pub fn can_consolidate(&self) -> bool {
        self.consolidate.is_some()
    }

    /// Distill accumulated memories into CortexDB's **knowledge graph**
    /// (entities + relations), server-side, via `knowledge_memory_consolidate`.
    /// CortexDB does the extraction — harness just triggers it, so this costs no
    /// tokens on our side. Run it periodically (e.g. from `harness-scheduler`)
    /// after a batch of turns has been recorded.
    ///
    /// Operates on this instance's `scope`/`namespace`/`user_id`. The server
    /// **requires** a `reflect.recall.query`; this method supplies a broad
    /// default plus `promote_to_knowledge: true`, so `consolidate(Value::Null)`
    /// works out of the box. Override either via `extra_args`, e.g.:
    ///
    /// ```ignore
    /// mem.consolidate(serde_json::json!({
    ///     "reflect": { "recall": { "query": "past tasks and tools", "max_memory_items": 30 } }
    /// })).await?;
    /// ```
    ///
    /// **Quality note:** in pure-lexical mode (no embedder configured on the
    /// CortexDB side) entity extraction is poor — configure an embedder for a
    /// clean graph. Returns the raw tool result. Errors if the server doesn't
    /// expose the tool (check [`can_consolidate`](Self::can_consolidate)).
    pub async fn consolidate(&self, extra_args: Value) -> Result<Value, MemoryError> {
        let tool = self.consolidate.as_ref().ok_or_else(|| {
            MemoryError::Backend(
                "this CortexDB build exposes no `knowledge_memory_consolidate` tool".into(),
            )
        })?;
        let args = build_consolidate_args(
            &self.scope,
            &self.namespace,
            self.user_id.as_deref(),
            extra_args,
        );
        let mut world = harness_context::default_world(".");
        let res = tool
            .invoke(args, &mut world)
            .await
            .map_err(|e| MemoryError::Backend(e.to_string()))?;
        if !res.ok {
            return Err(MemoryError::Backend(format!(
                "knowledge_memory_consolidate: {}",
                res.content
            )));
        }
        Ok(res.content)
    }

    fn next_id(&self) -> String {
        let n = self.seq.fetch_add(1, Ordering::SeqCst);
        let t = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        format!("harness-{t}-{n}")
    }
}

#[async_trait]
impl Memory for CortexdbMemory {
    async fn write(&self, entry: MemoryEntry) -> Result<(), MemoryError> {
        let id = if entry.id.is_empty() {
            self.next_id()
        } else {
            entry.id.clone()
        };
        let args = build_save_args(
            &id,
            &entry,
            &self.scope,
            &self.namespace,
            self.user_id.as_deref(),
        );

        let mut world = harness_context::default_world(".");
        let res = self
            .save
            .invoke(args, &mut world)
            .await
            .map_err(|e| MemoryError::Backend(e.to_string()))?;
        if !res.ok {
            return Err(MemoryError::Backend(format!(
                "memory_save: {}",
                res.content
            )));
        }
        Ok(())
    }

    async fn recall(&self, query: &str, k: usize) -> Result<Vec<MemoryEntry>, MemoryError> {
        if k == 0 || query.trim().is_empty() {
            return Ok(Vec::new());
        }
        let mut args = json!({
            "query": query,
            "top_k": k,
            "scope": self.scope,
            "namespace": self.namespace,
        });
        if let Some(u) = &self.user_id {
            args["user_id"] = json!(u);
        }

        let mut world = harness_context::default_world(".");
        let res = self
            .search
            .invoke(args, &mut world)
            .await
            .map_err(|e| MemoryError::Backend(e.to_string()))?;
        if !res.ok {
            return Err(MemoryError::Backend(format!(
                "memory_search: {}",
                res.content
            )));
        }
        // MemorySearchResponse { results: [ { memory: MemoryRecord, score } ] }
        let results = res
            .content
            .get("results")
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default();

        let mut out = Vec::with_capacity(results.len());
        for hit in &results {
            // The record is under `memory`; tolerate a flattened shape too.
            let rec = hit.get("memory").unwrap_or(hit);
            let content = rec
                .get("content")
                .and_then(|v| v.as_str())
                .unwrap_or_default()
                .to_string();
            if content.trim().is_empty() {
                continue;
            }
            let mut entry = MemoryEntry::new(content);
            if let Some(id) = rec.get("id").and_then(|v| v.as_str()) {
                entry.id = id.to_string();
            }
            if let Some(meta) = rec.get("metadata") {
                if let Some(tags) = meta.get("tags").and_then(|v| v.as_array()) {
                    entry.tags = tags
                        .iter()
                        .filter_map(|t| t.as_str().map(String::from))
                        .collect();
                }
                if let Some(src) = meta.get("source").and_then(|v| v.as_str()) {
                    entry.source = Some(src.to_string());
                }
            }
            out.push(entry);
        }
        Ok(out)
    }
}

/// Build the `memory_save` args for `entry`. Promotes `role:`/`session:` tags to
/// CortexDB's first-class `role` / `session_id` columns (so a transcript can be
/// queried and ordered by `session_id + role`, not just `json_extract` on
/// metadata); all remaining tags + `source` go into `metadata`.
/// Translate `MemoryEntry::expires_ms` — an absolute instant — into the
/// duration CortexDB wants. `None` stays `None`: absent, not zero, because
/// CortexDB reads `ttl_seconds: 0` as "retain forever", which is the opposite
/// of what an expired entry means. An instant already past floors at one
/// second so it dies on its own rather than being sent as a negative.
///
/// Both transports call this. They previously disagreed — the MCP path sent a
/// TTL and the gRPC path left the field at its `Default`, i.e. permanent — so
/// the same entry had two retentions depending on how it was written.
pub(crate) fn ttl_seconds_from(expires_ms: Option<i64>) -> Option<i64> {
    let expires_ms = expires_ms?;
    let now_ms = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0);
    Some((expires_ms - now_ms).div_euclid(1000).max(1))
}

fn build_save_args(
    id: &str,
    entry: &MemoryEntry,
    scope: &str,
    namespace: &str,
    user_id: Option<&str>,
) -> Value {
    let mut role: Option<String> = None;
    let mut session: Option<String> = None;
    let mut other_tags: Vec<String> = Vec::new();
    for t in &entry.tags {
        if let Some(r) = t.strip_prefix("role:") {
            role = Some(r.to_string());
        } else if let Some(s) = t.strip_prefix("session:") {
            session = Some(s.to_string());
        } else {
            other_tags.push(t.clone());
        }
    }

    let mut metadata = serde_json::Map::new();
    if !other_tags.is_empty() {
        metadata.insert("tags".into(), json!(other_tags));
    }
    if let Some(s) = &entry.source {
        metadata.insert("source".into(), json!(s));
    }

    let mut args = json!({
        "memory_id": id,
        "content": entry.content,
        "scope": scope,
        "namespace": namespace,
        "metadata": Value::Object(metadata),
    });
    if let Some(r) = role {
        args["role"] = json!(r);
    }
    if let Some(s) = session {
        args["session_id"] = json!(s);
    }
    if let Some(u) = user_id {
        args["user_id"] = json!(u);
    }
    // `expires_ms` is an absolute instant; CortexDB wants a duration. Translate
    // rather than drop: a dropped TTL is invisible — the write succeeds and the
    // entry simply never expires, so the caller learns nothing until the brain
    // is full of transcripts. Floor at 1s, because CortexDB reads `0` as
    // "retain forever", which is the opposite of what an expired entry means.
    if let Some(secs) = ttl_seconds_from(entry.expires_ms) {
        args["ttl_seconds"] = json!(secs);
    }
    args
}

/// Build the `knowledge_memory_consolidate` args: base scope/namespace/user_id,
/// merge caller `extra_args`, then fill the server-required `reflect.recall.query`
/// (+ `promote_to_knowledge`) defaults if the caller didn't supply them.
fn build_consolidate_args(
    scope: &str,
    namespace: &str,
    user_id: Option<&str>,
    extra_args: Value,
) -> Value {
    let mut args = json!({ "scope": scope, "namespace": namespace });
    if let Some(u) = user_id {
        args["user_id"] = json!(u);
    }
    if let Value::Object(extra) = extra_args
        && let Value::Object(base) = &mut args
    {
        base.extend(extra);
    }
    if args.get("reflect").is_none() {
        args["reflect"] = json!({
            "recall": {
                "query": "key facts, entities, decisions, tasks, and tools from recent conversations",
                "max_memory_items": 30
            }
        });
    }
    if args.get("promote_to_knowledge").is_none() {
        args["promote_to_knowledge"] = json!(true);
    }
    args
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn consolidate_null_fills_required_reflect_defaults() {
        let args = build_consolidate_args("global", "harness", Some("u1"), Value::Null);
        assert_eq!(args["scope"], "global");
        assert_eq!(args["user_id"], "u1");
        // Server-required reflect.recall.query is supplied.
        assert!(args["reflect"]["recall"]["query"].is_string());
        assert_eq!(args["reflect"]["recall"]["max_memory_items"], 30);
        assert_eq!(args["promote_to_knowledge"], true);
    }

    #[test]
    fn consolidate_extra_args_override_defaults() {
        let extra = json!({
            "reflect": { "recall": { "query": "custom", "max_memory_items": 5 } },
            "promote_to_knowledge": false
        });
        let args = build_consolidate_args("session", "app", None, extra);
        assert_eq!(args["reflect"]["recall"]["query"], "custom");
        assert_eq!(args["reflect"]["recall"]["max_memory_items"], 5);
        assert_eq!(args["promote_to_knowledge"], false);
        assert!(args.get("user_id").is_none());
    }

    #[test]
    fn role_and_session_promoted_to_native_params() {
        let entry = MemoryEntry::new("hi there")
            .with_source("transcript")
            .with_tags(["role:assistant", "session:sess-1", "topic:deploy"]);
        let args = build_save_args("m1", &entry, "session", "myapp", Some("u42"));

        // Promoted to first-class columns.
        assert_eq!(args["role"], "assistant");
        assert_eq!(args["session_id"], "sess-1");
        assert_eq!(args["user_id"], "u42");
        assert_eq!(args["scope"], "session");
        assert_eq!(args["namespace"], "myapp");

        // role:/session: stripped from metadata tags; unrelated tag kept.
        let tags = args["metadata"]["tags"].as_array().unwrap();
        assert_eq!(tags.len(), 1);
        assert_eq!(tags[0], "topic:deploy");
        assert_eq!(args["metadata"]["source"], "transcript");
    }

    #[test]
    fn ttl_is_carried_to_the_server_as_seconds() {
        // `MemoryEntry::expires_ms` is a contract: the doc on it says backends
        // MUST filter expired entries out of recall. Dropping it here made every
        // caller's TTL silently permanent — `memory_save`'s `ttl_days` argument
        // and `memory_layer`'s ttl both landed as "keep forever" in CortexDB
        // while the SurrealDB backend honoured them. Same entry, two answers.
        let entry = MemoryEntry::new("ephemeral").with_ttl_days(7);
        let args = build_save_args("m3", &entry, "session", "app", None);
        let ttl = args["ttl_seconds"].as_i64().expect("ttl_seconds sent");
        // 7 days, minus whatever the test itself took.
        assert!(
            (7 * 86_400 - 60..=7 * 86_400).contains(&ttl),
            "expected ~7d in seconds, got {ttl}"
        );
    }

    #[test]
    fn no_ttl_means_no_ttl_field() {
        // Absent, not zero: CortexDB reads `ttl_seconds: 0` as "retain forever",
        // so sending 0 would be right by accident. Say nothing instead.
        let entry = MemoryEntry::new("durable fact");
        let args = build_save_args("m4", &entry, "global", "app", None);
        assert!(args.get("ttl_seconds").is_none());
    }

    #[test]
    fn already_expired_entry_gets_the_shortest_ttl_not_a_negative_one() {
        // A negative ttl_seconds is not in CortexDB's vocabulary; it would be
        // rejected or, worse, read as unset. Floor at 1s so it dies on its own.
        let mut entry = MemoryEntry::new("stale");
        entry.expires_ms = Some(1); // 1970
        let args = build_save_args("m5", &entry, "session", "app", None);
        assert_eq!(args["ttl_seconds"], 1);
    }

    #[test]
    fn no_role_session_tags_omits_native_params() {
        let entry = MemoryEntry::new("note").with_tags(["misc"]);
        let args = build_save_args("m2", &entry, "global", "harness", None);
        assert!(args.get("role").is_none());
        assert!(args.get("session_id").is_none());
        assert!(args.get("user_id").is_none());
        assert_eq!(args["metadata"]["tags"][0], "misc");
    }
}