From 7d8641d8e3efb0b2c2e1ffb982631087c3adbf70 Mon Sep 17 00:00:00 2001 From: Shanu Date: Fri, 21 Aug 2026 00:11:42 +0530 Subject: [PATCH 1/2] adapters/remote: keyed CRUD stops enumerating the account (#69) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Dialect trait grows the keyed seam — namespace_entries and entry, with enumeration defaults that ARE the old behavior — and RemoteMemory routes get and namespace-scoped list through it. What each backend does with the seam: Supermemory: the per-tag fetch find_entry/delete always used now serves reads too; a keyed get asks for one container tag instead of walking every tag in the account. Mem0 self-hosted: the listing scopes by user_id (the namespace, which this adapter has always written there) — the one filter dimension every vector store supports — so the 1000-row refusal ceiling becomes per-namespace instead of a whole-store death sentence. Namespaces ride percent-encoded (they carry slashes). Mem0 hosted: a keyed get is ONE filtered request over the metadata this adapter has always stamped (tinymemory_key is top-level and equality-filtered, inside the platform's documented envelope), with verify-after-resolve: a server that ignores the filter clause and answers someone else's record is refused loudly, never served. The paged walk takes a caller-supplied filter so the namespace listing pages over one entity, not the account. Cognee: keyed ops resolve by the deterministic uploaded filename against the data listing the adapter already receives — the per-record raw-fetch fan-out (1 + D + N requests per get; ~10,002 on a 10k store) becomes three requests: dataset resolve, one listing, one raw, with the envelope still authoritative (a filename match that opens onto a different record refuses). Delete reads no envelopes at all. Upsert existence-probes through the same path on all three. Every scoped fetch is ALSO retained client-side: a backend that ignores its filter (an old OSS build, a proxy) must not leak sibling namespaces into keyed reads. Full walks (count, namespace_summaries, export) stay on entries() — the documented floor. Tests pin the request shapes, not just the answers: Supermemory names exactly one tag per keyed read; Mem0 self-hosted sends the encoded user_id; Mem0 cloud sends the metadata clause once and refuses a mismatched answer; Cognee performs exactly one listing and one raw per get and zero raws per delete. The cognee double now serves back the filename the adapter actually uploaded (the old hardcoded name was read by nothing); the never-clearing-cursor guard pins count(), since a poisoned cursor can no longer spin the keyed get at all. --- adapters/remote/src/cognee.rs | 156 ++++++++++++++---- adapters/remote/src/cognee_test.rs | 92 ++++++++++- adapters/remote/src/common.rs | 46 +++++- adapters/remote/src/failure_test.rs | 7 +- adapters/remote/src/mem0.rs | 201 +++++++++++++++++++----- adapters/remote/src/mem0_test.rs | 146 ++++++++++++++++- adapters/remote/src/supermemory.rs | 17 ++ adapters/remote/src/supermemory_test.rs | 53 ++++++- 8 files changed, 630 insertions(+), 88 deletions(-) diff --git a/adapters/remote/src/cognee.rs b/adapters/remote/src/cognee.rs index f91f6b33..0d86525a 100644 --- a/adapters/remote/src/cognee.rs +++ b/adapters/remote/src/cognee.rs @@ -279,6 +279,85 @@ impl CogneeDialect { Ok(entries) } + /// One dataset's data index — id and name per row, NO raw fetches + /// (issue #69). The listing already carries the uploaded filename, and + /// this adapter's filenames are deterministic (`Self::filename`), so a + /// key resolves by matching the name — the per-record raw-fetch loop the + /// first cut ran existed only because it read the key out of each + /// envelope body instead. + async fn data_index(&self, dataset: &Dataset) -> anyhow::Result> { + let response: Value = self + .client + .json( + Method::GET, + &format!("api/v1/datasets/{}/data", dataset.id), + None, + Attempts::RetryTransient, + ) + .await?; + Ok(response + .as_array() + .into_iter() + .flatten() + .filter_map(|data| { + Some(( + data.get("id")?.as_str()?.to_owned(), + data.get("name")?.as_str()?.to_owned(), + )) + }) + .collect()) + } + + /// Resolves a key to its data id by deterministic-filename match: + /// Cognee's loader strips the final `.json`, so both spellings count. + /// On a duplicate name (possible only if a historical blind re-add ever + /// raced), newest-listed wins deterministically — the listing is + /// insertion-ordered — rather than an arbitrary pick. + async fn find_data_id(&self, dataset: &Dataset, key: &str) -> anyhow::Result> { + let uploaded = Self::filename(key); + let stripped = uploaded.trim_end_matches(".json").to_owned(); + Ok(self + .data_index(dataset) + .await? + .into_iter() + .rev() + .find(|(_, name)| *name == uploaded || *name == stripped) + .map(|(id, _)| id)) + } + + /// Downloads and decodes ONE envelope by its ids, verifying it is the + /// record asked for — the envelope stays authoritative over the filename + /// match (a hash collision or a foreign file with our extension must not + /// serve as someone else's memory). + async fn fetch_entry( + &self, + dataset: &Dataset, + data_id: &str, + namespace: &str, + key: &str, + ) -> anyhow::Result> { + let raw = self + .client + .text( + Method::GET, + &format!("api/v1/datasets/{}/data/{data_id}/raw", dataset.id), + Attempts::RetryTransient, + ) + .await?; + let mut entry: StoredEntry = + serde_json::from_str(&raw).context("Cognee record envelope is invalid")?; + if entry.namespace != namespace || entry.key != key { + anyhow::bail!( + "Cognee data {data_id} matched key `{key}` by filename but its envelope names \ + {}/{} — refusing to serve a mismatched record", + entry.namespace, + entry.key + ); + } + entry.remote_id = format!("{}:{data_id}", dataset.id); + Ok(Some(entry)) + } + /// Resolves the dataset assigned to a namespace. async fn find_dataset(&self, namespace: &str) -> anyhow::Result> { let name = Self::dataset_name(namespace); @@ -288,22 +367,6 @@ impl CogneeDialect { .into_iter() .find(|dataset| dataset.name == name)) } - - /// Deletes a stored envelope using its composite remote identifier. - async fn delete_entry(&self, entry: &StoredEntry) -> anyhow::Result<()> { - let (dataset_id, data_id) = entry - .remote_id - .split_once(':') - .ok_or_else(|| anyhow!("Cognee record has no dataset id"))?; - self.client - .empty( - Method::DELETE, - &format!("api/v1/datasets/{dataset_id}/data/{data_id}"), - None, - ) - .await?; - Ok(()) - } } #[async_trait] @@ -313,13 +376,40 @@ impl Dialect for CogneeDialect { COGNEE_DRIVER_ID } + /// One namespace = one dataset: entries scoped without the cross-dataset + /// walk (issue #69). Content lives in the envelopes, so this still pays + /// one raw per record — that is the documented floor, not a regression. + async fn namespace_entries(&self, namespace: &str) -> anyhow::Result> { + match self.find_dataset(namespace).await? { + Some(dataset) => self.dataset_entries(&dataset).await, + None => Ok(Vec::new()), + } + } + + /// Keyed get in three requests — dataset resolve, one listing, one raw — + /// however large the store (issue #69: this replaced 1 + D + N serial + /// requests). + async fn entry(&self, namespace: &str, key: &str) -> anyhow::Result> { + let Some(dataset) = self.find_dataset(namespace).await? else { + return Ok(None); + }; + let Some(data_id) = self.find_data_id(&dataset, key).await? else { + return Ok(None); + }; + self.fetch_entry(&dataset, &data_id, namespace, key).await + } + /// Replaces an existing envelope and uploads the new exact record. async fn upsert(&self, entry: StoredEntry) -> anyhow::Result<()> { - let existing = self - .entries() - .await? - .into_iter() - .find(|item| item.namespace == entry.namespace && item.key == entry.key); + // Through the keyed seam: dataset + listing, no raw fan-out. The + // existing record's ids are all the replace path needs. + let existing = match self.find_dataset(&entry.namespace).await? { + Some(dataset) => self + .find_data_id(&dataset, &entry.key) + .await? + .map(|data_id| (dataset, data_id)), + None => None, + }; let body = serde_json::to_vec(&entry)?; let form = multipart::Form::new().part( "data", @@ -327,11 +417,8 @@ impl Dialect for CogneeDialect { .file_name(Self::filename(&entry.key)) .mime_str("application/json")?, ); - let (method, path, form) = if let Some(existing) = existing { - let (dataset_id, data_id) = existing - .remote_id - .split_once(':') - .ok_or_else(|| anyhow!("Cognee record has no dataset id"))?; + let (method, path, form) = if let Some((dataset, data_id)) = existing { + let dataset_id = dataset.id; ( Method::PATCH, format!("api/v1/update?data_id={data_id}&dataset_id={dataset_id}"), @@ -422,18 +509,21 @@ impl Dialect for CogneeDialect { /// Finds and deletes an exact TinyMemory logical record. async fn delete(&self, namespace: &str, key: &str) -> anyhow::Result { + // Three requests, no raw fan-out (issue #69): the filename match + // resolves the id, and delete needs nothing from the envelope. let Some(dataset) = self.find_dataset(namespace).await? else { return Ok(false); }; - let Some(entry) = self - .dataset_entries(&dataset) - .await? - .into_iter() - .find(|item| item.key == key) - else { + let Some(data_id) = self.find_data_id(&dataset, key).await? else { return Ok(false); }; - self.delete_entry(&entry).await?; + self.client + .empty( + Method::DELETE, + &format!("api/v1/datasets/{}/data/{data_id}", dataset.id), + None, + ) + .await?; Ok(true) } diff --git a/adapters/remote/src/cognee_test.rs b/adapters/remote/src/cognee_test.rs index 5afcc3e4..f13dd4e1 100644 --- a/adapters/remote/src/cognee_test.rs +++ b/adapters/remote/src/cognee_test.rs @@ -20,9 +20,25 @@ use tinymemory_api::{ }; #[derive(Clone, Default)] -struct AppState(Arc>>>); +struct AppState( + Arc>>>, + Arc>, + /// The filename the adapter actually uploaded — served back in the data + /// listing, because the issue #69 keyed path resolves BY that name. The + /// old double hardcoded a name nothing ever read. + Arc>>, +); + +/// Per-route request counters for the issue #69 fan-out assertions. +#[derive(Default, Clone, Copy)] +struct CallCounts { + datasets: usize, + listings: usize, + raws: usize, +} async fn datasets(State(state): State) -> Json { + state.1.lock().expect("counts").datasets += 1; let values = if state.0.lock().expect("state lock").is_some() { vec![json!({ "id": "dataset-1", @@ -34,16 +50,18 @@ async fn datasets(State(state): State) -> Json { Json(Value::Array(values)) } async fn data(State(state): State) -> Json { + state.1.lock().expect("counts").listings += 1; + let name = state.2.lock().expect("name lock").clone(); let values = if state.0.lock().expect("state lock").is_some() { - vec![ - json!({"id": "data-1", "name": "6b6579.tinymemory.json", "created_at": "2026-08-12T00:00:00Z"}), - ] + let name = name.unwrap_or_else(|| "6b6579.tinymemory".to_owned()); + vec![json!({"id": "data-1", "name": name, "created_at": "2026-08-12T00:00:00Z"})] } else { vec![] }; Json(Value::Array(values)) } async fn raw(State(state): State) -> impl IntoResponse { + state.1.lock().expect("counts").raws += 1; state.0.lock().expect("state lock").clone().map_or_else( || (StatusCode::NOT_FOUND, Vec::new()), |body| (StatusCode::OK, body), @@ -52,6 +70,11 @@ async fn raw(State(state): State) -> impl IntoResponse { async fn remember(State(state): State, mut multipart: Multipart) -> StatusCode { while let Some(field) = multipart.next_field().await.expect("multipart") { if field.name() == Some("data") { + if let Some(name) = field.file_name() { + // Cognee's loader strips the final `.json`; mirror it. + *state.2.lock().expect("name lock") = + Some(name.trim_end_matches(".json").to_owned()); + } *state.0.lock().expect("state lock") = Some(field.bytes().await.expect("body").to_vec()); } @@ -150,7 +173,7 @@ async fn native_cognee_round_trips_the_tinymemory_contract() { .route("/api/v1/update", patch(remember)) .route("/api/v1/recall", post(recall)) .route("/health", get(|| async { StatusCode::OK })) - .with_state(state); + .with_state(state.clone()); let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await .expect("bind"); @@ -232,3 +255,62 @@ async fn native_cognee_round_trips_the_tinymemory_contract() { assert!(!driver.forget("project", "key").await.expect("forget again")); assert!(driver.health().await.is_usable()); } + +/// Issue #69: the keyed get is three requests — dataset resolve, one +/// listing, ONE raw — however many records the store holds. The pre-seam +/// path raw-fetched every record in every dataset (1 + D + N), which is what +/// made a 10k-record hosted store cost ~10,002 serial requests per get. And +/// a keyed delete needs no envelope at all: zero raws. +#[tokio::test] +async fn keyed_ops_never_fan_out_over_raw_fetches() { + let state = AppState::default(); + let app = Router::new() + .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)) + .route("/api/v1/remember", post(remember)) + .route("/api/v1/update", patch(remember)) + .with_state(state.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let endpoint = format!("http://{}", listener.local_addr().expect("address")); + tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve"); + }); + let driver = + crate::cognee_provider(super::CogneeMemory::self_hosted(&endpoint, None).expect("client")); + + driver + .store( + "project", + "key", + "knowledge graph", + MemoryCategory::Core, + None, + MemoryTaint::Internal, + ) + .await + .expect("store"); + *state.1.lock().expect("counts") = CallCounts::default(); + + driver + .get("project", "key") + .await + .expect("get") + .expect("entry"); + let counts = *state.1.lock().expect("counts"); + assert_eq!(counts.raws, 1, "exactly one raw fetch per keyed get"); + assert_eq!(counts.listings, 1, "exactly one data listing per keyed get"); + assert!( + counts.datasets <= 1, + "one dataset resolve per keyed get, got {}", + counts.datasets + ); + + *state.1.lock().expect("counts") = CallCounts::default(); + assert!(driver.forget("project", "key").await.expect("forget")); + let counts = *state.1.lock().expect("counts"); + assert_eq!(counts.raws, 0, "a keyed delete reads no envelopes"); +} diff --git a/adapters/remote/src/common.rs b/adapters/remote/src/common.rs index f9ed7d12..4b2035b8 100644 --- a/adapters/remote/src/common.rs +++ b/adapters/remote/src/common.rs @@ -590,6 +590,34 @@ pub(crate) trait Dialect: Send + Sync + std::fmt::Debug { async fn upsert(&self, entry: StoredEntry) -> anyhow::Result<()>; /// Enumerates every record owned by this adapter. async fn entries(&self) -> anyhow::Result>; + /// One namespace's records (issue #69, the keyed-CRUD seam). + /// + /// The default enumerates and filters — exactly what every caller did + /// before the seam existed — so a dialect overrides only when its + /// backend can scope the fetch server-side (Supermemory's container + /// tags; Mem0's entity filters). Callers that genuinely need EVERY + /// record (`count`, `namespace_summaries`, export) stay on `entries`; + /// that full walk is the documented floor, not an accident. + async fn namespace_entries(&self, namespace: &str) -> anyhow::Result> { + Ok(self + .entries() + .await? + .into_iter() + .filter(|entry| entry.namespace == namespace) + .collect()) + } + /// One record by its exact logical key (issue #69). + /// + /// Default: the namespace's records, filtered — which itself defaults to + /// the full walk. A dialect with a true server-side keyed lookup (Mem0 + /// cloud metadata filters) overrides this directly. + async fn entry(&self, namespace: &str, key: &str) -> anyhow::Result> { + Ok(self + .namespace_entries(namespace) + .await? + .into_iter() + .find(|entry| entry.key == key)) + } /// Runs the backend's native recall operation. async fn search( &self, @@ -710,25 +738,33 @@ impl Memory for RemoteMemory { .collect()) } - /// Locates one record by its exact logical namespace and key. + /// Locates one record by its exact logical namespace and key — through + /// the dialect's keyed seam, so a backend that can resolve a key + /// server-side does (issue #69); the default is the old full walk. async fn get(&self, namespace: &str, key: &str) -> anyhow::Result> { Ok(self .dialect - .entries() + .entry(namespace, key) .await? - .into_iter() - .find(|entry| entry.namespace == namespace && entry.key == key) .map(StoredEntry::into_memory_entry)) } /// Enumerates records and applies exact category and session filters. + /// + /// A namespace-scoped list goes through the dialect's namespace seam + /// (issue #69): on a scoping backend that is one tag/entity fetch + /// instead of the whole account. The all-namespaces list has no scope to + /// exploit and stays on the full walk. async fn list( &self, namespace: Option<&str>, category: Option<&MemoryCategory>, session_id: Option<&str>, ) -> anyhow::Result> { - let mut entries = self.dialect.entries().await?; + let mut entries = match namespace { + Some(value) => self.dialect.namespace_entries(value).await?, + None => self.dialect.entries().await?, + }; entries.retain(|entry| { namespace.is_none_or(|value| entry.namespace == value) && category.is_none_or(|value| &entry.category == value) diff --git a/adapters/remote/src/failure_test.rs b/adapters/remote/src/failure_test.rs index ed524a4c..d7c120ce 100644 --- a/adapters/remote/src/failure_test.rs +++ b/adapters/remote/src/failure_test.rs @@ -229,9 +229,10 @@ async fn a_cursor_that_never_clears_is_refused_rather_than_walked_for_ever() { // 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; + // finishes far inside this. `count` is the walker now — issue #69 made + // the keyed `get` a single filtered request, so a poisoned cursor cannot + // spin it any more; the whole-store walk is where the ceiling lives. + let outcome = tokio::time::timeout(std::time::Duration::from_secs(60), memory.count()).await; let Ok(result) = outcome else { panic!("the hosted listing never terminated against a cursor that never clears"); diff --git a/adapters/remote/src/mem0.rs b/adapters/remote/src/mem0.rs index 8cfc839f..e50dac1a 100644 --- a/adapters/remote/src/mem0.rs +++ b/adapters/remote/src/mem0.rs @@ -304,42 +304,48 @@ impl Mem0Dialect { // 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}})), - Attempts::RetryTransient, - ) - .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 \ + Flavour::Cloud => self.cloud_walk(json!({"agent_id": CLOUD_AGENT_ID})).await, + } + } + + /// The hosted platform's paged POST listing, scoped by `filters` (issue + /// #69): the whole-account walk passes the agent filter alone; the + /// namespace-scoped walk ANDs the namespace's entity id in, which the + /// server applies before paging — so the page count scales with the + /// namespace, not the account. + async fn cloud_walk(&self, filters: Value) -> anyhow::Result> { + 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": filters})), + Attempts::RetryTransient, + ) + .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) - } + ); + page = page.saturating_add(1); } + Ok(all) } /// The search body both flavours send. @@ -423,6 +429,24 @@ impl Mem0Dialect { } } +/// Percent-encodes a value for a query-string position (RFC 3986 unreserved +/// set kept verbatim). Namespaces carry `/` and arbitrary user text; a raw +/// interpolation would split the query. Hand-rolled because the crate has no +/// direct `url`/`percent-encoding` dependency and eight lines do not justify +/// one. +fn percent_encode_query(value: &str) -> String { + let mut out = String::with_capacity(value.len()); + for byte in value.bytes() { + match byte { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { + out.push(byte as char); + } + _ => out.push_str(&format!("%{byte:02X}")), + } + } + out +} + #[async_trait] impl Dialect for Mem0Dialect { /// Returns the stable Mem0 driver identifier. @@ -432,11 +456,9 @@ impl Dialect for Mem0Dialect { /// Replaces an existing exact record or creates it with inference disabled. async fn upsert(&self, entry: StoredEntry) -> anyhow::Result<()> { - let existing = self - .entries() - .await? - .into_iter() - .find(|item| item.namespace == entry.namespace && item.key == entry.key); + // Through the keyed seam (issue #69): one filtered request on cloud, + // one namespace-scoped listing self-hosted — not the whole account. + let existing = self.entry(&entry.namespace, &entry.key).await?; let metadata = Self::metadata(&entry); if let Some(existing) = existing { // Both APIs take the same update body; only the path differs. @@ -470,6 +492,111 @@ impl Dialect for Mem0Dialect { Ok(()) } + /// One namespace's records, server-scoped on both flavours (issue #69). + /// + /// Self-hosted: the GET listing filters by `user_id` — the one dimension + /// every vector store supports — so the 1000-row refusal ceiling becomes + /// per-NAMESPACE instead of a whole-store death sentence. Cloud: the + /// paged walk ANDs the namespace's entity id into its mandatory filter. + async fn namespace_entries(&self, namespace: &str) -> anyhow::Result> { + let values = match self.flavour { + Flavour::SelfHosted => { + let top_k = Self::LISTING_TOP_K; + let encoded = percent_encode_query(namespace); + let response: Value = self + .client + .json( + Method::GET, + &format!("memories?top_k={top_k}&user_id={encoded}"), + None, + Attempts::RetryTransient, + ) + .await?; + let results = response + .get("results") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + anyhow::ensure!( + results.len() < top_k, + "mem0 returned {} memories for one namespace, this adapter's unpaginated \ + listing ceiling — refusing rather than answering exact reads wrongly \ + (a record past the window would read as absent)", + results.len() + ); + results + } + Flavour::Cloud => { + self.cloud_walk(json!({"AND": [ + {"agent_id": CLOUD_AGENT_ID}, + {"user_id": namespace}, + ]})) + .await? + } + }; + // Retained client-side even though the server was ASKED to scope: + // a server that ignores an unrecognised filter (an old OSS build, a + // proxy) would otherwise leak sibling namespaces into keyed reads — + // the same verify-don't-trust rule as `entry`. + Ok(values + .iter() + .filter_map(Self::decode) + .filter(|entry| entry.namespace == namespace) + .collect()) + } + + /// One record by key. On the hosted platform this is a single filtered + /// request — the metadata keys this adapter has ALWAYS written (issue + /// #69: `tinymemory_key` is top-level and equality-filtered, inside the + /// platform's documented envelope). Self-hosted inherits the + /// namespace-scoped default: the OSS list route has no metadata param. + /// + /// Verify-after-resolve: the decoded record must actually BE the asked- + /// for one. A server that ignores an unrecognised filter clause would + /// answer with someone else's record, and trusting it silently is how a + /// filter-grammar drift becomes a cross-record read. + async fn entry(&self, namespace: &str, key: &str) -> anyhow::Result> { + if self.flavour != Flavour::Cloud { + return Ok(self + .namespace_entries(namespace) + .await? + .into_iter() + .find(|entry| entry.key == key)); + } + let response: Value = self + .client + .json( + Method::POST, + &format!("v3/memories/?page=1&page_size={CLOUD_PAGE_SIZE}"), + Some(&json!({"filters": {"AND": [ + {"agent_id": CLOUD_AGENT_ID}, + {"user_id": namespace}, + {"metadata": {"tinymemory_key": key}}, + ]}})), + Attempts::RetryTransient, + ) + .await?; + let results = response + .get("results") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + for value in &results { + if let Some(entry) = Self::decode(value) { + anyhow::ensure!( + entry.namespace == namespace && entry.key == key, + "mem0's filtered lookup answered a DIFFERENT record ({}/{}) than asked \ + ({namespace}/{key}) — the server did not honor the metadata filter; \ + refusing rather than serving someone else's memory", + entry.namespace, + entry.key + ); + return Ok(Some(entry)); + } + } + Ok(None) + } + /// Enumerates and decodes TinyMemory-owned Mem0 records. async fn entries(&self) -> anyhow::Result> { Ok(self diff --git a/adapters/remote/src/mem0_test.rs b/adapters/remote/src/mem0_test.rs index 57fa8657..793bb9c2 100644 --- a/adapters/remote/src/mem0_test.rs +++ b/adapters/remote/src/mem0_test.rs @@ -18,10 +18,44 @@ use tinymemory_api::{ }; #[derive(Clone, Default)] -struct AppState(Arc>>); +struct AppState(Arc>>, Arc>>); -async fn list(State(state): State) -> Json { - Json(json!({"results": state.0.lock().expect("state lock").clone()})) +async fn list( + State(state): State, + axum::extract::RawQuery(query): axum::extract::RawQuery, +) -> Json { + state + .1 + .lock() + .expect("query lock") + .push(query.unwrap_or_default()); + // Honour the user_id filter the way the OSS server does (issue #69): a + // scoped request must not receive the whole store back. + let query = state + .1 + .lock() + .expect("query lock") + .last() + .cloned() + .unwrap_or_default(); + let user = query + .split('&') + .find_map(|pair| pair.strip_prefix("user_id=")) + .map(str::to_owned); + let rows = state.0.lock().expect("state lock").clone(); + let rows = match user { + Some(ref encoded) => rows + .into_iter() + .filter(|row| { + row["metadata"]["tinymemory_namespace"] + .as_str() + .map(|ns| super::percent_encode_query(ns) == *encoded) + == Some(true) + }) + .collect(), + None => rows, + }; + Json(json!({"results": rows})) } async fn add(State(state): State, Json(body): Json) -> Json { @@ -140,3 +174,109 @@ async fn native_mem0_round_trips_the_tinymemory_contract() { .expect("forget again")); assert!(driver.health().await.is_usable()); } + +/// Issue #69: a self-hosted keyed read scopes the listing to the namespace's +/// `user_id` — percent-encoded, since namespaces carry slashes — instead of +/// walking the whole store. +#[tokio::test] +async fn self_hosted_keyed_reads_scope_by_user_id() { + let state = AppState::default(); + let app = Router::new() + .route("/memories", get(list).post(add)) + .route("/memories/{id}", put(update).delete(remove)) + .with_state(state.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let endpoint = format!("http://{}", listener.local_addr().expect("address")); + tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve"); + }); + let driver = crate::mem0_provider( + super::Mem0Memory::self_hosted(&endpoint, Some("token")).expect("client"), + ); + driver + .store( + "oc/team a", + "decision", + "content", + MemoryCategory::Core, + None, + MemoryTaint::Internal, + ) + .await + .expect("store"); + state.1.lock().expect("query lock").clear(); + + let got = driver.get("oc/team a", "decision").await.expect("get"); + assert!( + got.is_some(), + "the scoped listing must still find the record" + ); + let queries = state.1.lock().expect("query lock").clone(); + assert_eq!(queries.len(), 1, "one scoped request: {queries:?}"); + assert!( + queries[0].contains("user_id=oc%2Fteam%20a"), + "the namespace rides percent-encoded: {queries:?}" + ); +} + +/// Issue #69: the hosted platform's keyed lookup is ONE filtered request +/// carrying the metadata key — and verify-after-resolve refuses a server +/// that answers with someone else's record instead of honoring the filter. +#[tokio::test] +async fn cloud_keyed_lookup_filters_by_metadata_and_verifies() { + use axum::routing::post; + let bodies: Arc>> = Arc::default(); + let answer: Arc> = Arc::default(); + let captured = bodies.clone(); + let served = answer.clone(); + let app = Router::new().route( + "/v3/memories/", + post(move |Json(body): Json| { + let captured = captured.clone(); + let served = served.clone(); + async move { + captured.lock().expect("bodies").push(body); + Json(json!({"results": [served.lock().expect("answer").clone()], "next": null})) + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let endpoint = format!("http://{}", listener.local_addr().expect("address")); + tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve"); + }); + let record = |ns: &str, key: &str| { + json!({"id": "mem-1", "memory": "content", "metadata": { + "tinymemory_namespace": ns, + "tinymemory_key": key, + "tinymemory_category": "core", + "tinymemory_taint": "internal", + }}) + }; + + let driver = crate::mem0_provider(super::Mem0Memory::api(&endpoint, "key").expect("client")); + *answer.lock().expect("answer") = record("project", "decision"); + let got = driver.get("project", "decision").await.expect("get"); + assert!(got.is_some()); + let sent = bodies.lock().expect("bodies").clone(); + assert_eq!(sent.len(), 1, "one filtered request, no account walk"); + let clauses = sent[0]["filters"]["AND"].as_array().expect("AND clauses"); + assert!( + clauses + .iter() + .any(|c| c["metadata"]["tinymemory_key"] == json!("decision")), + "the filter carries the key: {sent:?}" + ); + + // A server ignoring the filter answers a DIFFERENT record: refuse loudly. + *answer.lock().expect("answer") = record("project", "someone-elses"); + let err = driver.get("project", "decision").await; + assert!( + err.is_err(), + "a mismatched filtered answer must refuse, not serve another record" + ); +} diff --git a/adapters/remote/src/supermemory.rs b/adapters/remote/src/supermemory.rs index f2eab3a4..e1daeb32 100644 --- a/adapters/remote/src/supermemory.rs +++ b/adapters/remote/src/supermemory.rs @@ -384,6 +384,23 @@ impl Dialect for SupermemoryDialect { } /// Enumerates TinyMemory-owned Supermemory records. + /// Issue #69: one tag's records instead of the whole account — the + /// scoped fetch `find_entry`/`delete` always used, now serving reads. + async fn namespace_entries(&self, namespace: &str) -> anyhow::Result> { + // Retained client-side even though the tag scopes it server-side: a + // server ignoring the tag filter must not leak sibling namespaces. + Ok(self + .memories_in_tag(&Self::container_tag(namespace)) + .await? + .into_iter() + .filter(|entry| entry.namespace == namespace) + .collect()) + } + + async fn entry(&self, namespace: &str, key: &str) -> anyhow::Result> { + self.find_entry(namespace, key).await + } + async fn entries(&self) -> anyhow::Result> { self.memories().await } diff --git a/adapters/remote/src/supermemory_test.rs b/adapters/remote/src/supermemory_test.rs index d72a6b2f..3204b0b0 100644 --- a/adapters/remote/src/supermemory_test.rs +++ b/adapters/remote/src/supermemory_test.rs @@ -23,6 +23,9 @@ struct Fixture { records: Vec, container_tags: Vec, last_search_tag: Option, + /// Every /v4/memories/list request body, for the issue #69 scoping + /// assertions: a namespace-scoped read must ask for ONE tag. + list_bodies: Vec, } #[derive(Clone, Default)] @@ -39,8 +42,9 @@ async fn tags(State(state): State) -> Json { )) } -async fn list(State(state): State) -> Json { - let fixture = state.0.lock().expect("state lock"); +async fn list(State(state): State, Json(body): Json) -> Json { + let mut fixture = state.0.lock().expect("state lock"); + fixture.list_bodies.push(body); Json(json!({"memoryEntries": fixture.records, "pagination": {"totalPages": 1}})) } async fn add(State(state): State, Json(body): Json) -> Json { @@ -261,3 +265,48 @@ async fn native_supermemory_round_trips_the_tinymemory_contract() { assert!(driver.forget("project", "decision").await.expect("forget")); assert!(driver.health().await.is_usable()); } + +/// Issue #69: a keyed read asks the backend for ONE namespace's tag — never +/// the whole-account tag walk the pre-seam reads ran. +#[tokio::test] +async fn keyed_reads_scope_to_one_container_tag() { + let state = AppState::default(); + let app = Router::new() + .route("/v3/container-tags/list", get(tags)) + .route("/v4/memories/list", post(list)) + .route("/v4/memories", post(add).patch(update).delete(remove)) + .route("/", get(|| async { StatusCode::OK })) + .with_state(state.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let endpoint = format!("http://{}", listener.local_addr().expect("address")); + tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve"); + }); + let driver = crate::supermemory_provider( + super::SupermemoryMemory::self_hosted(&endpoint, "secret").expect("client"), + ); + driver + .store( + "project", + "decision", + "use Rust 2024", + MemoryCategory::Core, + None, + MemoryTaint::Internal, + ) + .await + .expect("store"); + state.0.lock().expect("state lock").list_bodies.clear(); + + driver.get("project", "decision").await.expect("get"); + let bodies = state.0.lock().expect("state lock").list_bodies.clone(); + assert_eq!(bodies.len(), 1, "one scoped list request, not a tag walk"); + let expected = super::SupermemoryDialect::container_tag("project"); + assert_eq!( + bodies[0]["containerTags"], + serde_json::json!([expected]), + "the request names exactly the namespace's tag" + ); +} From 4401ca4d1e7e2b837df1948f90ec0075e6b95c02 Mon Sep 17 00:00:00 2001 From: Shanu Date: Fri, 21 Aug 2026 00:23:39 +0530 Subject: [PATCH 2/2] =?UTF-8?q?adapters/remote:=20review=20fixes=20?= =?UTF-8?q?=E2=80=94=20scan=20the=20whole=20degraded=20page=20(#69)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mem0's cloud keyed lookup now scans every decoded result for the exact (namespace, key) before refusing: a server that ignores the metadata clause answers with the namespace's page, and the asked-for record may sit anywhere in it. All-foreign pages still refuse — an honored filter makes every result match, so foreign-only can only mean filter drift. The test double serves a results array and pins the sibling-ahead case. Also from review: supermemory keyed-seam docs corrected, cognee delete's filename-trust decision recorded next to the code, and the mem0 test handler captures its query once instead of re-locking for last(). --- adapters/remote/src/cognee.rs | 5 +++++ adapters/remote/src/mem0.rs | 29 ++++++++++++++++-------- adapters/remote/src/mem0_test.rs | 36 +++++++++++++++++------------- adapters/remote/src/supermemory.rs | 4 +++- 4 files changed, 48 insertions(+), 26 deletions(-) diff --git a/adapters/remote/src/cognee.rs b/adapters/remote/src/cognee.rs index 0d86525a..4f561ad9 100644 --- a/adapters/remote/src/cognee.rs +++ b/adapters/remote/src/cognee.rs @@ -511,6 +511,11 @@ impl Dialect for CogneeDialect { async fn delete(&self, namespace: &str, key: &str) -> anyhow::Result { // Three requests, no raw fan-out (issue #69): the filename match // resolves the id, and delete needs nothing from the envelope. + // Deliberately weaker than `fetch_entry`'s envelope check: the + // filename is the key's SHA-256 digest and the dataset scopes the + // namespace, so a wrong-record match would need a digest collision — + // and verifying the envelope would cost exactly the raw fetch this + // path exists to avoid. let Some(dataset) = self.find_dataset(namespace).await? else { return Ok(false); }; diff --git a/adapters/remote/src/mem0.rs b/adapters/remote/src/mem0.rs index e50dac1a..bf3eff1c 100644 --- a/adapters/remote/src/mem0.rs +++ b/adapters/remote/src/mem0.rs @@ -581,19 +581,30 @@ impl Dialect for Mem0Dialect { .and_then(Value::as_array) .cloned() .unwrap_or_default(); + // Scan the whole page before judging it: a server that ignores the + // metadata clause answers with the namespace's records, and the one + // asked for may sit anywhere in that page. An exact match anywhere + // wins. Decoded records with no match can only mean the filter was + // not honored — an honored filter makes every result match — so + // refuse rather than serve someone else's memory. + let mut foreign: Option = None; for value in &results { if let Some(entry) = Self::decode(value) { - anyhow::ensure!( - entry.namespace == namespace && entry.key == key, - "mem0's filtered lookup answered a DIFFERENT record ({}/{}) than asked \ - ({namespace}/{key}) — the server did not honor the metadata filter; \ - refusing rather than serving someone else's memory", - entry.namespace, - entry.key - ); - return Ok(Some(entry)); + if entry.namespace == namespace && entry.key == key { + return Ok(Some(entry)); + } + foreign.get_or_insert(entry); } } + if let Some(entry) = foreign { + anyhow::bail!( + "mem0's filtered lookup answered a DIFFERENT record ({}/{}) than asked \ + ({namespace}/{key}) — the server did not honor the metadata filter; \ + refusing rather than serving someone else's memory", + entry.namespace, + entry.key + ); + } Ok(None) } diff --git a/adapters/remote/src/mem0_test.rs b/adapters/remote/src/mem0_test.rs index 793bb9c2..4be548f1 100644 --- a/adapters/remote/src/mem0_test.rs +++ b/adapters/remote/src/mem0_test.rs @@ -24,20 +24,10 @@ async fn list( State(state): State, axum::extract::RawQuery(query): axum::extract::RawQuery, ) -> Json { - state - .1 - .lock() - .expect("query lock") - .push(query.unwrap_or_default()); + let query = query.unwrap_or_default(); + state.1.lock().expect("query lock").push(query.clone()); // Honour the user_id filter the way the OSS server does (issue #69): a // scoped request must not receive the whole store back. - let query = state - .1 - .lock() - .expect("query lock") - .last() - .cloned() - .unwrap_or_default(); let user = query .split('&') .find_map(|pair| pair.strip_prefix("user_id=")) @@ -238,7 +228,7 @@ async fn cloud_keyed_lookup_filters_by_metadata_and_verifies() { let served = served.clone(); async move { captured.lock().expect("bodies").push(body); - Json(json!({"results": [served.lock().expect("answer").clone()], "next": null})) + Json(json!({"results": served.lock().expect("answer").clone(), "next": null})) } }), ); @@ -259,7 +249,7 @@ async fn cloud_keyed_lookup_filters_by_metadata_and_verifies() { }; let driver = crate::mem0_provider(super::Mem0Memory::api(&endpoint, "key").expect("client")); - *answer.lock().expect("answer") = record("project", "decision"); + *answer.lock().expect("answer") = json!([record("project", "decision")]); let got = driver.get("project", "decision").await.expect("get"); assert!(got.is_some()); let sent = bodies.lock().expect("bodies").clone(); @@ -272,8 +262,22 @@ async fn cloud_keyed_lookup_filters_by_metadata_and_verifies() { "the filter carries the key: {sent:?}" ); - // A server ignoring the filter answers a DIFFERENT record: refuse loudly. - *answer.lock().expect("answer") = record("project", "someone-elses"); + // A server that ignores the metadata clause answers with the whole + // namespace: the asked-for record must still resolve even when a sibling + // rides ahead of it in the page. + *answer.lock().expect("answer") = json!([ + record("project", "someone-elses"), + record("project", "decision"), + ]); + let got = driver + .get("project", "decision") + .await + .expect("degraded-filter get") + .expect("record present in the degraded page"); + assert_eq!(got.key, "decision", "the exact match wins, not the sibling"); + + // A server answering ONLY foreign records: refuse loudly. + *answer.lock().expect("answer") = json!([record("project", "someone-elses")]); let err = driver.get("project", "decision").await; assert!( err.is_err(), diff --git a/adapters/remote/src/supermemory.rs b/adapters/remote/src/supermemory.rs index e1daeb32..65aee589 100644 --- a/adapters/remote/src/supermemory.rs +++ b/adapters/remote/src/supermemory.rs @@ -383,7 +383,7 @@ impl Dialect for SupermemoryDialect { Ok(()) } - /// Enumerates TinyMemory-owned Supermemory records. + /// Enumerates one namespace's TinyMemory-owned Supermemory records. /// Issue #69: one tag's records instead of the whole account — the /// scoped fetch `find_entry`/`delete` always used, now serving reads. async fn namespace_entries(&self, namespace: &str) -> anyhow::Result> { @@ -397,6 +397,8 @@ impl Dialect for SupermemoryDialect { .collect()) } + /// One record by key — `find_entry` already pages only this namespace's + /// container tag, so the keyed seam has nothing to add. async fn entry(&self, namespace: &str, key: &str) -> anyhow::Result> { self.find_entry(namespace, key).await }