Merge pull request #65 from YellowSnnowmann/feat/final-review-fixes

Close five defects the final production review found

Author: YellowSnnowmann

Cargo.lock

@@ -67,21 +67,41 @@ let provider = Arc::new(tinymemory::remote::supermemory_provider(backend));
 The remote adapter reaches only crates.io dependencies, so cargo resolves it
 without any `[patch]` entries.
 
-**The embedded engine (TinyCortex) — three patch entries:**
+**The embedded engine (TinyCortex) — vendor this repository as a submodule.**
+
+The remote recipe above works by git because the remote adapter reaches only
+published crates. The embedded engine does not: it pulls `tinycortex`,
+`tinycortex-api` and `tinyagents`, none of which are published, and
+`tinycortex-api` takes `tinymemory-api` *by git*, which cargo will resolve as a
+second copy of a crate this workspace also provides by path. Patching that away
+needs the crates on disk, so the embedded path is a submodule dependency until
+these crates are published:
+
+```sh
+git submodule add https://github.com/tinyhumansai/tinymemory vendor/tinymemory
+git -C vendor/tinymemory submodule update --init --recursive
+```
 
 ```toml
 [dependencies]
-tinymemory = { git = "https://github.com/tinyhumansai/tinymemory", features = ["tinycortex"] }
+tinymemory = { path = "vendor/tinymemory", features = ["tinycortex"] }
 
-# The engine and its api are unpublished; without these, cargo resolves a
-# second copy of each from the network and type identities split at the seam.
+# All four are required. The first three are unpublished crates the engine
+# needs; the fourth collapses `tinycortex-api`'s git dependency on
+# `tinymemory-api` onto the copy in this tree — without it two distinct
+# `tinymemory_api::MemoryEntry` types exist and the seam stops type-checking.
 [patch.crates-io]
-tinycortex = { git = "https://github.com/tinyhumansai/tinycortex" }
-tinycortex-api = { git = "https://github.com/tinyhumansai/tinycortex" }
+tinycortex = { path = "vendor/tinymemory/vendor/tinycortex" }
+tinycortex-api = { path = "vendor/tinymemory/vendor/tinycortex/api" }
+tinyagents = { path = "vendor/tinymemory/vendor/tinyagents" }
 [patch."https://github.com/tinyhumansai/tinymemory"]
-tinymemory-api = { git = "https://github.com/tinyhumansai/tinymemory" }
+tinymemory-api = { path = "vendor/tinymemory/api" }
 ```
 
