diff --git a/adapters/remote/src/cognee.rs b/adapters/remote/src/cognee.rs index f91f6b3..4f561ad 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,26 @@ 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. + // 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); }; - 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 5afcc3e..f13dd4e 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 f9ed7d1..4b2035b 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 ed524a4..d7c120c 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 8cfc839..bf3eff1 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,122 @@ 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(); + // 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) { + 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) + } + /// 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 57fa865..4be548f 100644 --- a/adapters/remote/src/mem0_test.rs +++ b/adapters/remote/src/mem0_test.rs @@ -18,10 +18,34 @@ 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 { + 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 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 +164,123 @@ 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") = json!([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 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(), + "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 f2eab3a..65aee58 100644 --- a/adapters/remote/src/supermemory.rs +++ b/adapters/remote/src/supermemory.rs @@ -383,7 +383,26 @@ 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> { + // 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()) + } + + /// 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 + } + 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 d72a6b2..3204b0b 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" + ); +}