diff --git a/README.md b/README.md index 1799f3e..149c7ef 100644 --- a/README.md +++ b/README.md @@ -119,7 +119,7 @@ working wiring. | --- | --- | --- | --- | | `tinycortex` | TinyCortex, in-process | embedded | 3 (mandatory) via `provider`; all 18 via `TinycortexProvider` | | `supermemory` | Supermemory, hosted | external | 3 (mandatory) | -| `mem0` | Mem0, self-hosted | external | 3 (mandatory) | +| `mem0` | Mem0, hosted (`cloud`) or self-hosted | external | 3 (mandatory) | | `cognee` | Cognee, hosted or self-hosted | external | 3 (mandatory) | | `memory-git` | add-on: git-backed diff snapshots | — | requires `tinycortex` | | *(none)* | `NullMemoryProvider` | null | contract + registry only, 40 crates | @@ -177,10 +177,10 @@ that skips enforcement is the entire reason the policy layer exists. ## Remote engines The `tinymemory-remote` crate supports the managed and self-hosted native APIs -of Supermemory and Cognee, plus self-hosted Mem0. Each adapter stores -TinyMemory's key, category, session, and provenance in backend metadata (or a -Cognee raw-data envelope), so exact CRUD and portability survive the seam while -recall remains engine-native. Provider-facing dataset names, container tags, +of Supermemory, Cognee, and Mem0. Each adapter stores TinyMemory's key, +category, session, and provenance in backend metadata (or a Cognee raw-data +envelope), so exact CRUD and portability survive the seam while recall remains +engine-native. Provider-facing dataset names, container tags, and filenames are bounded stable hashes, so every namespace and key accepted by the TinyMemory contract remains valid on the remote API. @@ -196,19 +196,23 @@ Managed APIs have explicit constructors so their authentication cannot be confused with a self-hosted token: ```rust -use tinymemory_remote::{CogneeMemory, SupermemoryMemory}; +use tinymemory_remote::{CogneeMemory, Mem0Memory, SupermemoryMemory}; -let cognee = CogneeMemory::cloud("cognee-api-key")?; -let supermemory = SupermemoryMemory::cloud("sm_...")?; +// Cognee Cloud issues a per-tenant base URL (the API-key dashboard shows it); +// there is no shared endpoint, so its constructor takes one. +let cognee = CogneeMemory::api("https://tenant-.aws.cognee.ai", "cognee-api-key")?; -// Cognee also issues tenant-specific API origins. -let tenant = CogneeMemory::api("https://tenant.example.cognee.ai", "api-key")?; -# Ok::<_, anyhow::Error>((cognee, supermemory, tenant)) +// Supermemory and Mem0 both serve one hosted origin, so theirs take only a key. +let supermemory = SupermemoryMemory::cloud("sm_...")?; +let mem0 = Mem0Memory::cloud("m0-...")?; +# Ok::<_, anyhow::Error>((cognee, supermemory, mem0)) ``` Cognee Cloud uses `X-Api-Key`; authenticated self-hosted Cognee uses a bearer -access token. Supermemory uses bearer API keys for both deployment modes. All -constructors redact credentials from `Debug` output and transport errors. +access token. Supermemory uses bearer API keys for both deployment modes. Mem0's +hosted platform uses `Authorization: Token`, and self-hosted Mem0 uses +`X-API-Key`. All constructors redact credentials from `Debug` output, from +transport errors, and from the request's own header rendering. All three advertise the mandatory Core, Recall, and Portability families. The live Docker harness and conformance command are documented in diff --git a/adapters/remote/src/cognee.rs b/adapters/remote/src/cognee.rs index e0bb8fc..4bcede4 100644 --- a/adapters/remote/src/cognee.rs +++ b/adapters/remote/src/cognee.rs @@ -13,9 +13,6 @@ use crate::common::{stable_id, Dialect, HttpClient, RemoteMemory, StoredEntry}; /// Stable driver id used by configuration and status output. pub use tinymemory_api::drivers::COGNEE_DRIVER_ID; -/// Default base URL for Cognee's managed API. -pub const COGNEE_API_ENDPOINT: &str = "https://api.cognee.ai"; - /// A Cognee managed or self-hosted service exposed through TinyMemory's contract. #[derive(Debug)] pub struct CogneeMemory { @@ -51,10 +48,20 @@ impl CogneeMemory { }) } - /// Connect to a Cognee managed API using `X-Api-Key` authentication. + /// Connect to Cognee Cloud using `X-Api-Key` authentication. + /// + /// `endpoint` is **your tenant's** base URL, which Cognee Cloud issues per + /// account and prints on the API-key dashboard — it looks like + /// `https://tenant-.aws.cognee.ai`. There is deliberately no shared + /// default: this crate carried a `COGNEE_API_ENDPOINT` pointing at + /// `api.cognee.ai`, and that host answers no TLS handshake at all (its DNS + /// record resolves, nothing listens), so every "just use the default" + /// caller met a confusing transport error instead of a working client. + /// The tenant URL is the only address that exists. /// - /// This accepts a custom endpoint because Cognee Cloud may issue a - /// tenant-specific base URL. Use [`Self::cloud`] for the shared default. + /// The tenant and user ids the dashboard shows alongside the URL are not + /// needed here: the tenant is identified by the hostname, and the API's + /// only security scheme is this key. /// /// # Errors /// @@ -70,15 +77,6 @@ impl CogneeMemory { }), }) } - - /// Connect to Cognee's shared managed API endpoint. - /// - /// # Errors - /// - /// Returns an error when `api_key` is blank. - pub fn cloud(api_key: &str) -> anyhow::Result { - Self::api(COGNEE_API_ENDPOINT, api_key) - } } #[async_trait] @@ -183,7 +181,7 @@ impl CogneeDialect { async fn datasets(&self) -> anyhow::Result> { let response: Value = self .client - .json(Method::GET, "api/v1/datasets", None) + .json(Method::GET, "api/v1/datasets/", None) .await?; Ok(response .as_array() diff --git a/adapters/remote/src/cognee_test.rs b/adapters/remote/src/cognee_test.rs index 64afe24..ad80378 100644 --- a/adapters/remote/src/cognee_test.rs +++ b/adapters/remote/src/cognee_test.rs @@ -139,7 +139,10 @@ fn cognee_remote_names_are_bounded_and_safe_for_arbitrary_contract_keys() { async fn native_cognee_round_trips_the_tinymemory_contract() { let state = AppState::default(); let app = Router::new() - .route("/api/v1/datasets", get(datasets)) + // The real API serves the collection at the slashed form and 307s the + // bare one; the adapter now asks for `/api/v1/datasets/` directly, so + // the double must answer there or it stops mirroring the service. + .route("/api/v1/datasets/", get(datasets)) .route("/api/v1/datasets/{dataset}/data", get(data)) .route("/api/v1/datasets/{dataset}/data/{data}/raw", get(raw)) .route("/api/v1/datasets/{dataset}/data/{data}", delete(remove)) diff --git a/adapters/remote/src/common.rs b/adapters/remote/src/common.rs index 9ba5fb5..a9aa53d 100644 --- a/adapters/remote/src/common.rs +++ b/adapters/remote/src/common.rs @@ -4,6 +4,7 @@ use std::collections::BTreeMap; use anyhow::{bail, Context}; use async_trait::async_trait; +use reqwest::header::{HeaderValue, AUTHORIZATION}; use reqwest::{Method, RequestBuilder, StatusCode, Url}; use serde::{de::DeserializeOwned, Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -28,6 +29,13 @@ enum Auth { None, Bearer(String), ApiKey(String), + /// `Authorization: Token ` — Mem0's hosted platform. + /// + /// Distinct from [`Auth::Bearer`] on the wire *and* in behaviour: + /// api.mem0.ai routes a `Bearer` credential into its JWT verifier and + /// answers `token_not_valid`, so sending the wrong one of the two reports + /// a failure in the wrong subsystem. + Token(String), } impl std::fmt::Debug for HttpClient { @@ -90,6 +98,27 @@ async fn read_capped(response: reqwest::Response, path: &str) -> anyhow::Result< Ok(body) } +/// Wraps a credential in a header value that will not be printed back out. +/// +/// `RequestBuilder::bearer_auth` marks its `Authorization` value sensitive on +/// the caller's behalf; `RequestBuilder::header` handed a plain string does +/// not. So the two schemes that have no such helper -- `X-API-Key` and +/// `Authorization: Token` -- would otherwise carry a live API key through +/// every `Debug` rendering of the request and through any middleware that +/// formats headers. The flag is set here instead. +/// +/// Parsing up front is the second half of the same fix: a credential holding a +/// newline or another byte no header may carry becomes an error at the call +/// site, naming the credential, rather than a deferred failure inside `send` +/// that reads as a transport fault. The parse error carries no value, so the +/// credential does not reach the message either. +fn credential_header(value: &str) -> anyhow::Result { + let mut header = + HeaderValue::from_str(value).context("credential is not a valid HTTP header value")?; + header.set_sensitive(true); + Ok(header) +} + impl HttpClient { /// Builds a client that optionally authenticates with a bearer token. pub(crate) fn bearer(endpoint: &str, credential: Option<&str>) -> anyhow::Result { @@ -99,6 +128,14 @@ impl HttpClient { ) } + /// A client authenticating with `Authorization: Token `. + pub(crate) fn token(endpoint: &str, credential: Option<&str>) -> anyhow::Result { + Self::new( + endpoint, + credential.map_or(Auth::None, |value| Auth::Token(value.into())), + ) + } + /// Builds a client that optionally authenticates with `X-API-Key`. pub(crate) fn api_key(endpoint: &str, credential: Option<&str>) -> anyhow::Result { Self::new( @@ -136,7 +173,10 @@ impl HttpClient { Ok(match &self.auth { Auth::None => request, Auth::Bearer(token) => request.bearer_auth(token), - Auth::ApiKey(key) => request.header("X-API-Key", key), + Auth::ApiKey(key) => request.header("X-API-Key", credential_header(key)?), + Auth::Token(key) => { + request.header(AUTHORIZATION, credential_header(&format!("Token {key}"))?) + } }) } @@ -180,12 +220,28 @@ impl HttpClient { /// rejected credential specifically, because "HTTP 401" three layers deep /// in an anyhow chain reads as "the engine is down" and sends the operator /// to the wrong runbook. - fn status_error(&self, path: &str, status: reqwest::StatusCode) -> anyhow::Error { + fn status_error(&self, path: &str, status: reqwest::StatusCode, body: &str) -> anyhow::Error { let host = self.endpoint.host_str().unwrap_or(""); + // Hosted engines explain a rejection in the response body — mem0 + // answers `{"detail": "..."}`, cognee likewise — and discarding it + // turned "this one field is invalid" into a bare status code that + // said only that something, somewhere, was wrong. Truncated because + // an error body is not a payload budget, and only ever an error + // body: success responses never reach here. + let detail = body.trim(); + let detail = if detail.is_empty() { + String::new() + } else { + let mut shown: String = detail.chars().take(300).collect(); + if detail.chars().count() > 300 { + shown.push('…'); + } + format!(" — {shown}") + }; match status.as_u16() { 401 | 403 => { let hint = match &self.auth { - Auth::ApiKey(_) => "check the API key", + Auth::ApiKey(_) | Auth::Token(_) => "check the API key", Auth::Bearer(_) => "check the bearer token", Auth::None => { "the endpoint requires credentials this client was not configured with" @@ -193,10 +249,10 @@ impl HttpClient { }; anyhow::anyhow!( "memory API {path} on {host}: the configured credential was rejected \ - (HTTP {status}) — {hint}" + (HTTP {status}) — {hint}{detail}" ) } - _ => anyhow::anyhow!("memory API {path} on {host} returned HTTP {status}"), + _ => anyhow::anyhow!("memory API {path} on {host} returned HTTP {status}{detail}"), } } @@ -216,7 +272,8 @@ impl HttpClient { .map_err(|error| self.transport_error(error))?; let status = response.status(); if !status.is_success() { - return Err(self.status_error(path, status)); + let body = response.text().await.unwrap_or_default(); + return Err(self.status_error(path, status, &body)); } let body = read_capped(response, path).await?; serde_json::from_slice(&body) @@ -232,7 +289,8 @@ impl HttpClient { .map_err(|error| self.transport_error(error))?; let status = response.status(); if !status.is_success() { - return Err(self.status_error(path, status)); + let body = response.text().await.unwrap_or_default(); + return Err(self.status_error(path, status, &body)); } let body = read_capped(response, path).await?; String::from_utf8(body).context("memory API response was not valid UTF-8") @@ -255,7 +313,8 @@ impl HttpClient { .map_err(|error| self.transport_error(error))?; let status = response.status(); if !status.is_success() { - return Err(self.status_error(path, status)); + let body = response.text().await.unwrap_or_default(); + return Err(self.status_error(path, status, &body)); } Ok(status) } @@ -587,6 +646,72 @@ fn classify_transport(is_timeout: bool, is_connect: bool, chain: &str) -> &'stat } } +#[cfg(test)] +mod credential_header_tests { + #![allow(clippy::expect_used, clippy::panic)] + + use super::{credential_header, Auth, HttpClient}; + + /// The point of the helper. `reqwest` only redacts a header value whose + /// sensitive flag is set, and `RequestBuilder::header` handed a plain + /// string leaves it clear -- which is how an API key ends up rendered in + /// full by anything that formats the request. + #[test] + fn a_credential_header_is_marked_sensitive() { + let header = credential_header("Token m0-secret").expect("a plain key is a valid header"); + assert!(header.is_sensitive()); + } + + /// The value still has to be the credential; marking it sensitive must not + /// change what goes on the wire. + #[test] + fn marking_it_sensitive_does_not_change_the_value() { + let header = credential_header("Token m0-secret").expect("valid"); + assert_eq!(header.as_bytes(), b"Token m0-secret"); + } + + /// A credential carrying a newline cannot be a header. Rejecting it here + /// names the credential; letting it through defers the failure into `send`, + /// where it reads as a transport fault. + #[test] + fn a_credential_that_cannot_be_a_header_is_refused_by_name() { + let error = credential_header("key\r\nX-Injected: 1").expect_err("must not be accepted"); + assert!(format!("{error}").contains("credential"), "got: {error}"); + } + + /// And the refusal must not print the credential it refused. + #[test] + fn the_refusal_does_not_echo_the_credential() { + let error = + credential_header("supersecret\nX-Injected: 1").expect_err("must not be accepted"); + let rendered = format!("{error:?}"); + assert!(!rendered.contains("supersecret"), "leaked: {rendered}"); + } + + /// Both credential-bearing schemes go through the helper, so both reach + /// the wire redacted. `Auth::Bearer` is covered by `reqwest`'s own + /// `bearer_auth`, which sets the flag itself. + #[test] + fn both_manual_schemes_send_a_sensitive_authorization_value() { + for auth in [ + Auth::ApiKey("cg-secret".into()), + Auth::Token("m0-secret".into()), + ] { + let client = HttpClient::new("https://example.test", auth).expect("valid endpoint"); + let request = client + .request(reqwest::Method::GET, "v1/thing") + .expect("a plain key builds") + .build() + .expect("request builds"); + let sensitive = request + .headers() + .values() + .any(reqwest::header::HeaderValue::is_sensitive); + assert!(sensitive, "no sensitive header on {:?}", request.headers()); + } + } +} + #[cfg(test)] mod transport_tests { use super::classify_transport; diff --git a/adapters/remote/src/conformance_test.rs b/adapters/remote/src/conformance_test.rs index 8adca44..6d5dd74 100644 --- a/adapters/remote/src/conformance_test.rs +++ b/adapters/remote/src/conformance_test.rs @@ -485,7 +485,10 @@ async fn cg_recall(State(sets): State, Json(body): Json) -> Jso async fn cognee_backend() -> String { let sets: Datasets = Arc::new(Mutex::new(BTreeMap::new())); let app = Router::new() - .route("/api/v1/datasets", get(cg_datasets)) + // The real API serves the collection at the slashed form and 307s the + // bare one; the adapter now asks for `/api/v1/datasets/` directly, so + // the double must answer there or it stops mirroring the service. + .route("/api/v1/datasets/", get(cg_datasets)) .route("/api/v1/datasets/{dataset}/data", get(cg_data)) .route("/api/v1/datasets/{dataset}/data/{data_id}/raw", get(cg_raw)) .route( diff --git a/adapters/remote/src/failure_test.rs b/adapters/remote/src/failure_test.rs index b6631b7..d1a2a66 100644 --- a/adapters/remote/src/failure_test.rs +++ b/adapters/remote/src/failure_test.rs @@ -180,6 +180,41 @@ async fn an_unreachable_backend_is_reported_rather_than_hanging() { } } +#[tokio::test] +async fn a_cursor_that_never_clears_is_refused_rather_than_walked_for_ever() { + // Mem0's hosted arm pages until the server says stop: an empty page or a + // null `next`. Both are things the *server* controls, so a server that + // keeps answering a page and a cursor -- a bug, a proxy replaying one + // response, a filter that never narrows -- would spin the request loop and + // grow the buffer until the process died. The self-hosted arm already + // refuses past its ceiling; this pins the hosted one doing the same. + let app = Router::new().fallback(any(|| async { + axum::Json(serde_json::json!({ + "count": 1, + "next": "https://api.mem0.ai/v3/memories/?page=2", + "previous": null, + "results": [{"id": "m-1", "memory": "x", "metadata": {}}] + })) + })); + let endpoint = serve(app).await; + let memory = Mem0Memory::api(&endpoint, "m0-test-key").expect("client"); + + // Bounded so a genuinely unbounded loop fails the test rather than hanging + // the suite: the ceiling is 500 requests against a local socket, which + // finishes far inside this. + let outcome = + tokio::time::timeout(std::time::Duration::from_secs(60), memory.get("ns", "k")).await; + + let Ok(result) = outcome else { + panic!("the hosted listing never terminated against a cursor that never clears"); + }; + let error = result.expect_err("a cursor that never clears cannot be answered correctly"); + assert!( + format!("{error:#}").contains("pages"), + "the refusal must name the page ceiling it hit, got: {error:#}" + ); +} + #[tokio::test] async fn a_paginated_export_terminates_instead_of_looping() { // The partial-page leg of §E6. A backend that keeps answering with a page diff --git a/adapters/remote/src/lib.rs b/adapters/remote/src/lib.rs index 70ab990..f3c14ae 100644 --- a/adapters/remote/src/lib.rs +++ b/adapters/remote/src/lib.rs @@ -13,8 +13,8 @@ mod common; pub mod mem0; pub mod supermemory; -pub use cognee::{CogneeMemory, COGNEE_API_ENDPOINT, COGNEE_DRIVER_ID}; -pub use mem0::{Mem0Memory, MEM0_DRIVER_ID}; +pub use cognee::{CogneeMemory, COGNEE_DRIVER_ID}; +pub use mem0::{Mem0Memory, MEM0_API_ENDPOINT, MEM0_DRIVER_ID}; pub use supermemory::{SupermemoryMemory, SUPERMEMORY_API_ENDPOINT, SUPERMEMORY_DRIVER_ID}; use std::sync::Arc; diff --git a/adapters/remote/src/mem0.rs b/adapters/remote/src/mem0.rs index 747e94b..e61dacd 100644 --- a/adapters/remote/src/mem0.rs +++ b/adapters/remote/src/mem0.rs @@ -1,4 +1,9 @@ -//! Self-hosted Mem0 REST adapter. +//! Mem0 REST adapter — self-hosted server and hosted platform. +//! +//! The two are different APIs behind one product name, and this adapter speaks +//! both. What differs is the credential header, the path shapes, and how a +//! listing is scoped; what does not differ is the record model, so `decode` +//! and `metadata` are shared verbatim. use anyhow::Context; use async_trait::async_trait; @@ -13,7 +18,50 @@ use crate::common::{category, Dialect, HttpClient, RemoteMemory, StoredEntry}; /// Stable driver id used by configuration and status output. pub use tinymemory_api::drivers::MEM0_DRIVER_ID; -/// A self-hosted Mem0 server exposed through TinyMemory's storage contract. +/// Base URL of Mem0's hosted platform. +pub const MEM0_API_ENDPOINT: &str = "https://api.mem0.ai"; + +/// The Mem0 API this client speaks. +/// +/// Selected by the constructor rather than sniffed: the two APIs answer the +/// same 401 to an unauthenticated probe, so a client that guessed would only +/// discover it guessed wrong after a credential was accepted. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Flavour { + /// The open-source server: `X-API-Key`, un-prefixed REST paths. + SelfHosted, + /// The hosted platform at api.mem0.ai: `Authorization: Token`, v3 paths + /// for add/search/list and v1 for the by-id operations. That version mix + /// is the platform's own, not an oversight here. + Cloud, +} + +/// The `agent_id` every record this adapter writes to the hosted platform +/// carries. +/// +/// The platform refuses a listing that names no entity id, so an adapter that +/// only ever set `user_id` could not enumerate across namespaces — and +/// `namespace_summaries`, `count`, and every exact-key lookup need exactly +/// that. Stamping one constant agent id makes "everything this adapter owns" +/// expressible as a filter, and keeps the adapter's records distinguishable +/// from anything else in the same Mem0 project. +const CLOUD_AGENT_ID: &str = "tinymemory"; + +/// Records requested per hosted-platform listing page. +const CLOUD_PAGE_SIZE: u32 = 200; + +/// The most pages one hosted-platform listing will walk. +/// +/// The walk already stops on an empty page and on a null `next`, which covers +/// a well-behaved server. It does not cover a server that keeps answering a +/// full page and a non-null cursor: that spins the loop and grows the buffer +/// until the process dies. 500 pages is 100_000 records -- far past any real +/// account this adapter writes, and small enough that the failure arrives as a +/// message rather than an OOM. +const CLOUD_MAX_PAGES: u32 = 500; + +/// A Mem0 service — self-hosted or hosted — exposed through TinyMemory's +/// storage contract. #[derive(Debug)] pub struct Mem0Memory { inner: RemoteMemory, @@ -29,9 +77,50 @@ impl Mem0Memory { /// /// Returns an error when `endpoint` is not an HTTP(S) URL. pub fn new(endpoint: &str, api_key: Option<&str>) -> anyhow::Result { + Self::self_hosted(endpoint, api_key) + } + + /// Connect to a self-hosted Mem0 REST server (`X-API-Key`). + /// + /// # Errors + /// + /// Returns an error when `endpoint` is not an HTTP(S) URL. + pub fn self_hosted(endpoint: &str, api_key: Option<&str>) -> anyhow::Result { Ok(Self { inner: RemoteMemory::new(Mem0Dialect { client: HttpClient::api_key(endpoint, api_key)?, + flavour: Flavour::SelfHosted, + }), + }) + } + + /// Connect to Mem0's hosted platform at [`MEM0_API_ENDPOINT`]. + /// + /// Authenticates with `Authorization: Token ` — the platform's + /// scheme, and not interchangeable with a bearer token: a `Bearer` + /// credential reaches the platform's JWT verifier instead and fails as + /// `token_not_valid`, which reads as a broken token rather than a wrong + /// header. + /// + /// # Errors + /// + /// Returns an error when `api_key` is blank. + pub fn cloud(api_key: &str) -> anyhow::Result { + anyhow::ensure!(!api_key.trim().is_empty(), "mem0 API key must not be empty"); + Self::api(MEM0_API_ENDPOINT, api_key) + } + + /// Connect to a Mem0 platform deployment at a custom base URL. + /// + /// # Errors + /// + /// Returns an error when `endpoint` is invalid or `api_key` is blank. + pub fn api(endpoint: &str, api_key: &str) -> anyhow::Result { + anyhow::ensure!(!api_key.trim().is_empty(), "mem0 API key must not be empty"); + Ok(Self { + inner: RemoteMemory::new(Mem0Dialect { + client: HttpClient::token(endpoint, Some(api_key))?, + flavour: Flavour::Cloud, }), }) } @@ -116,6 +205,7 @@ impl Memory for Mem0Memory { /// Mem0-specific REST operations and wire-format conversion. struct Mem0Dialect { client: HttpClient, + flavour: Flavour, } impl Mem0Dialect { @@ -144,27 +234,111 @@ impl Mem0Dialect { /// properly needs Mem0's paging parameters verified against a live /// service; guessing them here would trade a loud failure for a quiet one. async fn values(&self) -> anyhow::Result> { - let top_k = Self::LISTING_TOP_K; - let response: Value = self - .client - .json(Method::GET, &format!("memories?top_k={top_k}"), None) - .await?; - let results = response - .get("results") - .and_then(Value::as_array) - .cloned() - .unwrap_or_default(); - if results.len() >= top_k { - anyhow::bail!( - "mem0 returned {} memories, this adapter's unpaginated listing ceiling. \ - Exact reads (get/list/count/export) cannot be answered correctly beyond \ - it -- a record past the window would read as absent -- so the adapter \ - refuses rather than answering wrongly. Recall is unaffected (it queries \ - mem0's search API directly).", - results.len() - ); + match self.flavour { + // Self-hosted: main's unpaginated listing with its truncation + // guard, unchanged. The guard is why the cloud arm below had to + // paginate rather than inherit this shape. + Flavour::SelfHosted => { + let top_k = Self::LISTING_TOP_K; + let response: Value = self + .client + .json(Method::GET, &format!("memories?top_k={top_k}"), None) + .await?; + let results = response + .get("results") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + if results.len() >= top_k { + anyhow::bail!( + "mem0 returned {} memories, this adapter's unpaginated listing ceiling. \ + Exact reads (get/list/count/export) cannot be answered correctly beyond \ + it -- a record past the window would read as absent -- so the adapter \ + refuses rather than answering wrongly. Recall is unaffected (it queries \ + mem0's search API directly).", + results.len() + ); + } + Ok(results) + } + // The hosted platform pages properly: it lists by POST with a + // mandatory entity filter and answers + // `{count, next, previous, results}`. That is the paging this + // adapter's self-hosted arm documents as unverified — here it is + // verified against Mem0's API reference, so this arm has no + // ceiling to refuse at for a *correct* server. Paging stops on an + // empty page as well as a null `next`, so a server that omits the + // cursor cannot spin the loop -- but one that keeps answering a + // full page and a non-null `next` still can, so the walk is + // bounded below and fails loudly at the bound rather than + // collecting for ever. + Flavour::Cloud => { + let mut all = Vec::new(); + let mut page = 1_u32; + loop { + let response: Value = self + .client + .json( + Method::POST, + &format!("v3/memories/?page={page}&page_size={CLOUD_PAGE_SIZE}"), + Some(&json!({"filters": {"agent_id": CLOUD_AGENT_ID}})), + ) + .await?; + let results = response + .get("results") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + let exhausted = + results.is_empty() || response.get("next").is_none_or(Value::is_null); + all.extend(results); + if exhausted { + break; + } + anyhow::ensure!( + page < CLOUD_MAX_PAGES, + "mem0's hosted platform still reported more memories after \ + {CLOUD_MAX_PAGES} pages of {CLOUD_PAGE_SIZE}. A cursor that never \ + clears is a server fault, not a large account, and continuing \ + would neither terminate nor answer correctly." + ); + page = page.saturating_add(1); + } + Ok(all) + } + } + } + + /// The search body both flavours send. + /// + /// `threshold` is **omitted** rather than sent as null when the caller set + /// no minimum score: the platform types it as a number in 0..=1 and + /// rejects an explicit null with a 400, which is how a recall against + /// mem0's hosted API failed while store and list succeeded. `top_k` is + /// clamped to the documented 1..=1000 for the same reason — a limit + /// outside it is a validation error, not a smaller result set. + fn search_body(query: &str, limit: usize, filters: Value, min_score: Option) -> Value { + let mut body = json!({ + "query": query, + "filters": filters, + "top_k": limit.clamp(1, 1000), + }); + if let (Some(object), Some(threshold)) = (body.as_object_mut(), min_score) { + object.insert("threshold".into(), json!(threshold)); + } + body + } + + /// The path addressing one record by its remote id. + /// + /// The platform serves the by-id operations under **v1** while add, + /// search and list are v3. That mix is the platform's own; keeping it in + /// one place stops it being re-derived (or "corrected") at each call site. + fn by_id_path(&self, remote_id: &str) -> String { + match self.flavour { + Flavour::SelfHosted => format!("memories/{remote_id}"), + Flavour::Cloud => format!("v1/memories/{remote_id}/"), } - Ok(results) } /// Decodes a Mem0 result containing TinyMemory-owned metadata. @@ -232,27 +406,33 @@ impl Dialect for Mem0Dialect { .find(|item| item.namespace == entry.namespace && item.key == entry.key); let metadata = Self::metadata(&entry); if let Some(existing) = existing { + // Both APIs take the same update body; only the path differs. self.client .empty( Method::PUT, - &format!("memories/{}", existing.remote_id), + &self.by_id_path(&existing.remote_id), Some(&json!({"text": entry.content, "metadata": metadata})), ) .await?; } else { - self.client - .empty( - Method::POST, - "memories", - Some(&json!({ - "messages": [{"role": "user", "content": entry.content}], - "user_id": entry.namespace, - "run_id": entry.session_id, - "metadata": metadata, - "infer": false - })), - ) - .await?; + let mut body = json!({ + "messages": [{"role": "user", "content": entry.content}], + "user_id": entry.namespace, + "run_id": entry.session_id, + "metadata": metadata, + "infer": false + }); + if self.flavour == Flavour::Cloud { + // Makes this record enumerable — see `CLOUD_AGENT_ID`. + if let Some(object) = body.as_object_mut() { + object.insert("agent_id".into(), json!(CLOUD_AGENT_ID)); + } + } + let path = match self.flavour { + Flavour::SelfHosted => "memories", + Flavour::Cloud => "v3/memories/add/", + }; + self.client.empty(Method::POST, path, Some(&body)).await?; } Ok(()) } @@ -274,20 +454,45 @@ impl Dialect for Mem0Dialect { limit: usize, opts: RecallOpts<'_>, ) -> anyhow::Result> { - let mut filters = serde_json::Map::new(); - if let Some(namespace) = opts.namespace { - filters.insert("user_id".into(), json!(namespace)); - } - let response: Value = self - .client - .json( - Method::POST, - "search", - Some(&json!({ - "query": query, "filters": filters, "top_k": limit, "threshold": opts.min_score - })), - ) - .await?; + let response: Value = match self.flavour { + Flavour::SelfHosted => { + let mut filters = serde_json::Map::new(); + if let Some(namespace) = opts.namespace { + filters.insert("user_id".into(), json!(namespace)); + } + self.client + .json( + Method::POST, + "search", + Some(&Self::search_body( + query, + limit, + Value::Object(filters), + opts.min_score, + )), + ) + .await? + } + // The platform requires entity ids inside `filters` and supports + // AND/OR; scoping to this adapter's agent id keeps a search from + // returning records written by anything else in the project. + Flavour::Cloud => { + let filters = match opts.namespace { + Some(namespace) => json!({"AND": [ + {"agent_id": CLOUD_AGENT_ID}, + {"user_id": namespace} + ]}), + None => json!({"agent_id": CLOUD_AGENT_ID}), + }; + self.client + .json( + Method::POST, + "v3/memories/search/", + Some(&Self::search_body(query, limit, filters, opts.min_score)), + ) + .await? + } + }; let values = response .get("results") .and_then(Value::as_array) @@ -307,11 +512,7 @@ impl Dialect for Mem0Dialect { return Ok(false); }; self.client - .empty( - Method::DELETE, - &format!("memories/{}", entry.remote_id), - None, - ) + .empty(Method::DELETE, &self.by_id_path(&entry.remote_id), None) .await .context("failed to delete Mem0 memory")?; Ok(true) @@ -326,3 +527,42 @@ impl Dialect for Mem0Dialect { #[cfg(test)] #[path = "mem0_test.rs"] mod test; + +#[cfg(test)] +mod search_body_tests { + use super::*; + + /// A recall with no minimum score must omit `threshold`, not send null. + /// The hosted platform types it as a number in 0..=1 and answers 400 to + /// an explicit null — store and list succeeded while recall failed. + #[test] + fn an_unset_min_score_omits_the_threshold_field() { + let body = Mem0Dialect::search_body("q", 10, json!({"user_id": "ns"}), None); + assert!( + body.get("threshold").is_none(), + "threshold must be absent, not null: {body}" + ); + assert_eq!(body["top_k"], 10); + assert_eq!(body["query"], "q"); + } + + #[test] + fn a_set_min_score_is_sent() { + let body = Mem0Dialect::search_body("q", 10, json!({"user_id": "ns"}), Some(0.25)); + assert_eq!(body["threshold"], 0.25); + } + + /// `top_k` outside the documented 1..=1000 is a validation error, so a + /// caller's limit is clamped rather than forwarded into a 400. + #[test] + fn top_k_is_clamped_to_the_documented_range() { + assert_eq!( + Mem0Dialect::search_body("q", 0, json!({}), None)["top_k"], + 1 + ); + assert_eq!( + Mem0Dialect::search_body("q", 5000, json!({}), None)["top_k"], + 1000 + ); + } +}