diff --git a/README.md b/README.md index 01d569e1..a6a9c932 100644 --- a/README.md +++ b/README.md @@ -400,6 +400,8 @@ AI: β†’ memory_branch(name="eval_sqlite") Full API details: [API Reference Skill](skills/api-reference/SKILL.md) +Retry-safe single-memory writes: [Idempotent store protocol](docs/api/idempotent_store.md). + --- ## πŸ”§ Commands diff --git a/docs/api/idempotent_store.md b/docs/api/idempotent_store.md new file mode 100644 index 00000000..f0775ae4 --- /dev/null +++ b/docs/api/idempotent_store.md @@ -0,0 +1,62 @@ +# Idempotent single-memory writes + +`POST /v1/memories/idempotent` accepts the normal single-store body plus a required +`idempotency_key` containing 1–128 ASCII graphic bytes. It uses the same scoped +memory-write permission, owner-only deployment authentication, branch validation, +sensitivity checks and user database routing as ordinary memory writes. + +The authenticated owner and resolved storage branch scope the key. An omitted or +blank branch follows the current checkout; specifying that same branch explicitly +replays the same receipt. The operation freezes the routed SQL store and branch +table before embedding, so a concurrent checkout cannot retarget the insert or +conflict lookup. An invalid/deleted branch fails instead of silently moving a keyed +write to another branch. Clients needing a stable scope across separate retries +must specify it explicitly (for example, `"branch": "main"`). + +Clients cannot choose another +owner through JSON. Successful retries return the same persisted memory ID. +Concurrent requests use the memory primary key to arbitrate one insertion; +only the insertion winner reports a new memory and schedules entity extraction. +The idempotent path bypasses approximate vector deduplication so two different +session operations cannot acquire each other's record or session metadata. + +A key reused with a different content, type, session, trust tier, confidence, +observed time, author, subject or caller metadata fails. The backend preserves a +request fingerprint in reserved `_memoria_store_request_sha256` metadata; callers +cannot set or query that reserved field, and API memory responses hide it while +returning caller metadata unchanged. Payload fingerprints canonicalize nested +JSON object keys. Ordinary vector dedup excludes receipt-bearing rows, so an +ordinary store cannot supersede a receipt or replace its metadata. A retry never +reactivates an inactive receipt. The guarantee depends on retaining the record/tombstone; +physical database or record erasure ends that receipt's lifetime. Account/branch +retention and deletion policies remain authoritative. + +Protocol v1 encodes `["memoria.store.v1", owner, resolved_branch, key]` as compact +JSON, hashes it with SHA-256 and uses the first 16 bytes as a lowercase 32-character +hex memory ID. A caller can reject a successful response with an incompatible ID +before treating it as a write receipt. The identity is an operation key, not an +authorization credential. Normal single-store calls may also supply the optional +key; batch store rejects keys because it has no corresponding atomic batch +receipt contract. + +The dedicated path allows callers such as Astra's durable session-memory journal +to fail before writing when a deployment uses an old backend. They must retain +source state and retry/reconcile that same frozen payload. They must not fall back +to ordinary store on 404, 405, a malformed receipt or a lost response. + +Example (the token remains in the application's credential boundary): + +```json +{ + "content": "[@episode/summary] Session completed its task", + "memory_type": "episodic", + "session_id": "session-example", + "trust_tier": "T3", + "branch": "main", + "idempotency_key": "session-example:episode:revision-1" +} +``` + +Astra must deploy this backend endpoint before enabling its matched durable +finalization client. Repeated exit/close, process restart and partial-stage retry +can then reuse receipts without manufacturing duplicate episode or scene nodes. diff --git a/memoria/crates/memoria-api/src/auth.rs b/memoria/crates/memoria-api/src/auth.rs index 0b23a4c9..095aa166 100644 --- a/memoria/crates/memoria-api/src/auth.rs +++ b/memoria/crates/memoria-api/src/auth.rs @@ -353,6 +353,7 @@ pub async fn group_main_write_guard( let branch_aware_body_route = matches!( path, "/v1/memories" + | "/v1/memories/idempotent" | "/v1/memories/batch" | "/v1/memories/correct" | "/v1/memories/purge" diff --git a/memoria/crates/memoria-api/src/lib.rs b/memoria/crates/memoria-api/src/lib.rs index d4eabe69..9a96dc16 100644 --- a/memoria/crates/memoria-api/src/lib.rs +++ b/memoria/crates/memoria-api/src/lib.rs @@ -126,6 +126,7 @@ fn should_mark_metrics_dirty( if method == "POST" { return match path { "/v1/memories" + | "/v1/memories/idempotent" | "/v1/memories/batch" | "/v1/memories/correct" | "/v1/memories/purge" @@ -195,6 +196,10 @@ pub fn build_router(state: AppState) -> Router { // In group mode these are forbidden when checked-out on `main`. let write_routes = Router::new() .route("/v1/memories", post(routes::memory::store_memory)) + .route( + "/v1/memories/idempotent", + post(routes::memory::store_memory_idempotent), + ) .route("/v1/memories/batch", post(routes::memory::batch_store)) .route( "/v1/memories/correct", diff --git a/memoria/crates/memoria-api/src/models.rs b/memoria/crates/memoria-api/src/models.rs index 8e2ee200..38027631 100644 --- a/memoria/crates/memoria-api/src/models.rs +++ b/memoria/crates/memoria-api/src/models.rs @@ -10,6 +10,9 @@ use std::str::FromStr; #[derive(Deserialize)] pub struct StoreRequest { + /// Stable owner/branch-scoped write identity; retries return the same receipt. + #[serde(default)] + pub idempotency_key: Option, pub content: String, #[serde(default = "default_memory_type")] pub memory_type: String, @@ -457,7 +460,10 @@ pub struct MemoryResponse { } impl From for MemoryResponse { - fn from(m: Memory) -> Self { + fn from(mut m: Memory) -> Self { + if let Some(metadata) = m.extra_metadata.as_mut() { + metadata.remove(memoria_core::STORE_REQUEST_HASH_FIELD); + } Self { memory_id: m.memory_id, user_id: m.user_id, diff --git a/memoria/crates/memoria-api/src/routes/memory.rs b/memoria/crates/memoria-api/src/routes/memory.rs index 75bd7aaf..5c0a54f7 100644 --- a/memoria/crates/memoria-api/src/routes/memory.rs +++ b/memoria/crates/memoria-api/src/routes/memory.rs @@ -233,6 +233,22 @@ pub async fn fulltext_search_memories( Ok(Json(memories.into_iter().map(Into::into).collect())) } +/// Separate route rejects old servers before they can perform a non-idempotent +/// write. All normal authentication, owner scoping and validation still apply. +pub async fn store_memory_idempotent( + state: State, + auth: AuthUser, + request: Json, +) -> Result<(StatusCode, Json), (StatusCode, String)> { + if request.idempotency_key.is_none() { + return Err(( + StatusCode::UNPROCESSABLE_ENTITY, + "idempotency_key is required".into(), + )); + } + store_memory(state, auth, request).await +} + pub async fn store_memory( State(state): State, auth: AuthUser, @@ -267,7 +283,7 @@ pub async fn store_memory( }; let m = state .service - .store_memory_with_metadata_on_branch( + .store_memory_with_metadata_and_key_on_branch( auth.scope_id(), branch_param(req.branch.as_deref()), &req.content, @@ -279,6 +295,7 @@ pub async fn store_memory( author, req.subject_id, req.extra_metadata, + req.idempotency_key.as_deref(), ) .await .map_err(|e| { @@ -295,6 +312,16 @@ pub async fn batch_store( auth: AuthUser, Json(req): Json, ) -> Result<(StatusCode, Json>), (StatusCode, String)> { + if req + .memories + .iter() + .any(|memory| memory.idempotency_key.is_some()) + { + return Err(( + StatusCode::UNPROCESSABLE_ENTITY, + "idempotency_key requires the single-memory store endpoint".into(), + )); + } if req.memories.len() > 100 { return Err(( StatusCode::UNPROCESSABLE_ENTITY, diff --git a/memoria/crates/memoria-api/tests/idempotent_store.rs b/memoria/crates/memoria-api/tests/idempotent_store.rs new file mode 100644 index 00000000..1b175f3c --- /dev/null +++ b/memoria/crates/memoria-api/tests/idempotent_store.rs @@ -0,0 +1,439 @@ +//! Public authenticated idempotency protocol (the normal DB integration lane). +mod support; +use serde_json::{json, Value}; + +struct CheckoutEmbedder { + entered: tokio::sync::Semaphore, + resume: tokio::sync::Semaphore, +} + +#[async_trait::async_trait] +impl memoria_core::interfaces::EmbeddingProvider for CheckoutEmbedder { + async fn embed(&self, text: &str) -> Result, memoria_core::MemoriaError> { + if text == "checkout while embedding" { + self.entered.add_permits(1); + self.resume.acquire().await.unwrap().forget(); + } + memoria_core::interfaces::EmbeddingProvider::embed( + &memoria_embedding::MockEmbedder::new(4), + text, + ) + .await + } + + fn dimension(&self) -> usize { + 4 + } +} + +#[tokio::test] +async fn receipts_follow_resolved_branch_and_checkout_cannot_retarget_inflight_write() { + use std::{sync::Arc, time::Duration}; + let master = "branch-receipt-test-master"; + let embedder = Arc::new(CheckoutEmbedder { + entered: tokio::sync::Semaphore::new(0), + resume: tokio::sync::Semaphore::new(0), + }); + let server = support::multi_db::spawn_api_server( + "branch_receipt", + 4, + master.into(), + Some(embedder.clone()), + None, + None, + false, + ) + .await; + let owner = format!("branch_idem_{}", uuid::Uuid::new_v4().simple()); + let request = |path: &str| { + server + .client + .post(format!("{}{path}", server.base)) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .timeout(Duration::from_secs(30)) + }; + let response = request("/v1/branches") + .json(&json!({"name":"reviewb"})) + .send() + .await + .unwrap(); + let status = response.status(); + assert_eq!(status, 201, "{}", response.text().await.unwrap()); + assert_eq!( + request("/v1/branches/reviewb/checkout") + .send() + .await + .unwrap() + .status(), + 200 + ); + let body = json!({"content":"branch receipt", "memory_type":"episodic", "session_id":"branch-session", "trust_tier":"T3", "idempotency_key":"operation-a"}); + let response = request("/v1/memories/idempotent") + .json(&body) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 201); + let implicit: Value = response.json().await.unwrap(); + assert_eq!( + implicit["memory_id"], + memoria_service::service::idempotent_memory_id(&owner, Some("reviewb"), "operation-a") + .unwrap() + ); + let mut explicit_body = body.clone(); + // The HTTP boundary normalizes explicit branch spelling. + explicit_body["branch"] = json!(" reviewb "); + for endpoint in ["/v1/memories/idempotent", "/v1/memories"] { + let response = request(endpoint).json(&explicit_body).send().await.unwrap(); + assert_eq!(response.status(), 201); + let explicit: Value = response.json().await.unwrap(); + assert_eq!(explicit["memory_id"], implicit["memory_id"]); + } + let sql = server.user_store(&owner).await; + let table = sql.table_for_branch(&owner, Some("reviewb")).await.unwrap(); + let main = sql.table_for_branch(&owner, Some("main")).await.unwrap(); + assert_eq!( + sql.list_active_from(&table, &owner, 10) + .await + .unwrap() + .len(), + 1 + ); + assert!(sql + .list_active_from(&main, &owner, 10) + .await + .unwrap() + .is_empty()); + + let mut inflight_body = body.clone(); + inflight_body["content"] = json!("checkout while embedding"); + inflight_body["idempotency_key"] = json!("inflight-operation"); + let pending_request = request("/v1/memories/idempotent").json(&inflight_body); + let pending = tokio::spawn(async move { pending_request.send().await.unwrap() }); + tokio::time::timeout(Duration::from_secs(15), embedder.entered.acquire()) + .await + .unwrap() + .unwrap() + .forget(); + let checkout = request("/v1/branches/main/checkout").send().await.unwrap(); + // Release before checking to keep failed assertions from stranding the request. + embedder.resume.add_permits(1); + assert_eq!(checkout.status(), 200); + let response = pending.await.unwrap(); + assert_eq!(response.status(), 201); + let inflight: Value = response.json().await.unwrap(); + assert_eq!( + inflight["memory_id"], + memoria_service::service::idempotent_memory_id( + &owner, + Some("reviewb"), + "inflight-operation" + ) + .unwrap() + ); + assert_eq!( + sql.list_active_from(&table, &owner, 10) + .await + .unwrap() + .len(), + 2 + ); + assert!(sql + .list_active_from(&main, &owner, 10) + .await + .unwrap() + .is_empty()); + inflight_body["branch"] = json!("reviewb"); + let response = request("/v1/memories/idempotent") + .json(&inflight_body) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 201); + assert_eq!( + response.json::().await.unwrap()["memory_id"], + inflight["memory_id"] + ); + // The same key in main has a separate identity, even with identical payload. + let response = request("/v1/memories/idempotent") + .json(&body) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 201); + let main_receipt: Value = response.json().await.unwrap(); + assert_ne!(main_receipt["memory_id"], implicit["memory_id"]); + explicit_body["branch"] = json!("main"); + let response = request("/v1/memories/idempotent") + .json(&explicit_body) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 201); + assert_eq!( + response.json::().await.unwrap()["memory_id"], + main_receipt["memory_id"] + ); + assert_eq!( + sql.list_active_from(&main, &owner, 10).await.unwrap().len(), + 1 + ); + explicit_body["branch"] = json!("missing-branch"); + assert_eq!( + request("/v1/memories/idempotent") + .json(&explicit_body) + .send() + .await + .unwrap() + .status(), + 404 + ); +} + +#[tokio::test] +async fn owner_scoped_store_replays_real_receipts_and_rejects_missing_keys() { + let master = "idempotent-store-test-master"; + let server = support::multi_db::spawn_api_server( + "idempotent_store", + 4, + master.into(), + Some(std::sync::Arc::new(memoria_embedding::MockEmbedder::new(4))), + None, + None, + false, + ) + .await; + let owner = format!("idem_api_{}", uuid::Uuid::new_v4().simple()); + let url = format!("{}/v1/memories/idempotent", server.base); + let body = json!({"content":"[@episode/active] confirmed session", "memory_type":"episodic", "session_id":"session-a", "trust_tier":"T3", "idempotency_key":"same-operation", + "extra_metadata":{"scene":"qa", "agent":{"b":2,"a":1}}}); + let mut id = None; + for _ in 0..3 { + let response = server + .client + .post(&url) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .json(&body) + .send() + .await + .unwrap(); + let status = response.status(); + let response: Value = response.json().await.unwrap(); + assert_eq!(status, 201, "{response}"); + assert_eq!(response["user_id"], owner); + assert_eq!(response["session_id"], "session-a"); + assert_eq!(response["extra_metadata"], body["extra_metadata"]); + assert!(response["extra_metadata"] + .get(memoria_core::STORE_REQUEST_HASH_FIELD) + .is_none()); + if let Some(ref id) = id { + assert_eq!(&response["memory_id"], id); + } else { + id = Some(response["memory_id"].clone()); + } + } + let reordered: Value = serde_json::from_str(r#"{"content":"[@episode/active] confirmed session","memory_type":"episodic","session_id":"session-a","trust_tier":"T3","idempotency_key":"same-operation","extra_metadata":{"agent":{"a":1,"b":2},"scene":"qa"}}"#).unwrap(); + let response: Value = server + .client + .post(&url) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .json(&reordered) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(Some(response["memory_id"].clone()), id); + let get: Value = server + .client + .get(format!( + "{}/v1/memories/{}", + server.base, + response["memory_id"].as_str().unwrap() + )) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(get["extra_metadata"], body["extra_metadata"]); + // Ordinary vector dedup must leave exact-key receipts untouched, while + // remaining effective for ordinary rows. Same embedding/content exercises + // the metadata-refresh branch that previously overwrote the receipt hash. + let mut ordinary = body.clone(); + ordinary.as_object_mut().unwrap().remove("idempotency_key"); + ordinary["session_id"] = json!("ordinary-session"); + ordinary["extra_metadata"] = json!({"scene":"ordinary"}); + let ordinary_url = format!("{}/v1/memories", server.base); + let mut ordinary_id = None; + for _ in 0..2 { + let response: Value = server + .client + .post(&ordinary_url) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .json(&ordinary) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_ne!(Some(response["memory_id"].clone()), id); + assert_eq!(response["session_id"], "ordinary-session"); + if let Some(expected) = ordinary_id.as_ref() { + assert_eq!(&response["memory_id"], expected); + } + ordinary_id = Some(response["memory_id"].clone()); + } + let replay: Value = server + .client + .post(&url) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .json(&body) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(Some(replay["memory_id"].clone()), id); + assert_eq!(replay["extra_metadata"], body["extra_metadata"]); + let mut reserved = body.clone(); + reserved["extra_metadata"][memoria_core::STORE_REQUEST_HASH_FIELD] = json!("forged"); + for endpoint in [url.clone(), format!("{}/v1/memories", server.base)] { + let mut request = reserved.clone(); + if endpoint == ordinary_url { + request.as_object_mut().unwrap().remove("idempotency_key"); + } + let response = server + .client + .post(endpoint) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .json(&request) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 422); + } + let response = server + .client + .post(format!("{}/v1/memories/query", server.base)) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .json(&json!({"extra_metadata_filter":{(memoria_core::STORE_REQUEST_HASH_FIELD):"hidden"}})) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 422); + let mut changed = body.clone(); + changed["content"] = json!("different payload"); + let response = server + .client + .post(&url) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .json(&changed) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 422); + let mut missing = body.clone(); + missing.as_object_mut().unwrap().remove("idempotency_key"); + assert_eq!( + server + .client + .post(&url) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .json(&missing) + .send() + .await + .unwrap() + .status(), + 422 + ); + assert_eq!( + server + .client + .post(&url) + .header("X-User-Id", &owner) + .json(&body) + .send() + .await + .unwrap() + .status(), + 401 + ); + // Batch writes have no atomic replay protocol and must reject keys before + // inserting even the first row. + let response = server + .client + .post(format!("{}/v1/memories/batch", server.base)) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .json(&json!({"memories":[body.clone()]})) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 422); + let mut reserved_batch = reserved.clone(); + reserved_batch + .as_object_mut() + .unwrap() + .remove("idempotency_key"); + let response = server + .client + .post(format!("{}/v1/memories/batch", server.base)) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .json(&json!({"memories":[reserved_batch]})) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 422); + let mut no_metadata = body.clone(); + no_metadata + .as_object_mut() + .unwrap() + .remove("extra_metadata"); + no_metadata["idempotency_key"] = json!("no-metadata-operation"); + let response: Value = server + .client + .post(&url) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &owner) + .json(&no_metadata) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert!(response.get("extra_metadata").is_none()); + assert!(response["memory_id"].as_str().is_some()); + let other = format!("idem_other_{}", uuid::Uuid::new_v4().simple()); + let response: Value = server + .client + .post(&url) + .header("Authorization", format!("Memoria-Owner {master}")) + .header("X-User-Id", &other) + .json(&body) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(response["user_id"], other); + assert_ne!(response["memory_id"], id.unwrap()); +} diff --git a/memoria/crates/memoria-core/src/lib.rs b/memoria/crates/memoria-core/src/lib.rs index 26e0d2f8..86f98f3e 100644 --- a/memoria/crates/memoria-core/src/lib.rs +++ b/memoria/crates/memoria-core/src/lib.rs @@ -41,3 +41,6 @@ pub fn truncate_utf8(s: &str, max_bytes: usize) -> &str { } &s[..end] } + +/// Reserved persistence fingerprint for atomic store receipts; not caller metadata. +pub const STORE_REQUEST_HASH_FIELD: &str = "_memoria_store_request_sha256"; diff --git a/memoria/crates/memoria-service/src/service.rs b/memoria/crates/memoria-service/src/service.rs index eae62394..f36c6010 100644 --- a/memoria/crates/memoria-service/src/service.rs +++ b/memoria/crates/memoria-service/src/service.rs @@ -1312,6 +1312,42 @@ impl MemoryService { author_id: Option, subject_id: Option, extra_metadata: Option>, + ) -> Result { + self.store_memory_with_metadata_and_key_on_branch( + user_id, + branch, + content, + memory_type, + session_id, + trust_tier, + observed_at, + initial_confidence, + author_id, + subject_id, + extra_metadata, + None, + ) + .await + } + + /// Atomic, owner/branch-scoped retry receipts. The memory primary key is + /// deterministic; approximate/vector deduplication is deliberately bypassed. + #[allow(clippy::too_many_arguments)] + #[tracing::instrument(skip_all)] + pub async fn store_memory_with_metadata_and_key_on_branch( + &self, + user_id: &str, + branch: Option<&str>, + content: &str, + memory_type: MemoryType, + session_id: Option, + trust_tier: Option, + observed_at: Option>, + initial_confidence: Option, + author_id: Option, + subject_id: Option, + mut extra_metadata: Option>, + idempotency_key: Option<&str>, ) -> Result { let subject_id = normalize_opt_string(subject_id); if let Some(ref sid) = subject_id { @@ -1326,6 +1362,52 @@ impl MemoryService { "content exceeds 32 KiB limit".into(), )); } + reject_reserved_metadata(extra_metadata.as_ref())?; + let idempotency = if let Some(key) = idempotency_key { + if self.sql_store.is_none() { + return Err(MemoriaError::Validation( + "idempotent writes require SQL storage".into(), + )); + } + use sha2::{Digest, Sha256}; + let mut payload = serde_json::json!([ + content, + memory_type.to_string(), + session_id, + trust_tier + .as_ref() + .unwrap_or(&TrustTier::T1Verified) + .to_string(), + observed_at, + initial_confidence, + author_id, + subject_id, + extra_metadata + ]); + // Canonicalize recursively even if a consumer enables preserve_order. + payload.sort_all_objects(); + let hash = format!("{:x}", Sha256::digest(serde_json::to_vec(&payload)?)); + let sql = self.user_sql_store(user_id).await?; + // Snapshot checkout once. Explicit resolution uses that same branch, + // never another read of the mutable active-branch pointer after embed. + let resolved_branch = match branch.map(str::trim).filter(|branch| !branch.is_empty()) { + Some(branch) => branch.to_owned(), + None => sql.active_branch_name(user_id).await?, + }; + let id = idempotent_memory_id(user_id, Some(&resolved_branch), key)?; + let table = sql + .table_for_branch(user_id, Some(&resolved_branch)) + .await?; + if let Some(existing) = sql.get_from(&table, &id).await? { + return validate_idempotent_receipt(existing, user_id, &hash); + } + extra_metadata + .get_or_insert_with(HashMap::new) + .insert(STORE_REQUEST_HASH_FIELD.into(), serde_json::json!(hash)); + Some((id, hash, sql, table)) + } else { + None + }; let t0 = std::time::Instant::now(); // Sensitivity check β€” block HIGH tier, redact MEDIUM tier let sensitivity = check_sensitivity(content); @@ -1344,7 +1426,10 @@ impl MemoryService { let embedding = self.embed(content).await?; let t_embed = t0.elapsed(); let memory = Memory { - memory_id: Uuid::now_v7().simple().to_string(), + memory_id: idempotency + .as_ref() + .map(|(id, ..)| id.clone()) + .unwrap_or_else(|| Uuid::now_v7().simple().to_string()), user_id: user_id.to_string(), author_id, subject_id, @@ -1365,6 +1450,37 @@ impl MemoryService { trust_tier: effective_tier, retrieval_score: None, }; + if let Some((id, hash, sql, table)) = idempotency { + // INSERT's primary key arbitrates concurrent instances and lost + // acknowledgments. Never resurrect an inactive/deleted receipt. + match sql.insert_into(&table, &memory).await { + Ok(()) => { + let payload = serde_json::json!({"content": &memory.content, "type": memory.memory_type.to_string()}).to_string(); + self.send_edit_log( + user_id, + "inject", + Some(&id), + Some(&payload), + "store_memory:idempotent", + None, + ); + self.report(StatsEvent::MemoryStored { + user_id: user_id.into(), + memory_type: memory.memory_type.to_string(), + trust_tier: memory.trust_tier.to_string(), + }); + self.enqueue_entity_extraction(user_id, &id, &memory.content) + .await; + return Ok(memory); + } + Err(error) => { + if let Some(existing) = sql.get_from(&table, &id).await? { + return validate_idempotent_receipt(existing, user_id, &hash); + } + return Err(error); + } + } + } // Dedup: if embedding exists, check for near-duplicate and supersede // TODO(concurrency): race between dedup check and insert can create duplicates if self.sql_store.is_some() { @@ -1493,8 +1609,12 @@ impl MemoryService { memory_type: memory.memory_type.to_string(), trust_tier: memory.trust_tier.to_string(), }); - self.enqueue_entity_extraction(user_id, &memory.memory_id, &memory.content) - .await; + self.enqueue_entity_extraction( + user_id, + &memory.memory_id, + &memory.content, + ) + .await; return Ok(memory); } } @@ -2834,6 +2954,7 @@ impl MemoryService { let mut contents = Vec::with_capacity(items.len()); let mut checked_items = Vec::with_capacity(items.len()); for (content, mt, session_id, tier, subject_id, extra_metadata) in items { + reject_reserved_metadata(extra_metadata.as_ref())?; let subject_id = normalize_opt_string(subject_id); if let Some(ref sid) = subject_id { if sid.len() > 128 { @@ -3501,3 +3622,83 @@ impl MemoryService { sql.get_feedback_by_tier(user_id).await } } + +use memoria_core::STORE_REQUEST_HASH_FIELD; + +fn reject_reserved_metadata( + metadata: Option<&HashMap>, +) -> Result<(), MemoriaError> { + if metadata.is_some_and(|metadata| metadata.contains_key(STORE_REQUEST_HASH_FIELD)) { + return Err(MemoriaError::Validation( + "reserved idempotency metadata field".into(), + )); + } + Ok(()) +} + +/// Protocol v1 uses the first 128 SHA-256 bits as a 32-character memory ID. +/// The backend derives the owner from authentication, never from request JSON. +/// Callers must supply the resolved branch; `None` is an explicit main shorthand, +/// not a request to follow a mutable checkout pointer. +pub fn idempotent_memory_id( + user_id: &str, + branch: Option<&str>, + key: &str, +) -> Result { + if key.is_empty() || key.len() > 128 || !key.bytes().all(|b| b.is_ascii_graphic()) { + return Err(MemoriaError::Validation( + "idempotency_key must contain 1..128 ASCII graphic bytes".into(), + )); + } + use sha2::{Digest, Sha256}; + let identity = serde_json::json!(["memoria.store.v1", user_id, branch.unwrap_or("main"), key]); + let digest = Sha256::digest(serde_json::to_vec(&identity)?); + let mut receipt = [0_u8; 16]; + receipt.copy_from_slice(&digest[..16]); + Ok(format!("{:032x}", u128::from_be_bytes(receipt))) +} + +fn validate_idempotent_receipt( + memory: Memory, + owner: &str, + request_hash: &str, +) -> Result { + if memory.user_id != owner + || !memory.is_active + || memory + .extra_metadata + .as_ref() + .and_then(|metadata| metadata.get(STORE_REQUEST_HASH_FIELD)) + .and_then(serde_json::Value::as_str) + != Some(request_hash) + { + return Err(MemoriaError::Validation( + "idempotency_key was already used for another payload or an inactive receipt".into(), + )); + } + Ok(memory) +} + +#[cfg(test)] +mod idempotency_tests { + use super::idempotent_memory_id; + #[test] + fn stable_receipts_are_owner_branch_scoped_and_keys_are_exact() { + assert_eq!( + idempotent_memory_id("owner-a", None, "operation-a").unwrap(), + "2015485c7a5b7d3edaf5f9e3b0365b69" + ); + assert_ne!( + idempotent_memory_id("owner-a", None, "operation-a").unwrap(), + idempotent_memory_id("owner-b", None, "operation-a").unwrap() + ); + assert_ne!( + idempotent_memory_id("owner-a", None, "operation-a").unwrap(), + idempotent_memory_id("owner-a", Some("branch-b"), "operation-a").unwrap() + ); + for key in ["", " padded", "line\n", "non-ascii-δΈ­ζ–‡"] { + assert!(idempotent_memory_id("owner-a", None, key).is_err()); + } + assert!(idempotent_memory_id("owner-a", None, &"x".repeat(129)).is_err()); + } +} diff --git a/memoria/crates/memoria-service/tests/idempotent_store_db_it.rs b/memoria/crates/memoria-service/tests/idempotent_store_db_it.rs new file mode 100644 index 00000000..49d56e7b --- /dev/null +++ b/memoria/crates/memoria-service/tests/idempotent_store_db_it.rs @@ -0,0 +1,141 @@ +//! Atomic write receipts against isolated MatrixOne (the CI integration lane). +use memoria_core::{MemoryType, TrustTier}; +use memoria_service::MemoryService; +use memoria_storage::SqlMemoryStore; +use std::sync::Arc; + +async fn store( + service: &MemoryService, + owner: &str, + session: &str, + key: &str, + content: &str, +) -> Result { + service + .store_memory_with_metadata_and_key_on_branch( + owner, + None, + content, + MemoryType::Episodic, + Some(session.into()), + Some(TrustTier::T3Inferred), + None, + None, + None, + None, + None, + Some(key), + ) + .await +} + +#[tokio::test] +async fn concurrent_replays_have_one_receipt_conflicts_do_not_mutate_and_deletion_is_final() { + let url = + std::env::var("DATABASE_URL").expect("designate isolated MatrixOne with DATABASE_URL"); + let sql = Arc::new( + SqlMemoryStore::connect(&url, 4, uuid::Uuid::new_v4().to_string()) + .await + .unwrap(), + ); + sql.migrate().await.unwrap(); + let service = Arc::new(MemoryService::new_sql_with_llm(sql.clone(), None, None).await); + let owner = format!("idem-{}", uuid::Uuid::new_v4().simple()); + let mut tasks = tokio::task::JoinSet::new(); + for _ in 0..8 { + let service = service.clone(); + let owner = owner.clone(); + tasks.spawn(async move { + store( + &service, + &owner, + "session-a", + "operation-a", + "confirmed episode", + ) + .await + .unwrap() + }); + } + let mut receipt = None; + while let Some(result) = tasks.join_next().await { + let memory = result.unwrap(); + assert_eq!(memory.user_id, owner); + assert_eq!(memory.session_id.as_deref(), Some("session-a")); + if let Some(ref id) = receipt { + assert_eq!(&memory.memory_id, id); + } else { + receipt = Some(memory.memory_id); + } + } + let id = receipt.unwrap(); + assert_eq!( + id, + memoria_service::service::idempotent_memory_id(&owner, None, "operation-a").unwrap() + ); + let replay = store( + &service, + &owner, + "session-a", + "operation-a", + "confirmed episode", + ) + .await + .unwrap(); + assert_eq!(replay.memory_id, id); + assert!(store( + &service, + &owner, + "session-a", + "operation-a", + "changed payload" + ) + .await + .is_err()); + assert!(store( + &service, + &owner, + "session-b", + "operation-a", + "confirmed episode" + ) + .await + .is_err()); + let table = sql.table_for_branch(&owner, None).await.unwrap(); + assert_eq!( + sql.get_from(&table, &id).await.unwrap().unwrap().content, + "confirmed episode" + ); + let different_session = store( + &service, + &owner, + "session-b", + "operation-b", + "confirmed episode", + ) + .await + .unwrap(); + assert_ne!(different_session.memory_id, id); + let other = store( + &service, + "another-owner", + "session-a", + "operation-a", + "confirmed episode", + ) + .await + .unwrap(); + assert_ne!(other.memory_id, id); + assert_eq!(other.user_id, "another-owner"); + sql.soft_delete_from(&table, &id).await.unwrap(); + assert!(store( + &service, + &owner, + "session-a", + "operation-a", + "confirmed episode" + ) + .await + .is_err()); + assert!(sql.get_from(&table, &id).await.unwrap().is_none()); +} diff --git a/memoria/crates/memoria-storage/src/store.rs b/memoria/crates/memoria-storage/src/store.rs index 82666569..1c1dd6f3 100644 --- a/memoria/crates/memoria-storage/src/store.rs +++ b/memoria/crates/memoria-storage/src/store.rs @@ -37,6 +37,11 @@ pub fn validate_extra_metadata_filter( ))); } for (key, value) in filter { + if key == memoria_core::STORE_REQUEST_HASH_FIELD { + return Err(MemoriaError::Validation( + "reserved receipt metadata is not queryable".into(), + )); + } let mut chars = key.chars(); let valid_first = chars .next() @@ -5229,10 +5234,12 @@ impl SqlMemoryStore { FROM {table} \ WHERE user_id = ? AND is_active = 1 \ AND memory_type = ? \ + AND JSON_EXTRACT(extra_metadata, '$.{receipt_hash}') IS NULL \ AND embedding IS NOT NULL AND vector_dims(embedding) > 0 \ AND memory_id != ? \ {subject_clause} \ - ORDER BY l2_dist ASC LIMIT 1 by rank with option 'mode=post'" + ORDER BY l2_dist ASC LIMIT 1 by rank with option 'mode=post'", + receipt_hash = memoria_core::STORE_REQUEST_HASH_FIELD, ); let rows = sqlx::query(&sql) .bind(user_id)