+This exact patch set is what the reference consumer in `examples/` and the
+repository's own root manifest use; a build missing any of the four fails at
+resolution, before compiling a line.
+
 ```rust,ignore
 use std::sync::Arc;
 use tinymemory::tinycortex::{provider, InMemoryMemoryStore};

README.md

@@ -16,9 +16,14 @@ async-trait = "0.1"
 # The storage trait deliberately uses opaque backend errors.
 anyhow = "1"
 # Native self-hosted APIs are HTTP/JSON; multipart is required by Cognee.
-reqwest = { version = "0.12", default-features = false, features = ["json", "multipart", "rustls-tls"] }
+# `stream` is for `bytes_stream()`: response bodies are read against a byte
+# cap rather than buffered whole, because the endpoint is operator-supplied
+# and a broken or hostile one must not be able to OOM the host.
+reqwest = { version = "0.12", default-features = false, features = ["json", "multipart", "rustls-tls", "stream"] }
 # Remote records are translated through a private, lossless envelope.
 serde = { version = "1", features = ["derive"] }
+# Streaming a capped body needs a Stream combinator.
+futures = "0.3"
 serde_json = "1"
 # Supermemory custom ids are bounded, so namespace/key identities use SHA-256.
 sha2 = "0.10"

adapters/remote/Cargo.toml

@@ -40,6 +40,56 @@ impl std::fmt::Debug for HttpClient {
     }
 }
 
+/// Largest response body any hosted engine may return.
+///
+/// The endpoint is operator-supplied (`SupermemoryMemory::api`,
+/// `Mem0Memory::new`, `CogneeMemory::self_hosted` all take an arbitrary URL),
+/// so a broken or hostile server must not be able to exhaust the host's
+/// memory. 64 MiB is far above any real memory payload -- the largest thing
+/// these APIs return is a page of records -- and far below a size that
+/// threatens a process.
+const MAX_RESPONSE_BYTES: u64 = 64 * 1024 * 1024;
+
+/// Read a response body, failing once it exceeds [`MAX_RESPONSE_BYTES`].
+///
+/// `Response::json()`/`text()` buffer the whole body before any size check, so
+/// a server that omits or understates `Content-Length` (a chunked response,
+/// say) could OOM the process despite a declared limit. Reading incrementally
+/// 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.
+async fn read_capped(response: reqwest::Response, path: &str) -> anyhow::Result<Vec<u8>> {
+    use futures::StreamExt;
+    if let Some(len) = response.content_length() {
+        if len > MAX_RESPONSE_BYTES {
+            anyhow::bail!(
+                "memory API {path} response exceeds {MAX_RESPONSE_BYTES}-byte limit \
+                 (Content-Length={len})"
+            );
+        }
+    }
+    let mut body = Vec::new();
+    let mut stream = response.bytes_stream();
+    while let Some(chunk) = stream.next().await {
+        let chunk = chunk.with_context(|| format!("memory API {path} body read failed"))?;
+        // Check BEFORE appending: one oversized chunk would otherwise be
+        // allocated in full before the limit is noticed, which is the
+        // allocation this cap exists to prevent.
+        let next_len = body
+            .len()
+            .checked_add(chunk.len())
+            .context("memory API response length overflowed")?;
+        if next_len as u64 > MAX_RESPONSE_BYTES {
+            anyhow::bail!(
+                "memory API {path} response exceeds {MAX_RESPONSE_BYTES}-byte limit \
+                 (would reach {next_len} bytes)"
+            );
+        }
+        body.extend_from_slice(&chunk);
+    }
+    Ok(body)
+}
+
 impl HttpClient {
     /// Builds a client that optionally authenticates with a bearer token.
     pub(crate) fn bearer(endpoint: &str, credential: Option<&str>) -> anyhow::Result<Self> {
@@ -168,9 +218,8 @@ impl HttpClient {
         if !status.is_success() {
             return Err(self.status_error(path, status));
         }
-        response
-            .json()
-            .await
+        let body = read_capped(response, path).await?;
+        serde_json::from_slice(&body)
             .with_context(|| format!("memory API {path} returned invalid JSON"))
     }
 
@@ -185,10 +234,8 @@ impl HttpClient {
         if !status.is_success() {
             return Err(self.status_error(path, status));
         }
-        response
-            .text()
-            .await
-            .context("memory API response was unreadable")
+        let body = read_capped(response, path).await?;
+        String::from_utf8(body).context("memory API response was not valid UTF-8")
     }
 
     /// Sends a request whose successful response body is not needed.

adapters/remote/src/common.rs

@@ -119,17 +119,52 @@ struct Mem0Dialect {
 }
 
 impl Mem0Dialect {
+    /// Largest listing this adapter will request in one call.
+    ///
+    /// Every exact-CRUD path here enumerates through [`Self::values`], so this
+    /// is the ceiling on the whole store, not on one page.
+    const LISTING_TOP_K: usize = 1000;
+
     /// Fetches Mem0's administrative memory listing.
+    ///
+    /// # A hard ceiling, deliberately loud
+    ///
+    /// This is a single unpaginated request, and it is the ONLY enumeration
+    /// path in this adapter -- `get`, `list`, `count` and `export_page` all
+    /// route through it. Past the ceiling the results are not merely
+    /// incomplete, they are silently WRONG: `get(ns, key)` for a record beyond
+    /// the cut-off returns `Ok(None)`, which the contract defines as "no such
+    /// entry", so a caller reads "deleted" where the truth is "present but
+    /// past the window".
+    ///
+    /// Returning an error instead is the honest failure. A full response is
+    /// indistinguishable from a truncated one -- both are exactly `top_k`
+    /// items -- so this cannot detect truncation, only its own boundary, and
+    /// it refuses at that boundary rather than answering wrongly. Paginating
+    /// properly needs Mem0's paging parameters verified against a live
+    /// service; guessing them here would trade a loud failure for a quiet one.
     async fn values(&self) -> anyhow::Result<Vec<Value>> {
+        let top_k = Self::LISTING_TOP_K;
         let response: Value = self
             .client
-            .json(Method::GET, "memories?top_k=1000", None)
+            .json(Method::GET, &format!("memories?top_k={top_k}"), None)
             .await?;
-        Ok(response
+        let results = response
             .get("results")
             .and_then(Value::as_array)
             .cloned()
-            .unwrap_or_default())
+            .unwrap_or_default();
+        if results.len() >= top_k {
+            anyhow::bail!(
+                "mem0 returned {} memories, this adapter's unpaginated listing ceiling. \
+                 Exact reads (get/list/count/export) cannot be answered correctly beyond \
+                 it -- a record past the window would read as absent -- so the adapter \
+                 refuses rather than answering wrongly. Recall is unaffected (it queries \
+                 mem0's search API directly).",
+                results.len()
+            );
+        }
+        Ok(results)
     }
 
     /// Decodes a Mem0 result containing TinyMemory-owned metadata.

adapters/remote/src/mem0.rs

@@ -1024,6 +1024,7 @@ impl MemorySourceSink for TinycortexProvider {
         items: Vec<SourceItem>,
         taint: MemoryTaint,
     ) -> Result<IngestOutcome, MemoryError> {
+        let items_len = items.len();
         let namespace = format!("source:{source_id}");
         let mut outcome = IngestOutcome::default();
         for item in items {
@@ -1063,8 +1064,20 @@ impl MemorySourceSink for TinycortexProvider {
                     outcome.written = outcome.written.saturating_add(1);
                     outcome.ids.push(id);
                 }
-                Err(_) => {
-                    outcome.skipped = outcome.skipped.saturating_add(1);
+                // A write failure is NOT `skipped`. The contract defines that
+                // field as "units the driver recognised as already present"
+                // (`IngestOutcome::skipped`), so counting a failed write there
+                // reports a locked database, a full disk or a dead embedder as
+                // a successful no-op: the sync caller marks the items done and
+                // they are never written. Propagate instead — a partial batch
+                // has no truthful representation in `IngestOutcome`, and a
+                // caller that wants best-effort ingestion can catch this.
+                Err(error) => {
+                    return Err(MemoryError::Other(anyhow::anyhow!(
+                        "source ingest failed after {} of {} item(s) were written: {error}",
+                        outcome.written,
+                        items_len
+                    )));
                 }
             }
         }