diff --git a/crates/tinymemory-remote/src/cognee.rs b/crates/tinymemory-remote/src/cognee.rs index 4f561ad..1bb7789 100644 --- a/crates/tinymemory-remote/src/cognee.rs +++ b/crates/tinymemory-remote/src/cognee.rs @@ -1,6 +1,6 @@ //! Self-hosted Cognee REST adapter. -use anyhow::{anyhow, Context}; +use anyhow::Context; use async_trait::async_trait; use reqwest::{multipart, Method}; use serde_json::{json, Value}; @@ -265,14 +265,7 @@ impl CogneeDialect { serde_json::from_str(&raw).context("Cognee record envelope is invalid")?; entry.remote_id = format!("{}:{id}", dataset.id); if entry.timestamp.is_empty() { - entry.timestamp = data - .get("updatedAt") - .or_else(|| data.get("updated_at")) - .or_else(|| data.get("createdAt")) - .or_else(|| data.get("created_at")) - .and_then(Value::as_str) - .unwrap_or_default() - .to_owned(); + entry.timestamp = Self::listing_timestamp(data); } entries.push(entry); } @@ -285,7 +278,21 @@ impl CogneeDialect { /// 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> { + /// The listing row's write time. Checked per candidate with `find_map`, + /// NOT an `or_else` chain over `Value::get`: real Cognee serializes + /// `"updatedAt": null` for every never-updated record, and `get` on a + /// present-but-null key answers `Some(Null)` — an `or_else` chain commits + /// to it and never reaches `createdAt`, emptying every timestamp + /// (issue #75). + fn listing_timestamp(data: &Value) -> String { + ["updatedAt", "updated_at", "createdAt", "created_at"] + .iter() + .find_map(|key| data.get(key).and_then(Value::as_str)) + .unwrap_or_default() + .to_owned() + } + + async fn data_index(&self, dataset: &Dataset) -> anyhow::Result> { let response: Value = self .client .json( @@ -303,6 +310,12 @@ impl CogneeDialect { Some(( data.get("id")?.as_str()?.to_owned(), data.get("name")?.as_str()?.to_owned(), + // Kept alongside the id: the envelope this adapter + // uploads carries an empty timestamp, so the listing is + // the ONLY source a keyed fetch can backfill from (the + // enumeration path already does — the keyed path must + // agree with it). + Self::listing_timestamp(data), )) }) .collect()) @@ -313,16 +326,23 @@ impl CogneeDialect { /// 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> { + 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(); + let stripped = uploaded + .strip_suffix(".json") + .unwrap_or(&uploaded) + .to_owned(); Ok(self .data_index(dataset) .await? .into_iter() .rev() - .find(|(_, name)| *name == uploaded || *name == stripped) - .map(|(id, _)| id)) + .find(|(_, name, _)| *name == uploaded || *name == stripped) + .map(|(id, _, timestamp)| (id, timestamp))) } /// Downloads and decodes ONE envelope by its ids, verifying it is the @@ -333,6 +353,7 @@ impl CogneeDialect { &self, dataset: &Dataset, data_id: &str, + listing_timestamp: &str, namespace: &str, key: &str, ) -> anyhow::Result> { @@ -355,6 +376,13 @@ impl CogneeDialect { ); } entry.remote_id = format!("{}:{data_id}", dataset.id); + // The uploaded envelope's timestamp is empty by construction + // (StoredEntry::new), so without this backfill every keyed get would + // answer an empty timestamp while the enumeration path answers the + // listing's — the same record disagreeing with itself. + if entry.timestamp.is_empty() { + entry.timestamp = listing_timestamp.to_owned(); + } Ok(Some(entry)) } @@ -393,21 +421,29 @@ impl Dialect for CogneeDialect { let Some(dataset) = self.find_dataset(namespace).await? else { return Ok(None); }; - let Some(data_id) = self.find_data_id(&dataset, key).await? else { + let Some((data_id, listing_timestamp)) = self.find_data_id(&dataset, key).await? else { return Ok(None); }; - self.fetch_entry(&dataset, &data_id, namespace, key).await + self.fetch_entry(&dataset, &data_id, &listing_timestamp, namespace, key) + .await } /// Replaces an existing envelope and uploads the new exact record. async fn upsert(&self, entry: StoredEntry) -> anyhow::Result<()> { // Through the keyed seam: dataset + listing, no raw fan-out. The - // existing record's ids are all the replace path needs. + // existing record's ids are all the replace path needs. Like delete, + // the PATCH trusts the dataset-scoped filename match without an + // envelope read — the name is the key's SHA-256 digest, so a wrong + // target needs a digest collision, and what a collision would cost + // here is an overwrite of the colliding record's content with THIS + // key's envelope (recoverable by that record's next upsert, unlike + // delete's unrecoverable removal — which is the sharper case and got + // this same argument first). 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)), + .map(|(data_id, _)| (dataset, data_id)), None => None, }; let body = serde_json::to_vec(&entry)?; @@ -432,19 +468,7 @@ impl Dialect for CogneeDialect { .text("run_in_background", "false"), ) }; - let response = self - .client - .multipart(method, &path)? - .multipart(form) - .send() - .await?; - if !response.status().is_success() { - return Err(anyhow!( - "memory API {path} returned HTTP {}", - response.status() - )); - } - Ok(()) + self.client.send_multipart(method, &path, form).await } /// Enumerates records across all TinyMemory-owned Cognee datasets. @@ -463,7 +487,18 @@ impl Dialect for CogneeDialect { limit: usize, opts: RecallOpts<'_>, ) -> anyhow::Result> { - let datasets = opts.namespace.map(Self::dataset_name); + // Resolve the dataset BEFORE asking recall (issue #75): real Cognee + // 404s ("No datasets found") when a named dataset resolves to + // nothing, so the first recall in a fresh namespace — before its + // first store — errored where every sibling op answers empty. One + // extra listing request, the same price entry()/delete() pay. + let datasets = match opts.namespace { + Some(namespace) => match self.find_dataset(namespace).await? { + Some(dataset) => Some(vec![dataset.name]), + None => return Ok(Vec::new()), + }, + None => None, + }; let response: Value = self .client .json( @@ -472,7 +507,7 @@ impl Dialect for CogneeDialect { Some(&json!({ "query": query, "search_type": "CHUNKS", - "datasets": datasets.map(|name| vec![name]), + "datasets": datasets, "top_k": limit, "only_context": true, "session_id": opts.session_id @@ -519,7 +554,7 @@ impl Dialect for CogneeDialect { let Some(dataset) = self.find_dataset(namespace).await? else { return Ok(false); }; - let Some(data_id) = self.find_data_id(&dataset, key).await? else { + let Some((data_id, _)) = self.find_data_id(&dataset, key).await? else { return Ok(false); }; self.client diff --git a/crates/tinymemory-remote/src/cognee_test.rs b/crates/tinymemory-remote/src/cognee_test.rs index b8cc420..73584e4 100644 --- a/crates/tinymemory-remote/src/cognee_test.rs +++ b/crates/tinymemory-remote/src/cognee_test.rs @@ -54,7 +54,15 @@ async fn data(State(state): State) -> Json { let name = state.2.lock().expect("name lock").clone(); let values = if state.0.lock().expect("state lock").is_some() { let name = name.unwrap_or_else(|| "6b6579.tinymemory".to_owned()); - vec![json!({"id": "data-1", "name": name, "created_at": "2026-08-12T00:00:00Z"})] + // `updatedAt` present-but-null is what real Cognee serializes for a + // never-updated record: the backfill must fall through to + // `created_at` instead of committing to the null (issue #75). + vec![json!({ + "id": "data-1", + "name": name, + "updatedAt": null, + "created_at": "2026-08-12T00:00:00Z" + })] } else { vec![] }; @@ -266,6 +274,23 @@ async fn native_cognee_round_trips_the_tinymemory_contract() { .expect("entry"); assert_eq!(entry.content, "updated knowledge graph"); assert_eq!(entry.taint, MemoryTaint::ExternalSync); + // The uploaded envelope's timestamp is empty by construction, so the + // keyed fetch must backfill from the listing the way the enumeration + // path always did — get and list answering different timestamps for the + // same record was the #71-review regression, and a null `updatedAt` + // halting the fallback chain was issue #75's. + assert_eq!( + entry.timestamp, "2026-08-12T00:00:00Z", + "keyed get backfills the listing timestamp past the null updatedAt" + ); + let listed = driver + .list(Some("project"), None, None) + .await + .expect("list"); + assert_eq!( + listed[0].timestamp, entry.timestamp, + "keyed get and namespace list agree on the timestamp" + ); assert_eq!( driver .recall( @@ -366,3 +391,103 @@ async fn keyed_ops_never_fan_out_over_raw_fetches() { let counts = *state.1.lock().expect("counts"); assert_eq!(counts.raws, 0, "a keyed delete reads no envelopes"); } + +/// The envelope-over-filename trust boundary, pinned the way its mem0 twin +/// is. A file whose deterministic name matches the asked key but whose +/// envelope names a DIFFERENT record (hash collision, foreign file wearing +/// our extension) must refuse — never serve someone else's memory. +#[tokio::test] +async fn a_filename_match_with_a_foreign_envelope_is_refused() { + use crate::common::StoredEntry; + + let foreign = serde_json::to_vec(&StoredEntry::new( + "someone-elses-ns", + "decision", + "not yours", + MemoryCategory::Conversation, + None, + MemoryTaint::Internal, + )) + .expect("envelope"); + let name = super::CogneeDialect::filename("decision"); + let app = Router::new() + .route( + "/api/v1/datasets/", + get(|| async { + Json(json!([{ + "id": "dataset-1", + "name": super::CogneeDialect::dataset_name("project") + }])) + }), + ) + .route( + "/api/v1/datasets/{dataset}/data", + get(move || { + let name = name.clone(); + async move { + Json(json!([{ + "id": "data-1", + "name": name, + "created_at": "2026-08-12T00:00:00Z" + }])) + } + }), + ) + .route( + "/api/v1/datasets/{dataset}/data/{data}/raw", + get(move || { + let body = foreign.clone(); + async move { (StatusCode::OK, body) } + }), + ); + 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")); + let err = driver.get("project", "decision").await; + let message = format!("{:?}", err.expect_err("mismatched envelope must refuse")); + assert!( + message.contains("envelope names"), + "the refusal names the mismatch: {message}" + ); +} + +/// Issue #75: recall in a namespace that has never stored anything answers +/// EMPTY, like every sibling op — real Cognee 404s a recall naming a dataset +/// that resolves to nothing. The double serves no /api/v1/recall route at +/// all, so this also proves the guard short-circuits before asking. +#[tokio::test] +async fn recall_in_a_fresh_namespace_is_empty_not_an_error() { + use tinymemory_api::recall::OwnedRecallOpts; + + let app = Router::new().route("/api/v1/datasets/", get(|| async { Json(json!([])) })); + 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")); + let hits = driver + .recall( + "anything", + 3, + &OwnedRecallOpts { + namespace: Some("never-stored".into()), + ..OwnedRecallOpts::default() + }, + None, + ) + .await + .expect("a fresh namespace recalls empty, not an error"); + assert!(hits.is_empty()); +} diff --git a/crates/tinymemory-remote/src/common.rs b/crates/tinymemory-remote/src/common.rs index 4b2035b..6bae433 100644 --- a/crates/tinymemory-remote/src/common.rs +++ b/crates/tinymemory-remote/src/common.rs @@ -67,6 +67,31 @@ const MAX_RESPONSE_BYTES: u64 = 64 * 1024 * 1024; /// enforces the cap while the bytes arrive. Same argument, and same shape, as /// `tinymemory-sources`' `read_body_capped` -- that guard was written for the /// web-page reader and simply had not been applied on this path. +/// Error bodies get a far smaller cap: [`status_error`] surfaces ~300 chars, +/// so 64 KiB preserves every message any API writes while denying a hostile +/// endpoint the unbounded buffer `Response::text()` would hand it — the exact +/// threat [`MAX_RESPONSE_BYTES`] names, which the error paths had skipped +/// (issue #75). Truncation is silent by design: an error body is diagnostic +/// text, not data. +const MAX_ERROR_BODY_BYTES: usize = 64 * 1024; + +/// Reads at most [`MAX_ERROR_BODY_BYTES`] of a non-success body, never +/// failing: the caller is already about to return the status error, and a +/// body-read fault must not mask it. +async fn read_error_body(response: reqwest::Response) -> String { + use futures::StreamExt; + let mut body = Vec::new(); + let mut stream = response.bytes_stream(); + while let Some(Ok(chunk)) = stream.next().await { + let room = MAX_ERROR_BODY_BYTES.saturating_sub(body.len()); + body.extend_from_slice(&chunk[..chunk.len().min(room)]); + if body.len() >= MAX_ERROR_BODY_BYTES { + break; + } + } + String::from_utf8_lossy(&body).into_owned() +} + async fn read_capped(response: reqwest::Response, path: &str) -> anyhow::Result> { use futures::StreamExt; if let Some(len) = response.content_length() { @@ -409,7 +434,7 @@ impl HttpClient { .map_err(|error| self.transport_error(error))?; let status = response.status(); if !status.is_success() { - let body = response.text().await.unwrap_or_default(); + let body = read_error_body(response).await; return Err(self.status_error(path, status, &body)); } let body = read_capped(response, path).await?; @@ -432,7 +457,7 @@ impl HttpClient { .map_err(|error| self.transport_error(error))?; let status = response.status(); if !status.is_success() { - let body = response.text().await.unwrap_or_default(); + let body = read_error_body(response).await; return Err(self.status_error(path, status, &body)); } let body = read_capped(response, path).await?; @@ -461,7 +486,7 @@ impl HttpClient { .map_err(|error| self.transport_error(error))?; let status = response.status(); if !status.is_success() { - let body = response.text().await.unwrap_or_default(); + let body = read_error_body(response).await; return Err(self.status_error(path, status, &body)); } Ok(status) @@ -472,6 +497,33 @@ impl HttpClient { self.request(method, path) } + /// Sends a prepared multipart form and types its failures like every + /// other path: transport faults through [`Self::transport_error`], + /// non-success statuses through [`Self::status_error`]. The upload leg + /// previously spoke raw `anyhow!` strings, so a 400 refusal reached + /// callers as `Other` instead of `Invalid`, a 401 was not `Unauthorized`, + /// and a 429/503 was never retried-or-classified `Unavailable` — the one + /// write path outside the §A4 taxonomy (issue #75). + pub(crate) async fn send_multipart( + &self, + method: Method, + path: &str, + form: reqwest::multipart::Form, + ) -> anyhow::Result<()> { + let response = self + .multipart(method, path)? + .multipart(form) + .send() + .await + .map_err(|error| self.transport_error(error))?; + let status = response.status(); + if !status.is_success() { + let body = read_error_body(response).await; + return Err(self.status_error(path, status, &body)); + } + Ok(()) + } + /// Probes a GET endpoint and reports WHY it failed, typed (issue #18 /// follow-up U4). The boolean `healthy` below discards status, body and /// transport class; this keeps them, so a health surface can distinguish @@ -484,7 +536,7 @@ impl HttpClient { .map_err(|error| self.transport_error(error))?; let status = response.status(); if !status.is_success() { - let body = response.text().await.unwrap_or_default(); + let body = read_error_body(response).await; return Err(self.status_error(path, status, &body)); } Ok(()) diff --git a/crates/tinymemory-remote/src/conformance_test.rs b/crates/tinymemory-remote/src/conformance_test.rs index 6d5dd74..4f3df53 100644 --- a/crates/tinymemory-remote/src/conformance_test.rs +++ b/crates/tinymemory-remote/src/conformance_test.rs @@ -280,10 +280,23 @@ async fn sm_create( Ok(Json(json!({ "memories": [{ "id": id }] }))) } -async fn sm_update(State(store): State, Json(body): Json) -> Json { +async fn sm_update( + State(store): State, + Json(body): Json, +) -> Result, axum::http::StatusCode> { + // Mirrors `sm_create`: the real PATCH requires `containerTag` and 400s + // without it (issue #75) — and the value must match the row it scopes: a + // foreign tag on a PATCH is a cross-container write, not a detail. + let Some(sent_tag) = body["containerTag"].as_str().filter(|tag| !tag.is_empty()) else { + return Err(axum::http::StatusCode::BAD_REQUEST); + }; + let sent_tag = sent_tag.to_owned(); let mut store = store.lock().expect("store lock"); let id = body["id"].as_str().unwrap_or_default().to_owned(); if let Some(row) = store.rows.get_mut(&id) { + if row.tag != sent_tag { + return Err(axum::http::StatusCode::BAD_REQUEST); + } if let Some(text) = body["newContent"].as_str() { row.content = text.to_owned(); } @@ -291,7 +304,7 @@ async fn sm_update(State(store): State, Json(body): Json) -> Json< row.metadata = body["metadata"].clone(); } } - Json(json!({ "id": id })) + Ok(Json(json!({ "id": id }))) } async fn sm_delete(State(store): State, Json(body): Json) -> Json { diff --git a/crates/tinymemory-remote/src/failure_test.rs b/crates/tinymemory-remote/src/failure_test.rs index d7c120c..b8a9c91 100644 --- a/crates/tinymemory-remote/src/failure_test.rs +++ b/crates/tinymemory-remote/src/failure_test.rs @@ -244,6 +244,77 @@ async fn a_cursor_that_never_clears_is_refused_rather_than_walked_for_ever() { ); } +/// Supermemory's twin of the guard above (issue #75): its per-tag pager +/// loops on server-supplied `totalPages`, which is exactly as +/// server-controlled as mem0's cursor — a value that never lets the walk +/// finish must be refused, not walked for ever. +#[tokio::test] +async fn a_total_pages_that_never_lets_the_walk_finish_is_refused() { + use axum::routing::{get, post}; + let app = Router::new() + .route( + "/v3/container-tags/list", + get(|| async { axum::Json(serde_json::json!([{"containerTag": "tinymemory:tm_x"}])) }), + ) + .route( + "/v4/memories/list", + post(|| async { + axum::Json(serde_json::json!({ + "memoryEntries": [{"id": "sm-1", "memory": "x", "metadata": {}}], + "pagination": {"totalPages": 1_000_000} + })) + }), + ); + let endpoint = serve(app).await; + let memory = SupermemoryMemory::api(&endpoint, "sm-test-key").expect("client"); + + let outcome = tokio::time::timeout(std::time::Duration::from_secs(60), memory.count()).await; + let Ok(result) = outcome else { + panic!("the per-tag walk never terminated against a lying totalPages"); + }; + let error = result.expect_err("a totalPages that never clears cannot be answered correctly"); + assert!( + format!("{error:#}").contains("pages"), + "the refusal must name the page ceiling it hit, got: {error:#}" + ); +} + +/// Issue #75: non-2xx bodies are read through the 64 KiB error-body cap, not +/// `Response::text()`'s unbounded buffer. A hostile endpoint answering every +/// request with a 500 and a body that never ends must cost a bounded read and +/// a prompt typed error — not a buffer that grows until the process dies. +#[tokio::test] +async fn an_endless_error_body_is_capped_rather_than_buffered() { + use axum::body::Body; + use axum::http::Response; + use futures::stream; + + let app = Router::new().fallback(any(|| async { + let endless = stream::repeat_with(|| { + Ok::<_, std::convert::Infallible>(axum::body::Bytes::from_static(&[b'x'; 8192])) + }); + Response::builder() + .status(StatusCode::INTERNAL_SERVER_ERROR) + .body(Body::from_stream(endless)) + .expect("response") + })); + let endpoint = serve(app).await; + let memory = Mem0Memory::self_hosted(&endpoint, None).expect("client"); + + // Bounded: with the cap, the read stops at 64 KiB and the typed error + // surfaces immediately; reverting to `text()` hangs here accumulating + // the stream until timeout or OOM. + let outcome = tokio::time::timeout(std::time::Duration::from_secs(30), memory.count()).await; + let Ok(result) = outcome else { + panic!("an endless error body was buffered instead of capped"); + }; + let error = result.expect_err("a 500 must surface as an error"); + assert!( + format!("{error:#}").contains("500"), + "the status error surfaces despite the endless body: {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 @@ -389,6 +460,51 @@ async fn a_400_refusal_is_invalid_not_backend() { } } +/// Issue #75: the multipart UPLOAD leg speaks the same taxonomy. The generic +/// 400 test above never reaches it — cognee's preceding dataset GET fails +/// first against a fail-everything double — so this double lets the resolve +/// succeed and fails only the upload, pinning the one write path that used +/// to answer an untyped string. +#[tokio::test] +async fn a_400_on_the_multipart_upload_leg_is_invalid_not_other() { + use axum::routing::{get, post}; + let app = Router::new() + .route( + "/api/v1/datasets/", + get(|| async { axum::Json(serde_json::json!([])) }), + ) + .route( + "/api/v1/remember", + post(|| async { + ( + StatusCode::BAD_REQUEST, + r#"{"error":"file rejected"}"#.to_owned(), + ) + }), + ); + let endpoint = serve(app).await; + let memory = CogneeMemory::self_hosted(&endpoint, None).expect("client"); + + let error = memory + .store_with_taint( + "ns", + "k", + "content", + MemoryCategory::Core, + None, + MemoryTaint::Internal, + ) + .await + .expect_err("the upload 400 must surface"); + assert!( + matches!( + error.downcast_ref::(), + Some(MemoryError::Invalid(_)) + ), + "a multipart 400 must be Invalid, got: {error}" + ); +} + /// #68 review Major 4: the retry split is now a per-call statement. A 503 on /// a retrying READ is attempted three times; the same 503 on a WRITE path /// (`empty` — no marker, no retry machinery at all) is attempted once. The diff --git a/crates/tinymemory-remote/src/mem0.rs b/crates/tinymemory-remote/src/mem0.rs index bf3eff1..a53e180 100644 --- a/crates/tinymemory-remote/src/mem0.rs +++ b/crates/tinymemory-remote/src/mem0.rs @@ -605,6 +605,26 @@ impl Dialect for Mem0Dialect { entry.key ); } + // A NON-EMPTY answer in which nothing decodes is inconclusive, not + // absent, whenever the server admits more pages exist: a server that + // dropped the filter answers with the whole account, and the record + // asked for may sit past page 1. "More exists" shows up two ways — + // a full page, or a non-null `next` cursor on a short one (a server + // paginating below the requested page size). An EMPTY page is + // exhaustion regardless of the cursor, exactly as `cloud_walk` rules + // (an empty page cannot progress a walk): a filter-honoring server + // with the record would have put it on this first filtered page, so + // zero results IS the trustworthy absent — and the same verdict keeps + // `get`/`forget` agreeing with `list` about one response shape. + let page_exhausted = response.get("next").is_none_or(Value::is_null); + anyhow::ensure!( + results.is_empty() || (results.len() < CLOUD_PAGE_SIZE as usize && page_exhausted), + "mem0's filtered lookup answered {} records none of which are TinyMemory's, \ + with more pages remaining — the server did not honor the metadata filter, \ + and the record asked for ({namespace}/{key}) may sit beyond this page; \ + refusing to answer an untrustworthy `absent`", + results.len() + ); Ok(None) } @@ -675,13 +695,13 @@ impl Dialect for Mem0Dialect { } /// Finds and deletes an exact TinyMemory logical record. + /// + /// Through the keyed seam (issue #75): one filtered request on cloud, one + /// namespace-scoped listing self-hosted — the whole-account walk this + /// used to run died at the OSS ≥1000-record ceiling and paged the entire + /// hosted account to delete one record. async fn delete(&self, namespace: &str, key: &str) -> anyhow::Result { - let Some(entry) = self - .entries() - .await? - .into_iter() - .find(|item| item.namespace == namespace && item.key == key) - else { + let Some(entry) = self.entry(namespace, key).await? else { return Ok(false); }; self.client diff --git a/crates/tinymemory-remote/src/mem0_test.rs b/crates/tinymemory-remote/src/mem0_test.rs index 4be548f..29f4604 100644 --- a/crates/tinymemory-remote/src/mem0_test.rs +++ b/crates/tinymemory-remote/src/mem0_test.rs @@ -219,19 +219,37 @@ 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 next_cursor: Arc> = Arc::new(Mutex::new(Value::Null)); 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 cursor = next_cursor.clone(); + let deletes: Arc>> = Arc::default(); + let removed = deletes.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": cursor.lock().expect("cursor").clone(), + })) + } + }), + ) + .route( + "/v1/memories/{id}/", + axum::routing::delete(move |Path(id): Path| { + let removed = removed.clone(); + async move { + removed.lock().expect("deletes").push(id); + StatusCode::NO_CONTENT + } + }), + ); let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await .expect("bind"); @@ -283,4 +301,85 @@ async fn cloud_keyed_lookup_filters_by_metadata_and_verifies() { err.is_err(), "a mismatched filtered answer must refuse, not serve another record" ); + + // Issue #75 (re-landing the #71 review's M2, dropped by the crates-layout + // merge): a FULL page of records none of which even decode is + // inconclusive, not absent — the record may sit past page 1 of a + // filter-dropping server's account. + let junk: Vec = (0..200) + .map(|n| json!({"id": format!("foreign-{n}"), "memory": "not ours", "metadata": {}})) + .collect(); + *answer.lock().expect("answer") = json!(junk); + let err = driver.get("project", "decision").await; + assert!( + err.is_err(), + "a full undecodable page must refuse, not report absent" + ); + + // A SHORT page of undecodable records IS a trustworthy absent — but only + // under a terminal cursor: the server returned everything it had and + // ours was not among it. + *answer.lock().expect("answer") = + json!([{"id": "foreign-1", "memory": "not ours", "metadata": {}}]); + let got = driver + .get("project", "decision") + .await + .expect("short undecodable page"); + assert!(got.is_none(), "a short terminal page proves absence"); + + // The same short undecodable page with a NON-NULL `next` admits more + // pages exist (a server paginating below the requested size): absence is + // not proven, refuse. + *next_cursor.lock().expect("cursor") = json!("https://api.mem0.ai/v3/memories/?page=2"); + let err = driver.get("project", "decision").await; + assert!( + err.is_err(), + "a short undecodable page with a live cursor must refuse, not report absent" + ); + *next_cursor.lock().expect("cursor") = Value::Null; + + // An EMPTY page is exhaustion regardless of the cursor — `cloud_walk`'s + // own rule, mirrored so get/forget agree with list about one response + // shape: a filter-honoring server with the record would have put it on + // this first filtered page, so zero results is the trustworthy absent. + *answer.lock().expect("answer") = json!([]); + *next_cursor.lock().expect("cursor") = json!("https://api.mem0.ai/v3/memories/?page=2"); + let got = driver + .get("project", "decision") + .await + .expect("empty page with a live cursor is still absence, not an error"); + assert!(got.is_none()); + assert!( + !driver + .forget("project", "decision") + .await + .expect("forget of an absent key answers false, not an error"), + "forget must report false for an absent key" + ); + *next_cursor.lock().expect("cursor") = Value::Null; + + // Issue #75: delete rides the SAME keyed seam — one filtered resolve + // carrying the metadata key, then one DELETE by id. No account walk. + *answer.lock().expect("answer") = json!([record("project", "decision")]); + let before = bodies.lock().expect("bodies").len(); + let removed_ok = driver.forget("project", "decision").await.expect("forget"); + assert!(removed_ok); + let sent = bodies.lock().expect("bodies").clone(); + assert_eq!( + sent.len(), + before + 1, + "delete resolves with exactly one filtered request" + ); + let clauses = sent[before]["filters"]["AND"].as_array().expect("AND"); + assert!( + clauses + .iter() + .any(|c| c["metadata"]["tinymemory_key"] == json!("decision")), + "the delete's resolve carries the key filter: {sent:?}" + ); + assert_eq!( + deletes.lock().expect("deletes").as_slice(), + ["mem-1".to_owned()], + "one DELETE by resolved id" + ); } diff --git a/crates/tinymemory-remote/src/supermemory.rs b/crates/tinymemory-remote/src/supermemory.rs index 65aee58..c439ae7 100644 --- a/crates/tinymemory-remote/src/supermemory.rs +++ b/crates/tinymemory-remote/src/supermemory.rs @@ -278,54 +278,62 @@ impl SupermemoryDialect { /// over HTTP. async fn memories_in_tag(&self, container_tag: &str) -> anyhow::Result> { let mut entries = Vec::new(); - { - let mut page = 1_u64; - loop { - let response: Value = self - .client - .json( - Method::POST, - "v4/memories/list", - Some(&json!({ - "limit": 200, - "page": page, - "sort": "createdAt", - "order": "desc", - "containerTags": [container_tag] - })), - Attempts::RetryTransient, - ) - .await?; - let memories = response - .get("memoryEntries") - .and_then(Value::as_array) - .cloned() - .unwrap_or_default(); - for memory in memories { - let is_latest = memory - .get("isLatest") - .and_then(Value::as_bool) - .unwrap_or(true); - let is_forgotten = memory - .get("isForgotten") - .and_then(Value::as_bool) - .unwrap_or(false); - if is_latest && !is_forgotten { - let Some(entry) = Self::decode(&memory) else { - continue; - }; - entries.push(entry); - } - } - let total_pages = response - .pointer("/pagination/totalPages") - .and_then(Value::as_u64) - .unwrap_or(1); - if page >= total_pages { - break; + // Bounded like mem0's CLOUD_MAX_PAGES (issue #75): totalPages is + // server-supplied, and a lying or looping value must fail loudly + // instead of spinning the walk and growing the buffer forever. + const MAX_PAGES: u64 = 500; + let mut page = 1_u64; + loop { + let response: Value = self + .client + .json( + Method::POST, + "v4/memories/list", + Some(&json!({ + "limit": 200, + "page": page, + "sort": "createdAt", + "order": "desc", + "containerTags": [container_tag] + })), + Attempts::RetryTransient, + ) + .await?; + let memories = response + .get("memoryEntries") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + for memory in memories { + let is_latest = memory + .get("isLatest") + .and_then(Value::as_bool) + .unwrap_or(true); + let is_forgotten = memory + .get("isForgotten") + .and_then(Value::as_bool) + .unwrap_or(false); + if is_latest && !is_forgotten { + let Some(entry) = Self::decode(&memory) else { + continue; + }; + entries.push(entry); } - page += 1; } + let total_pages = response + .pointer("/pagination/totalPages") + .and_then(Value::as_u64) + .unwrap_or(1); + if page >= total_pages { + break; + } + anyhow::ensure!( + page < MAX_PAGES, + "supermemory kept answering more pages after {MAX_PAGES} — a \ + totalPages that never lets the walk finish is a server fault; \ + refusing rather than walking forever" + ); + page += 1; } Ok(entries) } @@ -360,7 +368,13 @@ impl Dialect for SupermemoryDialect { Some(&json!({ "id": existing.remote_id, "newContent": entry.content, - "metadata": metadata + "metadata": metadata, + // Required by PATCH /v4/memories ("Required to scope + // the operation"; 400 without it) — the POST and + // DELETE bodies always carried it, PATCH alone + // didn't, so the FIRST store of a key worked and + // every re-store failed (issue #75). + "containerTag": Self::container_tag(&entry.namespace), })), ) .await?; diff --git a/crates/tinymemory-remote/src/supermemory_test.rs b/crates/tinymemory-remote/src/supermemory_test.rs index 3204b0b..284b765 100644 --- a/crates/tinymemory-remote/src/supermemory_test.rs +++ b/crates/tinymemory-remote/src/supermemory_test.rs @@ -62,6 +62,7 @@ async fn add(State(state): State, Json(body): Json) -> Json, Json(body): Json) -> Json, Json(body): Json) -> StatusCode { + // The real PATCH /v4/memories requires `containerTag` ("Required to scope + // the operation") and 400s without it — a double that accepted a tagless + // PATCH hid exactly the adapter regression issue #75 found. And the VALUE + // matters as much as the presence: the tag scopes the operation, so a + // PATCH carrying another container's tag is a lost update or a + // cross-container write — refuse a mismatch instead of filing it. + let Some(sent_tag) = body["containerTag"].as_str().filter(|tag| !tag.is_empty()) else { + return StatusCode::BAD_REQUEST; + }; let id = body["id"].as_str().unwrap_or_default(); if let Some(record) = state .0 @@ -78,6 +88,9 @@ async fn update(State(state): State, Json(body): Json) -> Statu .iter_mut() .find(|r| r["id"] == id) { + if record["containerTag"] != sent_tag { + return StatusCode::BAD_REQUEST; + } record["memory"] = body["newContent"].clone(); record["metadata"] = body["metadata"].clone(); }