Keyed CRUD: reads stop enumerating the account (#69)

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<Vec<(String, String)>> {
+        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<Option<String>> {
+        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<Option<StoredEntry>> {
+        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<Option<Dataset>> {
         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,25 +376,49 @@ 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<Vec<StoredEntry>> {
+        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<Option<StoredEntry>> {
+        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",
             multipart::Part::bytes(body)
                 .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<bool> {
+        // 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)
     }
 

adapters/remote/src/cognee_test.rs

@@ -20,9 +20,25 @@ use tinymemory_api::{
 };
 
 #[derive(Clone, Default)]
-struct AppState(Arc<Mutex<Option<Vec<u8>>>>);
+struct AppState(
+    Arc<Mutex<Option<Vec<u8>>>>,
+    Arc<Mutex<CallCounts>>,
+    /// 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<Mutex<Option<String>>>,
+);
+
+/// 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<AppState>) -> Json<Value> {
+    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<AppState>) -> Json<Value> {
     Json(Value::Array(values))
 }
 async fn data(State(state): State<AppState>) -> Json<Value> {
+    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<AppState>) -> 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<AppState>) -> impl IntoResponse {
 async fn remember(State(state): State<AppState>, 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");
+}