From f9af052aed93eca79649d4d69afdaaf6da49db86 Mon Sep 17 00:00:00 2001 From: Ghost Scripter Date: Wed, 2 Sep 2026 19:10:20 +0530 Subject: [PATCH 1/5] Add a CortexDB adapter over its append-only event log CortexDB is not a keyed store. A key cannot be rewritten: reusing an idempotency key with a different body is refused with 409, there is no update route, and the idempotency ledger survives /v1/forget, so delete-then-rewrite loses the old value and still refuses the new one. So the dialect never asks the engine to replace anything. Every store appends a fresh event carrying the logical key in its payload, and reads fold the log down to newest-per-key. Replacement is reconstructed on the read side rather than performed on the write side. Three engine behaviours shape the rest of it, all measured against a running v0.9.8 rather than taken from the documentation: - writes are asynchronous. /v1/experience answers 202 and indexes after, so a write returns before any read can see it. The contract requires read-after-write, so upsert waits for the record to become readable. - the two read paths return different bytes for the same event. The listing returns stored text; recall renders it with the speaker prefixed. Parsing only the stored form yields a dialect that lists correctly and searches to nothing. - the listing emits every record twice and pages by cursor, and unknown query parameters are ignored rather than refused. Not wired into any registry yet: the upsert gap above is an open question with the vendor, and binding a driver that emulates replacement is a decision to take once that is answered. --- crates/tinymemory-remote/src/cortex.rs | 908 ++++++++++++++++++++ crates/tinymemory-remote/src/cortex_test.rs | 119 +++ crates/tinymemory-remote/src/lib.rs | 8 + 3 files changed, 1035 insertions(+) create mode 100644 crates/tinymemory-remote/src/cortex.rs create mode 100644 crates/tinymemory-remote/src/cortex_test.rs diff --git a/crates/tinymemory-remote/src/cortex.rs b/crates/tinymemory-remote/src/cortex.rs new file mode 100644 index 00000000..c6d5d132 --- /dev/null +++ b/crates/tinymemory-remote/src/cortex.rs @@ -0,0 +1,908 @@ +//! 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 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"); + 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 answer: Value = self + .client + .json( + Method::POST, + "v1/recall", + Some(&json!({ "scope": scope, "query": query })), + Attempts::RetryTransient, + ) + .await?; + 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 value no other write will reuse. +/// +/// Not a content hash and not TinyMemory's key: the engine refuses a reused key +/// carrying a different body, so re-deriving this from the record is exactly +/// what breaks the second store. Monotonic clock plus a process-local counter +/// keeps it unique within a run and across runs without adding a dependency to +/// a vendored crate for one string. +fn fresh_idempotency_key() -> String { + use std::sync::atomic::{AtomicU64, Ordering}; + static SEQ: AtomicU64 = AtomicU64::new(0); + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or_default(); + format!("tm-{nanos}-{}", SEQ.fetch_add(1, Ordering::Relaxed)) +} + +/// Percent-encodes the characters a scope path carries that a query string +/// would otherwise read as structure. +fn urlencoding(value: &str) -> String { + value + .chars() + .map(|c| match c { + ':' => "%3A".to_string(), + '/' => "%2F".to_string(), + other => other.to_string(), + }) + .collect() +} + +#[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(); + let mut hits = Self::fold(namespace, &events); + 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..a3445ed4 --- /dev/null +++ b/crates/tinymemory-remote/src/cortex_test.rs @@ -0,0 +1,119 @@ +//! 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`. + +#![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); +} 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 { From eb10b0355aedc56478c7695c687ceb0b461d1901 Mon Sep 17 00:00:00 2001 From: Ghost Scripter Date: Wed, 2 Sep 2026 19:10:41 +0530 Subject: [PATCH 2/5] Reproduce CortexDB's live HTTP behaviour in the conformance double The double was written from the same documentation as the adapter, so the two agreed with each other and the suite stayed green while the adapter was wrong against the engine in four separate ways. A double that only restates the documentation cannot catch that. Each of these was then observed on a live v0.9.8 and is now reproduced: - recall renders content for a reader, prefixing the speaker, while the listing returns it as stored - the listing emits every record twice, counts the duplicates against limit, pages by cursor/next_cursor and orders newest first - the forget selector's id field is memory_ids; an unrecognised field deserialises to an empty selector, meaning the whole scope, and the two interlocks that make that survivable are enforced here too - /v1/experience returns the id the event was actually stored under, which is what the adapter waits on Three tests come with it, each failing when its fix is reverted: a scope past one page must fold completely, deleting one key must not take its neighbours, and a tombstone alone must hide a key. --- .../tinymemory-remote/src/conformance_test.rs | 468 +++++++++++++++++- 1 file changed, 467 insertions(+), 1 deletion(-) diff --git a/crates/tinymemory-remote/src/conformance_test.rs b/crates/tinymemory-remote/src/conformance_test.rs index c1e7d2c9..f9bcb813 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,466 @@ 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. + 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) = 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" })), + ); + } + let replayed = log + .events + .iter() + .find(|e| e.pointer("/content/text").and_then(Value::as_str) == Some(payload.as_str())) + .and_then(|e| e.get("id").cloned()) + .unwrap_or(Value::Null); + return ( + axum::http::StatusCode::ACCEPTED, + Json(json!({ "event_id": replayed, "replayed_from_idempotency": true })), + ); + } + + log.idempotency.insert(key, payload.clone()); + log.next_offset += 2; + log.next_id += 1; + let offset = log.next_offset; + let id = format!("evt_{}", log.next_id); + 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 +} + +#[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()); +} From 597ee6a9026310e5fbaa438de18d1453af556c94 Mon Sep 17 00:00:00 2001 From: Ghost Scripter Date: Wed, 2 Sep 2026 19:10:41 +0530 Subject: [PATCH 3/5] Run the provider contract against a live CortexDB This target exists because a double cannot catch a service that behaves differently from its documentation, which is exactly what happened here: the offline suite passed throughout while the adapter had four defects that only the real engine exposed. Skipped without TINYMEMORY_TEST_CORTEX_URL and ..._KEY, so cargo test stays offline and deterministic by default. --- .../tests/live_remote_engines.rs | 22 ++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) 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(()) +} From fcc20029f44f6fc1fb2dd0ca823cf9a81dd04b35 Mon Sep 17 00:00:00 2001 From: Ghost Scripter Date: Wed, 2 Sep 2026 23:22:28 +0530 Subject: [PATCH 4/5] Keep the engine's ranking, and stop a settle probe failing a write MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three defects from review, each verified against the engine before fixing and each now covered by a test that fails without its fix. `search` folded the recall answer and then truncated it. The fold orders by key, which is right for a listing and wrong for a ranked answer: truncating an alphabetical order discards the engine's best hits and keeps whichever keys sort first. Hits are now restored to the order the engine returned them in, by each key's first appearance, before the cap. The conformance suite could not catch this — its recall fixture stores identical content under r1/r2/r3, where ranked and alphabetical order coincide. The settle probe propagated its own errors. Phase two of the write path is documented as best-effort, because a search index that has not caught up does not unmake a durable, keyed-readable write — but an error from the probe failed the whole store, which is exactly what the phase says it will not do. `urlencoding` escaped `:` and `/` only. That covers a scope, which carries nothing else, but not the cursor: it is opaque engine output, and a `+`, `&`, `=`, `#` or `?` in one would silently reshape the query string rather than fail. It now percent-encodes everything outside the URI unreserved set, by byte, so multi-byte UTF-8 stays recoverable. --- .../tinymemory-remote/src/conformance_test.rs | 99 +++++++++++++++++++ crates/tinymemory-remote/src/cortex.rs | 58 ++++++++--- crates/tinymemory-remote/src/cortex_test.rs | 16 +++ 3 files changed, 160 insertions(+), 13 deletions(-) diff --git a/crates/tinymemory-remote/src/conformance_test.rs b/crates/tinymemory-remote/src/conformance_test.rs index f9bcb813..64da6fc0 100644 --- a/crates/tinymemory-remote/src/conformance_test.rs +++ b/crates/tinymemory-remote/src/conformance_test.rs @@ -852,6 +852,65 @@ async fn cortex_backend() -> String { 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; @@ -1039,3 +1098,43 @@ async fn a_tombstone_alone_is_enough_to_hide_a_key() { .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" + ); +} diff --git a/crates/tinymemory-remote/src/cortex.rs b/crates/tinymemory-remote/src/cortex.rs index c6d5d132..004c6eaf 100644 --- a/crates/tinymemory-remote/src/cortex.rs +++ b/crates/tinymemory-remote/src/cortex.rs @@ -452,15 +452,23 @@ impl CortexDialect { 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 answer: Value = self + let probe = self .client - .json( + .json::( Method::POST, "v1/recall", Some(&json!({ "scope": scope, "query": query })), Attempts::RetryTransient, ) - .await?; + .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; } @@ -624,17 +632,23 @@ fn fresh_idempotency_key() -> String { format!("tm-{nanos}-{}", SEQ.fetch_add(1, Ordering::Relaxed)) } -/// Percent-encodes the characters a scope path carries that a query string -/// would otherwise read as structure. +/// 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 { - value - .chars() - .map(|c| match c { - ':' => "%3A".to_string(), - '/' => "%2F".to_string(), - other => other.to_string(), - }) - .collect() + 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] @@ -821,7 +835,25 @@ impl Dialect for CortexDialect { .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) } diff --git a/crates/tinymemory-remote/src/cortex_test.rs b/crates/tinymemory-remote/src/cortex_test.rs index a3445ed4..6a41ffbb 100644 --- a/crates/tinymemory-remote/src/cortex_test.rs +++ b/crates/tinymemory-remote/src/cortex_test.rs @@ -117,3 +117,19 @@ fn taint_survives_the_envelope() { 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"); +} From d15e9fca9f5be471354a164f0f3e8b79795e095c Mon Sep 17 00:00:00 2001 From: Ghost Scripter Date: Wed, 2 Sep 2026 23:53:09 +0530 Subject: [PATCH 5/5] Cap scope depth, salt the idempotency key, fix the double's replay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three more from review. The scope-depth limit was measured against a running engine rather than read off the grammar: 32 segments are accepted, 33 come back as 422 INVALID_BODY. scope_of now refuses the 33rd locally, which keeps the error specific instead of surfacing as a generic body rejection from the wire. fresh_idempotency_key combined a timestamp with a process-local counter. The counter separates writes within a 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. A third part, per-process entropy from RandomState, closes that. No new dependency: the standard library already seeds that per process, and a vendored crate should not grow a dependency for one string. The double resolved a replay by searching events for matching content, so two keys carrying identical text both replayed to the first event. That is not a hypothetical shape here — the adapter mints a fresh key per write, so a re-store of unchanged content is exactly it — and the write path waits on the id it is handed, which would have been the wrong record. The key now records the event it created. Also documents what the test-only expect_used allowance covers. --- .../tinymemory-remote/src/conformance_test.rs | 91 ++++++++++++++++--- crates/tinymemory-remote/src/cortex.rs | 53 +++++++++-- crates/tinymemory-remote/src/cortex_test.rs | 45 +++++++++ 3 files changed, 171 insertions(+), 18 deletions(-) diff --git a/crates/tinymemory-remote/src/conformance_test.rs b/crates/tinymemory-remote/src/conformance_test.rs index 64da6fc0..ebedace0 100644 --- a/crates/tinymemory-remote/src/conformance_test.rs +++ b/crates/tinymemory-remote/src/conformance_test.rs @@ -595,8 +595,11 @@ async fn the_cognee_double_actually_retains() { 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. - idempotency: BTreeMap, + /// `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, } @@ -619,7 +622,7 @@ async fn cortex_experience( .unwrap_or_default() .to_string(); - if let Some(seen) = log.idempotency.get(&key) { + if let Some((seen, event_id)) = log.idempotency.get(&key) { if seen != &payload { // The refusal the whole adapter is designed around. return ( @@ -627,23 +630,17 @@ async fn cortex_experience( Json(json!({ "error_code": "IDEMPOTENCY_CONFLICT" })), ); } - let replayed = log - .events - .iter() - .find(|e| e.pointer("/content/text").and_then(Value::as_str) == Some(payload.as_str())) - .and_then(|e| e.get("id").cloned()) - .unwrap_or(Value::Null); return ( axum::http::StatusCode::ACCEPTED, - Json(json!({ "event_id": replayed, "replayed_from_idempotency": true })), + Json(json!({ "event_id": event_id, "replayed_from_idempotency": true })), ); } - log.idempotency.insert(key, payload.clone()); 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) @@ -1138,3 +1135,75 @@ async fn recall_keeps_the_engine_ranking_when_it_truncates() { 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 index 004c6eaf..cd0218be 100644 --- a/crates/tinymemory-remote/src/cortex.rs +++ b/crates/tinymemory-remote/src/cortex.rs @@ -113,6 +113,12 @@ const SEGMENT_PLAIN: &str = "tm"; /// 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 @@ -300,6 +306,15 @@ impl CortexDialect { 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("/")) } @@ -615,21 +630,45 @@ impl CortexDialect { } } -/// A value no other write will reuse. +/// A fresh idempotency key for every write. /// -/// Not a content hash and not TinyMemory's key: the engine refuses a reused key -/// carrying a different body, so re-deriving this from the record is exactly -/// what breaks the second store. Monotonic clock plus a process-local counter -/// keeps it unique within a run and across runs without adding a dependency to -/// a vendored crate for one string. +/// 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-{nanos}-{}", SEQ.fetch_add(1, Ordering::Relaxed)) + format!( + "tm-{salt:016x}-{nanos}-{}", + SEQ.fetch_add(1, Ordering::Relaxed) + ) } /// Percent-encodes everything outside the URI unreserved set. diff --git a/crates/tinymemory-remote/src/cortex_test.rs b/crates/tinymemory-remote/src/cortex_test.rs index 6a41ffbb..397f7c2c 100644 --- a/crates/tinymemory-remote/src/cortex_test.rs +++ b/crates/tinymemory-remote/src/cortex_test.rs @@ -5,6 +5,9 @@ //! 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::*; @@ -133,3 +136,45 @@ fn a_cursor_is_escaped_far_beyond_the_characters_a_scope_carries() { // 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" + ); +}