diff --git a/crates/tinymemory-remote/src/conformance_test.rs b/crates/tinymemory-remote/src/conformance_test.rs index c1e7d2c9..ebedace0 100644 --- a/crates/tinymemory-remote/src/conformance_test.rs +++ b/crates/tinymemory-remote/src/conformance_test.rs @@ -31,7 +31,10 @@ use serde_json::{json, Value}; use tinymemory_api::capabilities::Capability; use tinymemory_api::provider::{MemoryCore, MemoryProvider}; -use crate::{mem0_provider, Mem0Memory}; +use tinymemory_api::traits::Memory; +use tinymemory_api::types::MemoryCategory; + +use crate::{cortex_provider, mem0_provider, CortexMemory, Mem0Memory}; /// A record as one of the vendor doubles holds it. #[derive(Clone, Debug)] @@ -573,3 +576,634 @@ async fn the_cognee_double_actually_retains() { "the Cognee double must retain writes, or the suite passes vacuously" ); } + +// ── CortexDB's native shapes ──────────────────────────────────────────────── +// +// This double is deliberately the least accommodating of the three. The others +// model keyed stores, so an adapter bug around replacement would still look +// like success. CortexDB is an append-only event log, and the whole reason its +// adapter exists in its current shape is that a key cannot be rewritten — so +// the double reproduces that constraint exactly, refusing a reused idempotency +// key carrying a different body with the same `409 IDEMPOTENCY_CONFLICT` the +// real engine returns. +// +// A permissive double here would prove nothing: the suite's upsert assertion +// would pass because the backend allowed an overwrite, not because the adapter +// folded the log correctly. + +#[derive(Default)] +struct CortexLog { + /// Every event ever appended, in order. Never mutated — that is the point. + events: Vec, + /// `idempotency_key` -> the body it was first seen with, and the id of the + /// event that body produced. The id is stored rather than looked up by + /// content, because two keys may legitimately carry identical text and a + /// replay must answer with its *own* event. + idempotency: BTreeMap, + next_offset: u64, + next_id: u64, +} + +type CortexStore = Arc>; + +async fn cortex_experience( + State(store): State, + Json(body): Json, +) -> (axum::http::StatusCode, Json) { + let mut log = store.lock().expect("cortex log"); + let key = body + .get("idempotency_key") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + let payload = body + .pointer("/content/text") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + + if let Some((seen, event_id)) = log.idempotency.get(&key) { + if seen != &payload { + // The refusal the whole adapter is designed around. + return ( + axum::http::StatusCode::CONFLICT, + Json(json!({ "error_code": "IDEMPOTENCY_CONFLICT" })), + ); + } + return ( + axum::http::StatusCode::ACCEPTED, + Json(json!({ "event_id": event_id, "replayed_from_idempotency": true })), + ); + } + + log.next_offset += 2; + log.next_id += 1; + let offset = log.next_offset; + let id = format!("evt_{}", log.next_id); + log.idempotency.insert(key, (payload.clone(), id.clone())); + let scope = body + .get("scope") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + log.events.push(json!({ + "id": id, + "scope": scope, + "wal_offset": offset, + "content": { "kind": "message", "role": "user", "text": payload }, + "context": { "recorded_at": "2026-09-02T00:00:00Z" }, + })); + // The real id, not a placeholder: `/v1/experience` answers with the id the + // event was actually stored under, and the adapter waits on that id + // becoming readable before it reports the write as done. + ( + axum::http::StatusCode::ACCEPTED, + Json(json!({ "event_id": id, "status": "captured" })), + ) +} + +async fn cortex_events( + State(store): State, + Query(params): Query>, +) -> Json { + let log = store.lock().expect("cortex log"); + let scope = params.get("scope").cloned().unwrap_or_default(); + let cursor: usize = params + .get("cursor") + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + let limit: usize = params + .get("limit") + .and_then(|v| v.parse().ok()) + .unwrap_or(50); + // Newest first, and every record emitted twice, because that is what the + // engine does. Both details are load-bearing: an adapter that trusted array + // order instead of `wal_offset`, or that assumed `items` held distinct + // events, would pass against a tidier double and fail in production. Note + // `limit` counts the duplicates, so a page holds half as many records as + // its size suggests. + let mut stream: Vec = Vec::new(); + for event in log + .events + .iter() + .rev() + .filter(|e| e.get("scope").and_then(Value::as_str) == Some(scope.as_str())) + { + stream.push(event.clone()); + stream.push(event.clone()); + } + let page: Vec = stream.iter().skip(cursor).take(limit).cloned().collect(); + let next = cursor + page.len(); + let has_more = next < stream.len(); + Json(json!({ + "items": page, + "has_more": has_more, + "next_cursor": next.to_string(), + })) +} + +/// The destructive endpoint, with the interlocks the real one has. +/// +/// Three behaviours here are not decoration; each one has caught something: +/// +/// - the selector's id field is `memory_ids`. An unrecognised field is **not** +/// rejected — it deserialises to an empty selector, which means "the whole +/// scope"; +/// - an empty selector without `confirm_all` is refused, which is what keeps +/// that mistake from being destructive on its own; +/// - a non-empty selector *with* `confirm_all` is refused as ambiguous, rather +/// than silently widened to the scope. +async fn cortex_forget( + State(store): State, + Json(body): Json, +) -> (axum::http::StatusCode, Json) { + let mut log = store.lock().expect("cortex log"); + let scope = body + .get("scope") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + let confirm_all = body + .get("confirm_all") + .and_then(Value::as_bool) + .unwrap_or(false); + let ids: Vec = body + .pointer("/selector/memory_ids") + .and_then(Value::as_array) + .map(|a| { + a.iter() + .filter_map(Value::as_str) + .map(str::to_string) + .collect() + }) + .unwrap_or_default(); + let narrowed = ["about_subject", "about_entity", "predicate"] + .iter() + .any(|f| body.pointer(&format!("/selector/{f}")).is_some()); + let selective = !ids.is_empty() || narrowed; + + if selective && confirm_all { + return ( + axum::http::StatusCode::BAD_REQUEST, + Json(json!({ "error_code": "AMBIGUOUS_SELECTOR_CONFIRM_ALL" })), + ); + } + if !selective && !confirm_all { + return ( + axum::http::StatusCode::UNPROCESSABLE_ENTITY, + Json(json!({ "error_code": "EMPTY_SELECTOR_WITHOUT_CONFIRMATION" })), + ); + } + + let before = log.events.len(); + if selective { + log.events.retain(|e| { + !ids.contains( + &e.get("id") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + ) + }); + } else { + log.events + .retain(|e| e.get("scope").and_then(Value::as_str) != Some(scope.as_str())); + } + // Faithful to the engine: forgetting an event does NOT release its + // idempotency key. An adapter that tried delete-then-rewrite would be + // refused here, exactly as it is in production. + let deleted = before - log.events.len(); + ( + axum::http::StatusCode::OK, + Json(json!({ + "deleted": { "events": deleted }, + "requested": ids.len(), + "matched": deleted, + })), + ) +} + +async fn cortex_recall(State(store): State, Json(body): Json) -> Json { + let log = store.lock().expect("cortex log"); + let scope = body + .get("scope") + .and_then(Value::as_str) + .unwrap_or_default(); + let query = body + .get("query") + .and_then(Value::as_str) + .unwrap_or_default(); + let hits: Vec = log + .events + .iter() + .filter(|e| e.get("scope").and_then(Value::as_str) == Some(scope)) + .filter(|e| { + query.is_empty() + || e.pointer("/content/text") + .and_then(Value::as_str) + .is_some_and(|t| t.to_lowercase().contains(&query.to_lowercase())) + }) + .map(|e| { + // Recall renders content for a reader rather than returning it as + // stored: the speaker is prefixed. The listing does not do this, + // so the two read paths hand back different bytes for the same + // event — which is why the adapter parses both forms. + let mut hit = e.clone(); + if let Some(text) = e.pointer("/content/text").and_then(Value::as_str) { + hit["content"]["text"] = json!(format!("[user] {text}")); + } + hit + }) + .collect(); + Json(json!({ "layers": { "events": hits } })) +} + +async fn cortex_scopes(State(store): State) -> Json { + let log = store.lock().expect("cortex log"); + let mut paths: Vec = log + .events + .iter() + .filter_map(|e| e.get("scope").and_then(Value::as_str)) + .map(str::to_string) + .collect(); + paths.sort(); + paths.dedup(); + Json(json!({ + "items": paths.into_iter().map(|p| json!({ "path": p })).collect::>() + })) +} + +async fn cortex_backend() -> String { + let store: CortexStore = Arc::new(Mutex::new(CortexLog::default())); + let app = Router::new() + .route("/v1/experience", post(cortex_experience)) + .route("/v1/events", get(cortex_events)) + .route("/v1/forget", post(cortex_forget)) + .route("/v1/recall", post(cortex_recall)) + .route("/v1/scopes/list", get(cortex_scopes)) + .route( + "/v1/admin/health", + get(|| async { Json(json!({ "status": "healthy" })) }), + ) + .with_state(store); + serve(app).await +} + +/// The same backend, but with ranked recall broken. +/// +/// Used to prove that a store still succeeds when the search index cannot be +/// reached — the write is durable and readable by key, and the settle probe is +/// explicitly best-effort. +async fn cortex_backend_with_recall_down() -> String { + let store: CortexStore = Arc::new(Mutex::new(CortexLog::default())); + let app = Router::new() + .route("/v1/experience", post(cortex_experience)) + .route("/v1/events", get(cortex_events)) + .route("/v1/forget", post(cortex_forget)) + .route( + "/v1/recall", + post(|| async { + ( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({ "error_code": "INTERNAL" })), + ) + }), + ) + .route("/v1/scopes/list", get(cortex_scopes)) + .route( + "/v1/admin/health", + get(|| async { Json(json!({ "status": "healthy" })) }), + ) + .with_state(store); + serve(app).await +} + +/// A durable, readable write must not be reported as a failure because the +/// search index is down. +/// +/// The write path waits twice: once for the keyed read path, which is required, +/// and once for ranked recall, which is not. The second wait exists to make +/// read-after-write hold for `search` in the common case, and it must degrade +/// to "the index will catch up" rather than turning a successful store into an +/// error. +#[tokio::test] +async fn a_store_succeeds_when_ranked_recall_is_unreachable() { + let endpoint = cortex_backend_with_recall_down().await; + let memory = CortexMemory::api(&endpoint, "test-key").expect("client"); + + memory + .store("tenant", "k", "content", MemoryCategory::Core, None) + .await + .expect("a store must not fail because the settle probe could not be answered"); + + assert_eq!( + memory + .get("tenant", "k") + .await + .expect("get") + .map(|e| e.content) + .as_deref(), + Some("content"), + "the record the store reported as written must be readable by key" + ); +} + +#[tokio::test] +async fn cortex_upholds_the_contract() { + let endpoint = cortex_backend().await; + let provider = cortex_provider(CortexMemory::api(&endpoint, "test-key").expect("client")); + tinymemory_conformance::assert_provider(Arc::new(provider)).await; +} + +/// The suite's write-path assertions only run when the driver retains. +#[tokio::test] +async fn the_cortex_double_actually_retains() { + let endpoint = cortex_backend().await; + let provider = cortex_provider(CortexMemory::api(&endpoint, "test-key").expect("client")); + assert!( + tinymemory_conformance::retains_writes(&provider).await, + "the CortexDB double must retain writes, or `assert_provider` skips every \ + assertion that matters and still reports success" + ); +} + +/// The double must refuse a reused key with a changed body, or the suite's +/// upsert assertion passes for the wrong reason. +/// +/// This is the constraint the adapter is built around. If the double ever +/// becomes permissive, `cortex_upholds_the_contract` would prove that a keyed +/// backend upholds the contract — which is true and irrelevant. +#[tokio::test] +async fn the_cortex_double_refuses_a_reused_key_with_a_changed_body() { + let endpoint = cortex_backend().await; + let client = reqwest::Client::new(); + let send = |text: &str| { + let body = json!({ + "scope": "tm:probe", + "idempotency_key": "fixed", + "content": { "kind": "message", "role": "user", "text": text }, + "context": {}, + }); + client + .post(format!("{endpoint}/v1/experience")) + .json(&body) + .send() + }; + assert_eq!(send("first").await.expect("send").status(), 202); + assert_eq!( + send("second").await.expect("send").status(), + 409, + "a permissive double would make the adapter's whole reason for existing untested" + ); +} + +/// A scope larger than one page must fold completely. +/// +/// The listing pages with `cursor`/`next_cursor`, and the engine ignores query +/// parameters it does not recognise rather than refusing them — so a wrong +/// parameter name does not surface as an error, it silently re-serves page one +/// until the adapter's page ceiling trips. Because the engine also emits every +/// record twice and counts the duplicates against `limit`, the boundary arrives +/// at roughly half the page size. This writes past it. +#[tokio::test] +async fn a_scope_past_one_page_folds_completely() { + let endpoint = cortex_backend().await; + let memory = CortexMemory::api(&endpoint, "test-key").expect("client"); + for i in 0..140 { + memory + .store( + "paged", + &format!("key-{i:03}"), + &format!("value {i}"), + MemoryCategory::Core, + None, + ) + .await + .expect("store"); + } + let entries = memory.list(Some("paged"), None, None).await.expect("list"); + assert_eq!( + entries.len(), + 140, + "the fold walked a truncated listing; every distinct record must survive paging" + ); + let found = memory.get("paged", "key-139").await.expect("get"); + assert_eq!(found.map(|e| e.content).as_deref(), Some("value 139")); +} + +/// Deleting one key must not take the scope with it. +/// +/// The destructive endpoint has two failure shapes that both end in an empty +/// selector: an unrecognised selector field, and `confirm_all` sent alongside a +/// real one. The first is silent. This asserts the adapter lands in neither. +#[tokio::test] +async fn deleting_one_key_leaves_its_neighbours_alone() { + let endpoint = cortex_backend().await; + let memory = CortexMemory::api(&endpoint, "test-key").expect("client"); + for key in ["alpha", "beta", "gamma"] { + memory + .store( + "tenant", + key, + &format!("{key} value"), + MemoryCategory::Core, + None, + ) + .await + .expect("store"); + } + // A second version of the doomed key, so the delete has to reach both. + memory + .store( + "tenant", + "beta", + "beta rewritten", + MemoryCategory::Core, + None, + ) + .await + .expect("store"); + + assert!(memory.forget("tenant", "beta").await.expect("forget")); + + assert!(memory.get("tenant", "beta").await.expect("get").is_none()); + assert_eq!( + memory + .get("tenant", "alpha") + .await + .expect("get") + .map(|e| e.content) + .as_deref(), + Some("alpha value"), + "a neighbour disappeared: the delete widened to the whole scope" + ); + assert_eq!( + memory + .get("tenant", "gamma") + .await + .expect("get") + .map(|e| e.content) + .as_deref(), + Some("gamma value") + ); + let left = memory.list(Some("tenant"), None, None).await.expect("list"); + assert_eq!( + left.len(), + 2, + "expected alpha and gamma to remain, got {left:?}" + ); +} + +/// A deleted key stays deleted even if the removal half fails. +/// +/// The tombstone is what makes that true, and it is why `delete` writes one +/// before touching the destructive endpoint at all. +#[tokio::test] +async fn a_tombstone_alone_is_enough_to_hide_a_key() { + let endpoint = cortex_backend().await; + let memory = CortexMemory::api(&endpoint, "test-key").expect("client"); + memory + .store("tenant", "doomed", "still here", MemoryCategory::Core, None) + .await + .expect("store"); + + // Append the tombstone by hand and never call forget, standing in for a + // removal whose second half was lost. + let client = reqwest::Client::new(); + let tombstone = json!({ "k": "doomed", "c": "", "d": true }); + let sent = client + .post(format!("{endpoint}/v1/experience")) + .json(&json!({ + "scope": "tm:tenant", + "idempotency_key": "hand-written-tombstone", + "content": { "kind": "message", "role": "user", "text": tombstone.to_string() }, + "context": {}, + })) + .send() + .await + .expect("send"); + assert_eq!(sent.status(), 202); + + assert!( + memory.get("tenant", "doomed").await.expect("get").is_none(), + "the fold ignored a tombstone, so a delete that lost its second half \ + would resurrect the record" + ); + assert!(memory + .list(Some("tenant"), None, None) + .await + .expect("list") + .is_empty()); +} + +/// Recall must return what the engine ranked, not what sorts first. +/// +/// The fold orders by key, which is right for a listing and wrong for a ranked +/// answer: truncating an alphabetical order to `limit` discards the engine's +/// best hits and keeps whichever keys happen to sort early. The conformance +/// suite cannot catch this on its own — its recall fixture stores identical +/// content under `r1`/`r2`/`r3`, where ranked and alphabetical order coincide. +#[tokio::test] +async fn recall_keeps_the_engine_ranking_when_it_truncates() { + let endpoint = cortex_backend().await; + let memory = CortexMemory::api(&endpoint, "test-key").expect("client"); + // Stored — and so ranked by the double — in the opposite order to the one + // the keys sort in. + for key in ["zulu", "alpha"] { + memory + .store( + "ranked", + key, + "shared needle text", + MemoryCategory::Core, + None, + ) + .await + .expect("store"); + } + + let opts = tinymemory_api::recall::RecallOpts { + namespace: Some("ranked"), + ..Default::default() + }; + let hits = memory.recall("needle", 1, opts).await.expect("recall"); + + assert_eq!(hits.len(), 1, "the limit must still be honoured"); + assert_eq!( + hits[0].key, "zulu", + "recall returned the alphabetically first key, not the highest ranked \ + one — the engine's ordering was thrown away before the truncation" + ); +} + +/// A replay must answer with its own event, not one that happens to match. +/// +/// Two idempotency keys may legitimately carry identical text — the adapter +/// mints a fresh key per write, so a re-store of unchanged content is exactly +/// this shape. A double that resolved a replay by searching content would hand +/// back the first matching event for both, and the write path waits on the id +/// it is given: it would be waiting on the wrong record. +#[tokio::test] +async fn a_replay_returns_the_event_its_own_key_created() { + let endpoint = cortex_backend().await; + let client = reqwest::Client::new(); + let send = |key: &'static str| { + let body = json!({ + "scope": "tm:probe", + "idempotency_key": key, + "content": { "kind": "message", "role": "user", "text": "identical text" }, + "context": {}, + }); + client + .post(format!("{endpoint}/v1/experience")) + .json(&body) + .send() + }; + let id_of = |v: &Value| { + v.get("event_id") + .and_then(Value::as_str) + .map(str::to_string) + }; + + let first: Value = send("key-a") + .await + .expect("send") + .json() + .await + .expect("json"); + let second: Value = send("key-b") + .await + .expect("send") + .json() + .await + .expect("json"); + let (a, b) = (id_of(&first), id_of(&second)); + assert_ne!(a, b, "two keys with the same text must create two events"); + + let replay_a: Value = send("key-a") + .await + .expect("send") + .json() + .await + .expect("json"); + let replay_b: Value = send("key-b") + .await + .expect("send") + .json() + .await + .expect("json"); + assert_eq!( + replay_a.get("replayed_from_idempotency"), + Some(&json!(true)) + ); + assert_eq!( + id_of(&replay_a), + a, + "key-a replayed with another key\'s event" + ); + assert_eq!( + id_of(&replay_b), + b, + "key-b replayed with another key\'s event" + ); +} diff --git a/crates/tinymemory-remote/src/cortex.rs b/crates/tinymemory-remote/src/cortex.rs new file mode 100644 index 00000000..cd0218be --- /dev/null +++ b/crates/tinymemory-remote/src/cortex.rs @@ -0,0 +1,979 @@ +//! CortexDB adapter. +//! +//! # Why this one looks different from its neighbours +//! +//! Supermemory, Mem0 and Cognee are keyed stores: `upsert` overwrites the row +//! at `(namespace, key)` and the dialect is a thin translation. CortexDB is an +//! **append-only event log**. A key, once written, cannot be given a different +//! value: +//! +//! - the same `idempotency_key` with a different body is refused with +//! `409 IDEMPOTENCY_CONFLICT`; +//! - there is no update route — `/v1/events` is read-only and `/v1/experience` +//! only appends; +//! - `/v1/forget` removes the event but **not** its idempotency record, so +//! delete-then-rewrite loses the old value and still refuses the new one. +//! +//! So this dialect does not try to hold TinyMemory's key. It writes every store +//! as a fresh event with its own idempotency key — which the engine always +//! accepts — and carries the logical key inside the payload. Reads then fold +//! the log down to one record per key, newest wins. The contract's replace +//! semantics are reconstructed on the read side rather than performed on the +//! write side. +//! +//! ## What that costs, stated plainly +//! +//! **Reads are a scan.** `/v1/events` has no metadata filter, so there is no +//! server-side lookup by our key. Every read fetches the scope and folds it, +//! and that walk grows with everything the namespace has ever held. The keyed +//! `entry` seam other dialects override to a single round trip cannot be +//! overridden here. +//! +//! **Superseded versions stay in the engine's own recall corpus.** We fold them +//! out of `entries`, `namespace_entries` and `entry`, but the search seam +//! delegates to CortexDB's ranked recall, which searches every event including +//! the ones we consider replaced. A caller can therefore see a stale value +//! through `recall` that `get` would never return. That is not a bug in this +//! adapter; it is the cost of emulating replacement on an engine that does not +//! offer it, and it is the reason to prefer a native upsert if CortexDB ever +//! exposes one. +//! +//! **Writes block until the record can be read.** `/v1/experience` answers +//! `202 captured` and indexes afterwards, so an accepted write is not yet a +//! readable one. The contract requires read-after-write, so `upsert` waits — +//! see `CortexDialect::await_readable` for the two waits and why only one of +//! them is fatal. Measured against a running engine that is roughly one to +//! four seconds per write, and it dominates: the full conformance suite takes +//! about three minutes here against seconds on a keyed engine. It is the cost +//! of a durable-then-indexed pipeline, not of anything this adapter does. +//! +//! Every cost above disappears the day the engine grows `on_conflict: +//! "replace"`, releases an idempotency key on forget, and offers a readiness +//! signal a writer can wait on. +//! +//! ## Engine behaviours worth knowing before editing this +//! +//! Each of these was measured against a live CortexDB, and each one was wrong +//! in this adapter first — the offline suite in `conformance_test.rs` was green +//! throughout, because a double written from documentation agrees with an +//! adapter written from the same documentation. The double reproduces all of +//! them now. +//! +//! - **The two read paths return different bytes for the same event.** +//! `/v1/events` returns stored text; `/v1/recall` renders it for a reader and +//! prefixes the speaker (`[user] {...}`). Parsing only the stored form yields +//! a dialect that lists correctly and searches to nothing. See +//! `CortexDialect::envelope_of`. +//! - **The listing emits every record twice**, and `limit` counts the +//! duplicates. Paging with the cursor still enumerates everything; a reader +//! that assumes uniqueness does not. See `CortexDialect::events`. +//! - **Unknown query parameters are ignored, not refused.** A wrong paging +//! parameter re-serves page one indefinitely rather than erroring. +//! - **The destructive selector's id field is `memory_ids`.** An unrecognised +//! field is read as an *empty* selector, which means the whole scope. Two +//! interlocks stop that being destructive on its own, and +//! `CortexDialect::delete` explains why `confirm_all` appears nowhere +//! here. +//! - **`/v1/experience/status` and the advertised `lifecycle_stream` are not +//! readiness signals.** The first never advances past `captured`; the second +//! accepts a connection and emits nothing. + +use async_trait::async_trait; +use reqwest::Method; +use serde_json::{json, Value}; +use tinymemory_api::recall::RecallOpts; +use tinymemory_api::traits::Memory; +use tinymemory_api::types::{MemoryCategory, MemoryTaint}; + +use crate::common::{Attempts, Dialect, HttpClient, RemoteMemory, StoredEntry}; + +/// Stable driver id used by configuration and status output. +pub const CORTEX_DRIVER_ID: &str = "cortex"; + +/// Default base URL for CortexDB's managed API. +pub const CORTEX_API_ENDPOINT: &str = "https://api-v1.cortexdb.ai"; + +/// Scope-id prefix for a namespace segment CortexDB's grammar accepts as-is. +/// +/// The scope grammar is `type:id`, so every segment needs a type. Keeping the +/// common case literal means a scope stays readable in the engine's own tools. +const SEGMENT_PLAIN: &str = "tm"; + +/// Scope-id prefix for a segment that had to be hex-encoded to fit. +/// +/// The contract allows characters in a namespace that a Cortex scope id does +/// not — `:` above all, which the grammar uses to separate type from id, and +/// which the contract uses to address a namespace *section* +/// (`conversation:thread-8f21`). Collapsing or refusing it are both wrong: the +/// first silently re-addresses the namespace out of its section, and the second +/// makes whole sections unstorable. So such a segment is encoded, and +/// [`CortexDialect::namespace_of`] decodes it — which matters more than it +/// looks, because `namespaces()` must report the *logical* namespace back, and +/// `Bound::recall` re-checks every returned record against the namespace it +/// asked for and silently drops what does not match. +const SEGMENT_ENCODED: &str = "tmx"; + +/// How many `/`-separated segments a CortexDB scope path may hold. +/// +/// Measured, not read off the grammar: 32 are accepted and 33 are refused with +/// `422 INVALID_BODY`. +const MAX_SCOPE_SEGMENTS: usize = 32; + +/// How many events one listing page asks for. +/// +/// The fold needs every event in a scope, so this bounds one request rather +/// than the walk. Larger pages mean fewer round trips through the same +/// unavoidable scan. +const PAGE_SIZE: usize = 200; + +/// How long a write waits for its own event to become readable. +/// +/// Ingestion is asynchronous — see `CortexDialect::await_readable`. Measured +/// against the staging engine, a record reaches the listing in about 1–4s and +/// ranked recall about a second later; this is +/// a wide margin over that, because the failure it guards is a write that +/// reports success and is then invisible to the next read. +const VISIBILITY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + +/// Gap between visibility polls. Short enough not to dominate the wait. +const VISIBILITY_POLL: std::time::Duration = std::time::Duration::from_millis(250); + +/// How long a write lets the *search* index catch up before giving up on it. +/// +/// Shorter than [`VISIBILITY_TIMEOUT`] and, unlike it, not fatal — see phase +/// two of `CortexDialect::await_readable`. +const RECALL_SETTLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +/// Longest query the settle probe will send. +/// +/// The probe queries with the text just written, and a record may be large — +/// the conformance suite stores 64 KiB. Sending all of it as a query is both +/// wasteful and worse at matching than a distinctive prefix. +const RECALL_QUERY_CAP: usize = 256; + +/// Ceiling on the pages one fold will walk. +/// +/// A scan with no ceiling is an outage waiting for a large enough namespace. +/// Hitting it is an error rather than a truncated answer: a silently short +/// listing would present a superseded value as current, which is the one +/// failure this whole adapter exists to avoid. +const MAX_PAGES: usize = 500; + +/// CortexDB, adapted to TinyMemory's keyed contract. +#[derive(Debug)] +pub struct CortexMemory { + inner: RemoteMemory, +} + +impl CortexMemory { + /// Rebuilds the HTTP transport with a different per-request deadline. + /// + /// # Errors + /// + /// Fails only if the underlying HTTP client cannot be rebuilt. + pub fn with_request_timeout(mut self, timeout: std::time::Duration) -> anyhow::Result { + let client = self + .inner + .dialect_mut() + .client + .clone() + .with_timeout(timeout)?; + self.inner.dialect_mut().client = client; + Ok(self) + } + + fn new(endpoint: &str, api_key: Option<&str>) -> anyhow::Result { + Ok(Self { + inner: RemoteMemory::new(CortexDialect { + client: HttpClient::bearer(endpoint, api_key)?, + }), + }) + } + + /// Connect to a CortexDB deployment using bearer authentication. + /// + /// # Errors + /// + /// Returns an error when `endpoint` is invalid or `api_key` is blank. + pub fn api(endpoint: &str, api_key: &str) -> anyhow::Result { + anyhow::ensure!( + !api_key.trim().is_empty(), + "cortex API key must not be empty" + ); + Self::new(endpoint, Some(api_key)) + } + + /// Connect to a self-hosted CortexDB server. + /// + /// # Errors + /// + /// Returns an error when `endpoint` is invalid or `api_key` is blank. + pub fn self_hosted(endpoint: &str, api_key: &str) -> anyhow::Result { + Self::api(endpoint, api_key) + } + + /// Connect to CortexDB's managed API endpoint. + /// + /// # Errors + /// + /// Returns an error when `api_key` is blank. + pub fn cloud(api_key: &str) -> anyhow::Result { + Self::api(CORTEX_API_ENDPOINT, api_key) + } +} + +/// The payload this adapter writes into an event's message text. +/// +/// CortexDB's experience envelope is a **closed schema** — an unknown field is +/// refused with `422` — so there is nowhere on the event itself to record which +/// TinyMemory record it is. The engine treats `content.text` as free-form, so +/// the record rides there and the log stays parseable by us alone. +/// +/// Anything the log cannot carry natively goes here: the key that identifies +/// the record, the category, the session, and the taint. +#[derive(Debug, serde::Serialize, serde::Deserialize)] +struct Envelope { + /// TinyMemory's logical key. The whole reason this wrapper exists. + k: String, + /// The caller's content, untouched. + c: String, + /// Category, as its wire string. + #[serde(default)] + cat: Option, + /// Session id, when the caller supplied one. + #[serde(default)] + s: Option, + /// Provenance taint. Persisted rather than dropped, because the default + /// `store_with_taint` silently launders `ExternalSync` into internal trust. + #[serde(default)] + t: Option, + /// Tombstone marker. A `true` here means "this key is deleted as of this + /// event". The fold reads newest-wins, so a tombstone written after the + /// last value makes the key read as absent even if the underlying events + /// are still on disk. See [`Dialect::delete`] for why we write one. + #[serde(default)] + d: bool, +} + +/// One event as the fold needs to see it. +struct Folded { + order: u64, + /// `None` when the newest version of this key is a tombstone. + entry: Option, +} + +#[derive(Debug)] +struct CortexDialect { + client: HttpClient, +} + +impl CortexDialect { + /// Maps a TinyMemory namespace onto a CortexDB scope path, reversibly. + /// + /// The two formats are incompatible and the translation is **not** cosmetic. + /// CortexDB scopes are slash-delimited `type:id` segments matching + /// `^[a-z][a-z0-9_]{0,31}:[A-Za-z0-9_-]{1,128}(/…){0,31}$`. TinyMemory + /// namespaces are plain slash-delimited words with no colon anywhere, so + /// passing one through unchanged is refused by the engine on every call. + /// + /// Reversibility is the load-bearing half. A host re-checks each returned + /// record against the namespace it asked for and drops mismatches, so a + /// scope this adapter cannot map *back* yields zero hits silently — a worse + /// failure than a rejection, because nothing reports it. + fn scope_of(namespace: &str) -> anyhow::Result { + let mut out = Vec::new(); + for segment in namespace.split('/').filter(|s| !s.is_empty()) { + let safe = segment + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_'); + let encoded = if safe { + format!("{SEGMENT_PLAIN}:{segment}") + } else { + // Hex, so the result is unambiguous and cannot itself contain a + // character the grammar rejects. Two bytes out per byte in. + let mut hex = String::with_capacity(segment.len() * 2); + for byte in segment.as_bytes() { + hex.push_str(&format!("{byte:02x}")); + } + format!("{SEGMENT_ENCODED}:{hex}") + }; + let id_len = encoded.len() - encoded.find(':').unwrap_or(0) - 1; + anyhow::ensure!( + id_len <= 128, + "namespace segment `{segment}` does not fit CortexDB's 128-character \ + scope id limit once encoded" + ); + out.push(encoded); + } + anyhow::ensure!(!out.is_empty(), "namespace must not be empty"); + // Measured against a running engine: 32 segments are accepted, 33 are + // refused with `422 INVALID_BODY`. Catching it here keeps the refusal + // local and specific, which is the same reason the per-segment checks + // above are not left to the wire. + anyhow::ensure!( + out.len() <= MAX_SCOPE_SEGMENTS, + "namespace has {} segments; CortexDB scope paths hold at most {MAX_SCOPE_SEGMENTS}", + out.len() + ); + Ok(out.join("/")) + } + + /// The inverse of [`Self::scope_of`]. + /// + /// Returns `None` for a scope this adapter did not write, which is what + /// keeps `scopes()` from reporting somebody else's Cortex scopes as + /// namespaces of ours. + fn namespace_of(scope: &str) -> Option { + let mut out = Vec::new(); + for segment in scope.split('/').filter(|s| !s.is_empty()) { + let (kind, body) = segment.split_once(':')?; + match kind { + SEGMENT_PLAIN => out.push(body.to_string()), + SEGMENT_ENCODED => { + if body.len() % 2 != 0 { + return None; + } + let mut bytes = Vec::with_capacity(body.len() / 2); + for pair in body.as_bytes().chunks(2) { + let pair = std::str::from_utf8(pair).ok()?; + bytes.push(u8::from_str_radix(pair, 16).ok()?); + } + out.push(String::from_utf8(bytes).ok()?); + } + _ => return None, + } + } + (!out.is_empty()).then(|| out.join("/")) + } + + /// Every distinct event in one scope, following `next_cursor` to the end. + /// + /// Two things about the listing are worth stating, because both are easy to + /// get wrong and neither is visible in a single-page test: + /// + /// - Paging is `cursor`/`next_cursor`. Unknown query parameters are ignored + /// rather than refused, so a wrong parameter name does not fail — it + /// silently re-serves page one until the page ceiling trips. + /// - The engine emits **every record twice** in `items`, and `limit` counts + /// the duplicates, so a page of `PAGE_SIZE` carries about half that many + /// distinct events. The cursor itself is honest: paged to the end, every + /// record is present. We drop the repeats by event id here so no caller + /// downstream has to know. + async fn events(&self, scope: &str) -> anyhow::Result> { + let mut all = Vec::new(); + let mut seen = std::collections::HashSet::new(); + let mut cursor: Option = None; + for _ in 0..MAX_PAGES { + let path = match &cursor { + Some(c) => format!( + "v1/events?scope={scope}&limit={PAGE_SIZE}&cursor={cursor}", + scope = urlencoding(scope), + cursor = urlencoding(c) + ), + None => format!( + "v1/events?scope={scope}&limit={PAGE_SIZE}", + scope = urlencoding(scope) + ), + }; + let page: Value = self + .client + .json(Method::GET, &path, None, Attempts::RetryTransient) + .await?; + let items = page + .get("items") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + for item in items { + match item.get("id").and_then(Value::as_str) { + Some(id) if !seen.insert(id.to_string()) => continue, + _ => all.push(item), + } + } + let next = page + .get("next_cursor") + .and_then(Value::as_str) + .map(str::to_string); + match (page.get("has_more").and_then(Value::as_bool), next) { + (Some(true), Some(next)) => cursor = Some(next), + _ => return Ok(all), + } + } + anyhow::bail!( + "listing scope `{scope}` exceeded {MAX_PAGES} pages; refusing to answer from a \ + truncated log, because a short listing would report a superseded value as current" + ) + } + + /// Blocks until an appended event can be read back by every read path. + /// + /// `/v1/experience` answers `202 captured` and indexes afterwards, so a + /// write that has been accepted is not yet a write that can be read. The + /// contract requires read-after-write, and this dialect has two read paths + /// that become ready at different times, so both are waited on. + /// + /// Four details decide the shape of this, and all four were measured + /// against a running engine rather than assumed: + /// + /// - `GET /v1/events/{id}` starts answering roughly 1.3s **before** the + /// scope listing carries the same event, so it is not a usable readiness + /// probe for `get`/`list`, which read the listing. + /// - Ranked recall lags the listing by about another second, so waiting on + /// the listing alone leaves `search` returning nothing for a record the + /// same adapter has just reported as stored. + /// - `/v1/experience/status` never advances past `captured`. It reports + /// durability, which is already true when the write returns, and says + /// nothing about visibility. + /// - Recall answers carry the event id, so the second wait can be exact + /// rather than a sleep: query with the text just written and look for the + /// id that came back from the write. + /// + /// The two waits do not fail the same way, and that asymmetry is the point. + /// A record that cannot be read by key has not been stored as far as the + /// contract is concerned, so phase one times out into an error. A search + /// index that has not caught up is a weaker claim — the record is durable + /// and keyed reads return it — so phase two gives up quietly rather than + /// failing a write that succeeded. + async fn await_readable(&self, scope: &str, event_id: &str, text: &str) -> anyhow::Result<()> { + let deadline = std::time::Instant::now() + VISIBILITY_TIMEOUT; + + // Phase one: the keyed read path, which folds the scope listing. + loop { + // Newest first, so one page is enough to see a write just made. + let path = format!( + "v1/events?scope={scope}&limit={PAGE_SIZE}", + scope = urlencoding(scope) + ); + let page: Value = self + .client + .json(Method::GET, &path, None, Attempts::RetryTransient) + .await?; + if Self::carries(page.get("items"), event_id) { + break; + } + Self::still_waiting(deadline, event_id, scope)?; + tokio::time::sleep(VISIBILITY_POLL).await; + } + + // Phase two: ranked recall, a separate index that settles later. + // + // Unlike phase one this is best-effort, and the difference is + // deliberate. A write whose record cannot be read by key has not + // happened as far as the contract is concerned, so phase one failing + // is an error. A search index that has not caught up yet is a + // different thing: the record is durable, keyed reads return it, and + // the ranking will include it shortly. Failing the write there would + // report a successful, readable store as an error. + let query: String = text.chars().take(RECALL_QUERY_CAP).collect(); + let settle_by = std::time::Instant::now() + RECALL_SETTLE_TIMEOUT; + while std::time::Instant::now() < settle_by { + let probe = self + .client + .json::( + Method::POST, + "v1/recall", + Some(&json!({ "scope": scope, "query": query })), + Attempts::RetryTransient, + ) + .await; + let Ok(answer) = probe else { + // Best-effort means best-effort. A probe that could not be + // answered says nothing about the write, which is durable and + // already readable by key — propagating this would report a + // successful store as a failure, which is the one thing this + // phase is documented not to do. + break; + }; + if Self::carries(answer.pointer("/layers/events"), event_id) { + break; + } + tokio::time::sleep(VISIBILITY_POLL).await; + } + Ok(()) + } + + /// Whether a listing or recall answer contains this event id. + fn carries(items: Option<&Value>, event_id: &str) -> bool { + items.and_then(Value::as_array).is_some_and(|items| { + items + .iter() + .any(|e| e.get("id").and_then(Value::as_str) == Some(event_id)) + }) + } + + /// Errors once the visibility deadline has passed, naming the read path + /// that never caught up. + fn still_waiting( + deadline: std::time::Instant, + event_id: &str, + scope: &str, + ) -> anyhow::Result<()> { + anyhow::ensure!( + std::time::Instant::now() < deadline, + "event `{event_id}` was accepted into scope `{scope}` but did not become \ + readable within {VISIBILITY_TIMEOUT:?}; reporting the write as succeeded \ + would break read-after-write" + ); + Ok(()) + } + + /// Reads our envelope out of one event's text, whichever read path it came + /// from. + /// + /// The two paths do not agree on the bytes. `/v1/events` returns the text + /// exactly as stored; `/v1/recall` renders it for a reader first, prefixing + /// the speaker as `[user] `. Parsing the raw form only is therefore a + /// dialect that lists correctly and searches to nothing — every recall hit + /// fails to parse and is dropped as somebody else's event, which looks like + /// an empty index rather than a bug. + /// + /// A prefix is only stripped when the text does not parse without it, so an + /// envelope whose content legitimately begins with a bracket is untouched. + fn envelope_of(text: &str) -> Option { + if let Ok(envelope) = serde_json::from_str::(text) { + return Some(envelope); + } + let rendered = text.strip_prefix('[')?; + let (_role, rest) = rendered.split_once("] ")?; + serde_json::from_str::(rest).ok() + } + + /// Folds a scope's log down to one record per logical key, newest wins. + /// + /// This is where the contract's replace semantics are reconstructed. The + /// engine keeps every version; `wal_offset` orders them, and the highest + /// offset for a key is the value a caller should see. + fn fold(namespace: &str, events: &[Value]) -> Vec { + let mut latest: std::collections::HashMap = + std::collections::HashMap::new(); + for event in events { + let Some(text) = event.pointer("/content/text").and_then(Value::as_str) else { + continue; + }; + // Anything this adapter did not write is not ours to interpret. + // A scope may hold events from a person using CortexDB directly. + let Some(envelope) = Self::envelope_of(text) else { + continue; + }; + let order = event + .get("wal_offset") + .and_then(Value::as_u64) + .unwrap_or_default(); + let replace = latest + .get(&envelope.k) + .is_none_or(|held| order >= held.order); + if !replace { + continue; + } + if envelope.d { + // Newest version of this key is a tombstone: the key is gone. + latest.insert(envelope.k, Folded { order, entry: None }); + continue; + } + let entry = StoredEntry { + remote_id: event + .get("id") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + namespace: namespace.to_string(), + key: envelope.k.clone(), + content: envelope.c, + category: crate::common::category(envelope.cat.as_deref()), + timestamp: event + .pointer("/context/recorded_at") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + session_id: envelope.s, + score: None, + taint: match envelope.t.as_deref() { + Some("external_sync") => MemoryTaint::ExternalSync, + _ => MemoryTaint::Internal, + }, + }; + latest.insert( + envelope.k, + Folded { + order, + entry: Some(entry), + }, + ); + } + let mut out: Vec = latest.into_values().filter_map(|f| f.entry).collect(); + out.sort_by(|a, b| a.key.cmp(&b.key)); + out + } + + /// Every scope this deployment holds that this adapter wrote. + async fn scopes(&self) -> anyhow::Result> { + let listing: Value = self + .client + .json( + Method::GET, + "v1/scopes/list", + None, + Attempts::RetryTransient, + ) + .await?; + Ok(listing + .get("items") + .and_then(Value::as_array) + .map(|items| { + items + .iter() + .filter_map(|s| s.get("path").and_then(Value::as_str)) + .filter_map(Self::namespace_of) + .collect() + }) + .unwrap_or_default()) + } +} + +/// A fresh idempotency key for every write. +/// +/// Never TinyMemory's key: reusing that is what produces +/// `409 IDEMPOTENCY_CONFLICT` on the second store, and the whole append-and-fold +/// design exists to avoid it. +/// +/// Three parts, because two are not enough. The counter separates writes within +/// one process, and the timestamp separates runs of it — but neither separates +/// two *concurrent* processes, which can read the same nanosecond and start +/// their counters at the same zero. The salt is per-process entropy from the +/// standard library, taken once, so independent writers cannot mint the same +/// key. Getting that wrong is not a silent fault — a reused key carrying a +/// different body is refused with a 409 — but it is a refusal of a legitimate +/// write, and it would be maddening to diagnose. +/// +/// No new dependency: `RandomState` is seeded by the OS per process, which is +/// exactly the entropy needed here, and a vendored crate should not grow a +/// dependency for one string. +fn fresh_idempotency_key() -> String { + use std::hash::{BuildHasher, Hasher}; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::sync::OnceLock; + + static SEQ: AtomicU64 = AtomicU64::new(0); + static SALT: OnceLock = OnceLock::new(); + + let salt = *SALT.get_or_init(|| { + std::collections::hash_map::RandomState::new() + .build_hasher() + .finish() + }); + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or_default(); + format!( + "tm-{salt:016x}-{nanos}-{}", + SEQ.fetch_add(1, Ordering::Relaxed) + ) +} + +/// Percent-encodes everything outside the URI unreserved set. +/// +/// A scope only ever carries `:` and `/`, so an escape list would do for that. +/// The cursor is the reason this is general: it is opaque engine output, and a +/// `+`, `&`, `=`, `#` or `?` in one would silently reshape the query string +/// rather than fail. Encoding by byte also keeps multi-byte UTF-8 correct. +fn urlencoding(value: &str) -> String { + let mut out = String::with_capacity(value.len()); + for byte in value.as_bytes() { + match byte { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => { + out.push(char::from(*byte)); + } + other => out.push_str(&format!("%{other:02X}")), + } + } + out +} + +#[async_trait] +impl Dialect for CortexDialect { + fn name(&self) -> &'static str { + CORTEX_DRIVER_ID + } + + /// Appends a new event carrying the record. + /// + /// Deliberately **not** an update. Every write gets a fresh idempotency key, + /// which the engine always accepts; reusing TinyMemory's key here is what + /// produces `409 IDEMPOTENCY_CONFLICT` on the second store. The previous + /// version stays in the log and is folded out on read. + async fn upsert(&self, entry: StoredEntry) -> anyhow::Result<()> { + let scope = Self::scope_of(&entry.namespace)?; + let envelope = serde_json::to_string(&Envelope { + k: entry.key.clone(), + c: entry.content.clone(), + cat: Some(entry.category.to_string()), + s: entry.session_id.clone(), + t: Some( + match entry.taint { + MemoryTaint::ExternalSync => "external_sync", + _ => "internal", + } + .to_string(), + ), + d: false, + })?; + let accepted: Value = self + .client + .json( + Method::POST, + "v1/experience", + Some(&json!({ + "scope": scope, + "modality": "observation", + // Fresh per write. See this method's own doc. + "idempotency_key": fresh_idempotency_key(), + "content": { "kind": "message", "role": "user", "text": envelope }, + "context": {}, + })), + Attempts::Once, + ) + .await?; + // Accepted is not readable yet — see `await_readable`. + if let Some(id) = accepted.get("event_id").and_then(Value::as_str) { + self.await_readable(&scope, id, &envelope).await?; + } + Ok(()) + } + + async fn entries(&self) -> anyhow::Result> { + let mut all = Vec::new(); + for namespace in self.scopes().await? { + all.extend(self.namespace_entries(&namespace).await?); + } + Ok(all) + } + + async fn namespace_entries(&self, namespace: &str) -> anyhow::Result> { + let scope = Self::scope_of(namespace)?; + let events = self.events(&scope).await?; + Ok(Self::fold(namespace, &events)) + } + + /// Deletes a key in two moves: a tombstone, then the events behind it. + /// + /// The tombstone goes first and is what makes the delete correct. It is an + /// ordinary append, so it cannot fail for any reason a write could not + /// already fail, and once it lands the fold reports the key absent — + /// whatever happens to the second move, and whatever a concurrent writer + /// was doing at the time. + /// + /// The second move is the real removal, and it is the one that matters for + /// [`Dialect::search`]: recall ranks over every event, so leaving the old + /// versions in place would let a deleted record resurface through `recall` + /// long after `get` stopped returning it. We name the events explicitly in + /// `selector.memory_ids` and send **no** `confirm_all` — that flag + /// authorises a scope-wide wipe, is only valid with an empty selector, and + /// pairing it with a selector is refused as ambiguous. Note that an + /// unrecognised selector field is not an error: the engine reads the + /// selector as empty, which is exactly the shape `confirm_all` would then + /// license. The two mistakes are only dangerous together, and this is why + /// the flag is not written anywhere in this file. + async fn delete(&self, namespace: &str, key: &str) -> anyhow::Result { + let scope = Self::scope_of(namespace)?; + let events = self.events(&scope).await?; + let mut ids = Vec::new(); + let mut live = false; + for event in &events { + let Some(text) = event.pointer("/content/text").and_then(Value::as_str) else { + continue; + }; + let Some(envelope) = Self::envelope_of(text) else { + continue; + }; + if envelope.k != key { + continue; + } + if !envelope.d { + live = true; + } + if let Some(id) = event.get("id").and_then(Value::as_str) { + ids.push(id.to_string()); + } + } + if !live { + // Never held, or already tombstoned. Either way there is nothing to + // delete, and appending a second tombstone would only add noise. + return Ok(false); + } + + let tombstone = serde_json::to_string(&Envelope { + k: key.to_string(), + c: String::new(), + cat: None, + s: None, + t: None, + d: true, + })?; + let accepted: Value = self + .client + .json( + Method::POST, + "v1/experience", + Some(&json!({ + "scope": scope, + "modality": "conversation", + "idempotency_key": fresh_idempotency_key(), + "content": { "kind": "message", "role": "user", "text": tombstone }, + "context": {}, + })), + Attempts::Once, + ) + .await?; + // The tombstone IS the delete; the next read must see it. + if let Some(id) = accepted.get("event_id").and_then(Value::as_str) { + self.await_readable(&scope, id, &tombstone).await?; + } + + if !ids.is_empty() { + self.client + .empty( + Method::POST, + "v1/forget", + Some(&json!({ + "scope": scope, + "layers": ["events"], + "selector": { "memory_ids": ids }, + "audit_note": "tinymemory: delete(namespace, key)", + })), + ) + .await?; + } + Ok(true) + } + + async fn search( + &self, + query: &str, + limit: usize, + opts: RecallOpts<'_>, + ) -> anyhow::Result> { + let Some(namespace) = opts.namespace else { + // Recall is scope-addressed here; an unscoped search has no scope to + // name. Empty rather than an error, matching the contract's rule + // that a non-matching query yields no hits. + return Ok(Vec::new()); + }; + let scope = Self::scope_of(namespace)?; + let answer: Value = self + .client + .json( + Method::POST, + "v1/recall", + Some(&json!({ "scope": scope, "query": query })), + Attempts::RetryTransient, + ) + .await?; + let events = answer + .pointer("/layers/events") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + // `fold` orders by key. That is right for a listing and wrong here: + // the engine ranked these, and truncating an alphabetical order would + // discard its best hits and keep whichever keys sort first. Restore the + // ranking, by each key's first appearance in the answer, before the cap. + let mut rank: std::collections::HashMap = std::collections::HashMap::new(); + for (position, event) in events.iter().enumerate() { + let Some(envelope) = event + .pointer("/content/text") + .and_then(Value::as_str) + .and_then(Self::envelope_of) + else { + continue; + }; + // First appearance wins: a key may have several versions in the + // answer, and the best-ranked one is the one that places it. + rank.entry(envelope.k).or_insert(position); + } + let mut hits = Self::fold(namespace, &events); + hits.sort_by_key(|hit| rank.get(hit.key.as_str()).copied().unwrap_or(usize::MAX)); + hits.truncate(limit); + Ok(hits) + } + + async fn health(&self) -> anyhow::Result<()> { + self.client.probe("v1/admin/health").await + } + + /// CortexDB's recall answers with no per-hit similarity score — the + /// response carries no score field at all — so a `min_score` filter would + /// drop every hit rather than narrow them. + fn scores_recall(&self) -> bool { + false + } +} + +#[async_trait] +impl Memory for CortexMemory { + fn name(&self) -> &str { + self.inner.name() + } + async fn store( + &self, + n: &str, + k: &str, + c: &str, + cat: MemoryCategory, + s: Option<&str>, + ) -> anyhow::Result<()> { + self.inner.store(n, k, c, cat, s).await + } + async fn store_with_taint( + &self, + n: &str, + k: &str, + c: &str, + cat: MemoryCategory, + s: Option<&str>, + taint: MemoryTaint, + ) -> anyhow::Result<()> { + self.inner.store_with_taint(n, k, c, cat, s, taint).await + } + async fn get( + &self, + n: &str, + k: &str, + ) -> anyhow::Result> { + self.inner.get(n, k).await + } + async fn forget(&self, n: &str, k: &str) -> anyhow::Result { + self.inner.forget(n, k).await + } + async fn list( + &self, + n: Option<&str>, + cat: Option<&MemoryCategory>, + s: Option<&str>, + ) -> anyhow::Result> { + self.inner.list(n, cat, s).await + } + async fn namespace_summaries( + &self, + ) -> anyhow::Result> { + self.inner.namespace_summaries().await + } + async fn count(&self) -> anyhow::Result { + self.inner.count().await + } + async fn health_check(&self) -> bool { + self.inner.health_check().await + } + async fn recall( + &self, + query: &str, + limit: usize, + opts: RecallOpts<'_>, + ) -> anyhow::Result> { + self.inner.recall(query, limit, opts).await + } +} + +#[cfg(test)] +#[path = "cortex_test.rs"] +mod test; diff --git a/crates/tinymemory-remote/src/cortex_test.rs b/crates/tinymemory-remote/src/cortex_test.rs new file mode 100644 index 00000000..397f7c2c --- /dev/null +++ b/crates/tinymemory-remote/src/cortex_test.rs @@ -0,0 +1,180 @@ +//! CortexDB adapter unit tests. +//! +//! These cover the pure functions the adapter reconstructs the contract +//! with — scope mapping, the fold, and the two shapes a stored envelope +//! comes back in. Behaviour against a live engine is in +//! `tests/live_remote_engines.rs`. + +// Test-only, and deliberately narrow: `expect` here names the invariant a +// failing setup step or assertion violated, which is the diagnostic a reader of +// the failure output needs. Production paths in this crate return `Result`. +#![allow(clippy::expect_used)] + +use super::*; +#[test] +fn a_namespace_maps_to_a_scope_and_back() { + let namespace = "oc/acme-0123456789abcdef0123456789abcdef/facts"; + let scope = CortexDialect::scope_of(namespace).expect("maps"); + assert_eq!( + scope, + "tm:oc/tm:acme-0123456789abcdef0123456789abcdef/tm:facts" + ); + assert_eq!( + CortexDialect::namespace_of(&scope).as_deref(), + Some(namespace), + "the round trip is load-bearing: a scope we cannot map back yields zero hits \ + silently, because the host re-checks every returned record against the \ + namespace it asked for" + ); +} + +#[test] +fn a_segment_cortex_would_reject_is_encoded_rather_than_refused() { + // The contract allows characters a Cortex scope id does not, and `:` is the + // one that matters: it addresses a namespace *section*. Refusing it would + // make whole sections unstorable, and collapsing it would silently + // re-address the namespace out of its section — which is exactly the + // regression `assert_namespaces_preserve_their_section` exists to catch. + for namespace in [ + "conversation:tinymemory-conformance/cortex/section-thread", + "oc/has space/facts", + "oc/ünicode/facts", + ] { + let mapped = CortexDialect::scope_of(namespace); + assert!( + mapped.is_ok(), + "`{namespace}` should encode, not refuse: {mapped:?}" + ); + let scope = mapped.expect("checked on the line above"); + assert_eq!( + CortexDialect::namespace_of(&scope).as_deref(), + Some(namespace), + "`{namespace}` did not survive the round trip through `{scope}`" + ); + } + + // A scope id is capped at 128 characters, and encoding doubles the ones it + // applies to — so the ceiling is real, and lower for an encoded segment. + assert!(CortexDialect::scope_of(&format!("oc/{}", "x".repeat(129))).is_err()); + assert!(CortexDialect::scope_of(&format!("oc/{}", ":".repeat(65))).is_err()); + assert!(CortexDialect::scope_of("").is_err()); + + // A scope this adapter did not write must not decode into a namespace. + assert_eq!(CortexDialect::namespace_of("user:alice/notes"), None); +} + +#[test] +fn the_fold_keeps_the_newest_write_per_key() { + // The whole contract this adapter reconstructs: the engine holds both + // versions, and a caller must see only the second. + let events = vec![ + json!({ + "id": "evt_1", "wal_offset": 10, + "content": { "text": r#"{"k":"billing-owner","c":"Ana"}"# }, + "context": { "recorded_at": "2026-09-02T00:00:00Z" } + }), + json!({ + "id": "evt_2", "wal_offset": 20, + "content": { "text": r#"{"k":"billing-owner","c":"Dev"}"# }, + "context": { "recorded_at": "2026-09-02T00:01:00Z" } + }), + json!({ + "id": "evt_3", "wal_offset": 30, + "content": { "text": r#"{"k":"oncall","c":"Priya"}"# }, + "context": { "recorded_at": "2026-09-02T00:02:00Z" } + }), + ]; + let folded = CortexDialect::fold("oc/acme/facts", &events); + assert_eq!( + folded.len(), + 2, + "one row per logical key, not one per write" + ); + let owner = folded + .iter() + .find(|e| e.key == "billing-owner") + .expect("key"); + assert_eq!(owner.content, "Dev", "the later write wins"); + assert_eq!(owner.remote_id, "evt_2"); +} + +#[test] +fn the_fold_ignores_events_this_adapter_did_not_write() { + // A scope can hold events written by someone using CortexDB directly. + // Those are not ours to interpret, and must not become phantom records. + let events = vec![ + json!({ "id": "evt_1", "wal_offset": 1, + "content": { "text": "just a sentence someone typed" } }), + json!({ "id": "evt_2", "wal_offset": 2, + "content": { "text": r#"{"unrelated":"json"}"# } }), + ]; + assert!(CortexDialect::fold("oc/acme/facts", &events).is_empty()); +} + +#[test] +fn taint_survives_the_envelope() { + let events = vec![json!({ + "id": "evt_1", "wal_offset": 1, + "content": { "text": r#"{"k":"a","c":"b","t":"external_sync"}"# } + })]; + let folded = CortexDialect::fold("oc/acme/facts", &events); + assert_eq!(folded[0].taint, MemoryTaint::ExternalSync); +} + +#[test] +fn a_cursor_is_escaped_far_beyond_the_characters_a_scope_carries() { + // A scope only ever holds `:` and `/`, so an escape list would cover it. + assert_eq!( + urlencoding("tm:oc/tm:acme-1/tm:facts"), + "tm%3Aoc%2Ftm%3Aacme-1%2Ftm%3Afacts" + ); + // The cursor is opaque engine output, and these are the characters that + // would silently reshape a query string rather than fail. + assert_eq!(urlencoding("a+b&c=d#e?f"), "a%2Bb%26c%3Dd%23e%3Ff"); + // Unreserved characters must survive untouched, or every request grows. + assert_eq!(urlencoding("Az09-._~"), "Az09-._~"); + // Encoding is by byte, so multi-byte UTF-8 stays recoverable. + assert_eq!(urlencoding("é"), "%C3%A9"); +} + +#[test] +fn a_scope_path_deeper_than_the_engine_accepts_is_refused_here() { + // Measured against a running engine: 32 segments are accepted, 33 come back + // as `422 INVALID_BODY`. Refusing locally keeps the error specific instead + // of surfacing as a generic body rejection from the wire. + let deep = (0..32) + .map(|i| format!("s{i}")) + .collect::>() + .join("/"); + assert!( + CortexDialect::scope_of(&deep).is_ok(), + "32 segments are within the grammar and must not be refused" + ); + + let deeper = (0..33) + .map(|i| format!("s{i}")) + .collect::>() + .join("/"); + assert!( + CortexDialect::scope_of(&deeper).is_err(), + "33 segments exceed what the engine accepts and must be refused here" + ); +} + +#[test] +fn two_writers_in_one_process_never_mint_the_same_key() { + // The counter covers this much; the per-process salt covers the case no + // unit test can reach, which is a second process minting concurrently. + let keys: std::collections::HashSet = + (0..1000).map(|_| fresh_idempotency_key()).collect(); + assert_eq!(keys.len(), 1000, "an idempotency key was reused"); + + // The salt is stable within a process, so a key is greppable back to the + // run that wrote it. + let salt_of = |k: &str| k.split('-').nth(1).map(str::to_string); + assert_eq!( + salt_of(&fresh_idempotency_key()), + salt_of(&fresh_idempotency_key()), + "the salt identifies the process and must not change between writes" + ); +} diff --git a/crates/tinymemory-remote/src/lib.rs b/crates/tinymemory-remote/src/lib.rs index 9f2b596c..ff615576 100644 --- a/crates/tinymemory-remote/src/lib.rs +++ b/crates/tinymemory-remote/src/lib.rs @@ -11,6 +11,7 @@ pub mod cognee; mod cognee_graph; mod common; +pub mod cortex; mod graph_provider; pub mod mem0; mod mem0_graph; @@ -20,6 +21,7 @@ pub mod supermemory; pub use agentmemory::{AgentMemoryMemory, AGENTMEMORY_API_ENDPOINT, AGENTMEMORY_DRIVER_ID}; pub use cognee::{CogneeMemory, COGNEE_DRIVER_ID}; pub use cognee_graph::CogneeGraph; +pub use cortex::{CortexMemory, CORTEX_API_ENDPOINT, CORTEX_DRIVER_ID}; pub use graph_provider::GraphMemoryProvider; pub use mem0::{Mem0Memory, MEM0_API_ENDPOINT, MEM0_DRIVER_ID}; pub use mem0_graph::Mem0Graph; @@ -48,6 +50,12 @@ pub fn cognee_provider(memory: CogneeMemory) -> MemoryTraitProvider { MemoryTraitProvider::new(Arc::new(memory), COGNEE_DRIVER_ID) } +/// Wrap a CortexDB HTTP backend as a bound TinyMemory provider. +#[must_use] +pub fn cortex_provider(memory: CortexMemory) -> MemoryTraitProvider { + MemoryTraitProvider::new(Arc::new(memory), CORTEX_DRIVER_ID) +} + /// Wrap an AgentMemory HTTP backend as a bound TinyMemory provider. #[must_use] pub fn agentmemory_provider(memory: AgentMemoryMemory) -> MemoryTraitProvider { diff --git a/crates/tinymemory-remote/tests/live_remote_engines.rs b/crates/tinymemory-remote/tests/live_remote_engines.rs index 6aed17be..31055a17 100644 --- a/crates/tinymemory-remote/tests/live_remote_engines.rs +++ b/crates/tinymemory-remote/tests/live_remote_engines.rs @@ -24,7 +24,7 @@ use std::sync::Arc; -use tinymemory_remote::{supermemory_provider, SupermemoryMemory}; +use tinymemory_remote::{cortex_provider, supermemory_provider, CortexMemory, SupermemoryMemory}; /// Reads one engine's endpoint and key, or `None` when either is unset. /// @@ -48,3 +48,23 @@ async fn live_supermemory_upholds_the_provider_contract() -> anyhow::Result<()> tinymemory_conformance::assert_provider(Arc::new(provider)).await; Ok(()) } + +/// Runs the full provider contract against a live CortexDB. +/// +/// Worth more here than for the keyed engines. The Cortex adapter emulates +/// replacement over an append-only log, so almost everything the contract +/// checks is reconstructed on the read side against behaviour the double can +/// only assert from documentation — paging, listing duplicates, the shape of +/// the destructive selector. Each of those was wrong in the double at some +/// point, and the offline suite was green throughout. +/// +/// Skipped without `TINYMEMORY_TEST_CORTEX_URL` and `..._KEY`. +#[tokio::test] +async fn live_cortex_upholds_the_provider_contract() -> anyhow::Result<()> { + let Some((url, key)) = credentials("CORTEX") else { + return Ok(()); + }; + let provider = cortex_provider(CortexMemory::api(&url, &key)?); + tinymemory_conformance::assert_provider(Arc::new(provider)).await; + Ok(()) +}