From 86528705150649e7e2608d282ade9b598baafdff Mon Sep 17 00:00:00 2001 From: Shanu Date: Fri, 21 Aug 2026 13:08:33 +0530 Subject: [PATCH] embedded: the mandatory path stops blocking the executor, and KV reads spell keys the way writes do (#75) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every Memory-trait method on UnifiedMemory that ran synchronous rusqlite work inline now hops to spawn_blocking, the same idiom the adapter's profile and episodic families always used: bodies extracted into synchronous associated fns, parameters owned before the move, join failures contexted. recall's two synchronous fts5 fetches hop the same way with join faults folded into their existing non-fatal arms; forget is split — only its lookup is synchronous, delete_document awaits internally and stays put. The adapter's offloading claims stop being aspirational, and the two Cargo.toml comments are narrowed to the families that actually hop (the KV/graph accessors go through the client shim inline). TinycortexProvider::kv_get and kv_list now canonicalize the caller key and prefix through the same canonical_identifier the write-path shim applies, restoring put→get and put→list symmetry — kv_delete always went through the shim, which is exactly the incoherent split that let a rewritten key be stored, deleted, but never read. Prefix matching is documented as over canonical stored keys. The conformance suite grows assert_kv_round_trip: put → get → list-by-prefix → delete → gone, with one leg keyed by a formatted national-ID shape the canonicalizer provably rewrites (guarded by assert_ne! so predicate drift fails loudly, not silently) — the KV family previously had zero conformance coverage, which is how the asymmetry stayed invisible. Adapter-level tests pin the same symmetry under a rewritten namespace as well. Closes the embedded batch of #75. --- crates/tinymemory-conformance/src/lib.rs | 8 +- .../tinymemory-conformance/src/suite/mod.rs | 93 ++++++ .../tinymemory-core/src/store/memory_trait.rs | 303 ++++++++++++------ crates/tinymemory-tinycortex/Cargo.toml | 10 +- .../tinymemory-tinycortex/src/engine/mod.rs | 17 +- .../tests/full_provider_conformance.rs | 103 ++++++ 6 files changed, 422 insertions(+), 112 deletions(-) diff --git a/crates/tinymemory-conformance/src/lib.rs b/crates/tinymemory-conformance/src/lib.rs index 3e99c02..727fe4d 100644 --- a/crates/tinymemory-conformance/src/lib.rs +++ b/crates/tinymemory-conformance/src/lib.rs @@ -44,10 +44,10 @@ pub mod suite; pub use reference::{InMemoryProvider, REFERENCE_DRIVER_ID}; pub use suite::{ assert_awkward_content_round_trips, assert_capability_audit, assert_export_cursor_terminates, - assert_export_import_round_trip, assert_forget_is_idempotent, assert_list_filters_narrow, - assert_namespaces_are_isolated, assert_provider, assert_recall_respects_limit_and_namespace, - assert_store_get_round_trip, assert_taint_is_preserved, - assert_upsert_replaces_rather_than_duplicates, + assert_export_import_round_trip, assert_forget_is_idempotent, assert_kv_round_trip, + assert_list_filters_narrow, assert_namespaces_are_isolated, assert_provider, + assert_recall_respects_limit_and_namespace, assert_store_get_round_trip, + assert_taint_is_preserved, assert_upsert_replaces_rather_than_duplicates, }; pub use suite::{ // Exported alongside the assertions because a caller standing up its own diff --git a/crates/tinymemory-conformance/src/suite/mod.rs b/crates/tinymemory-conformance/src/suite/mod.rs index ff27191..f6d3fa6 100644 --- a/crates/tinymemory-conformance/src/suite/mod.rs +++ b/crates/tinymemory-conformance/src/suite/mod.rs @@ -63,6 +63,7 @@ pub async fn assert_provider(provider: Arc) { assert_recall_respects_limit_and_namespace(p).await; assert_export_import_round_trip(p).await; assert_awkward_content_round_trips(p).await; + assert_kv_round_trip(p).await; } /// Whether this driver reads back what it stores. @@ -656,6 +657,98 @@ pub async fn assert_awkward_content_round_trips(provider: &dyn MemoryProvider) { cleanup(provider, &ns, &keys).await; } +/// The key/value family round-trips: put → get → list-by-prefix → delete. +/// +/// As wired in [`assert_provider`], also skipped for a driver that fails the +/// `retains_writes` probe (the early return precedes every assert): a +/// non-retaining driver would fail the put→get leg for retention reasons, +/// which is not the asymmetry this case exists to catch. +/// +/// Skipped when the driver does not serve the optional `Graph` family — the +/// `as_graph()` accessor is the negotiated surface, and +/// [`assert_capability_audit`] already pins that it agrees with the advertised +/// [`Capability`] set, so probing the accessor *is* the capability check. +/// +/// One leg uses a formatted national-ID key on purpose. A driver may +/// canonicalize identifiers on write (PII scrubbing rewrites the stored key), +/// and the contract this asserts is *symmetry*: whatever transform the write +/// path applies, every read path must apply too. A driver whose `kv_get` / +/// `kv_list` compare the raw caller key misses every rewritten key — put→get +/// answers `None` forever — while its canonicalizing `kv_delete` still +/// reports `true`, which is exactly the asymmetry that stays invisible +/// without this case. The read-back key is deliberately *not* asserted to +/// equal the caller's: it is the stored (possibly canonical) form. +/// +/// # Panics +/// +/// Panics when a just-put key cannot be read, listed under its own prefix, or +/// deleted, or when it is still readable after a `true` delete. +pub async fn assert_kv_round_trip(provider: &dyn MemoryProvider) { + let who = provider.driver_id(); + let Some(graph) = provider.as_graph() else { + return; + }; + let ns = ns(provider, "kv"); + // A plain key, and one shaped like a formatted national ID — the shape + // identifier-canonicalizing drivers rewrite on write. (Not an email and + // not a bare digit run: both are deliberately outside the strict PII + // boundary gates such drivers use, so neither would exercise a rewrite.) + // The value carries no PII or secret shapes — drivers may scrub *content* + // more aggressively than identifiers, and this case asserts key symmetry, + // not content fidelity (that is `assert_awkward_content_round_trips`'s + // job for the document tier). + for (marker, key) in [("plain", "plain-key"), ("rewritten", "ssn-123-45-6789")] { + let value = serde_json::json!({ "leg": marker }); + graph + .kv_put(Some(&ns), key, value.clone()) + .await + .unwrap_or_else(|e| panic!("{who}: kv_put of `{key}` failed: {e}")); + + let got = graph + .kv_get(Some(&ns), key) + .await + .unwrap_or_else(|e| panic!("{who}: kv_get of `{key}` failed: {e}")) + .unwrap_or_else(|| { + panic!( + "{who}: kv_get did not find `{key}` right after kv_put — the read \ + path does not apply the write path's key transform" + ) + }); + assert_eq!( + got.value, value, + "{who}: kv_get of `{key}` surfaced another record's value" + ); + + // The caller's key must work as a prefix of its own record: prefix + // matching is over stored keys, so a driver that rewrites the key on + // write has to rewrite the prefix on read the same way. + let listed = graph + .kv_list(Some(&ns), Some(key), 16) + .await + .unwrap_or_else(|e| panic!("{who}: kv_list under `{key}` failed: {e}")); + assert!( + listed.iter().any(|record| record.value == value), + "{who}: kv_list under the `{key}` prefix did not surface the record" + ); + + assert!( + graph + .kv_delete(Some(&ns), key) + .await + .unwrap_or_else(|e| panic!("{who}: kv_delete of `{key}` failed: {e}")), + "{who}: kv_delete did not find `{key}`" + ); + let gone = graph + .kv_get(Some(&ns), key) + .await + .unwrap_or_else(|e| panic!("{who}: kv_get after delete failed: {e}")); + assert!( + gone.is_none(), + "{who}: `{key}` is still readable after kv_delete reported true" + ); + } +} + /// A namespace unique to this driver and assertion. /// /// Prefixed so the suite can run against a live service holding real data diff --git a/crates/tinymemory-core/src/store/memory_trait.rs b/crates/tinymemory-core/src/store/memory_trait.rs index 0ad0d60..c76da77 100644 --- a/crates/tinymemory-core/src/store/memory_trait.rs +++ b/crates/tinymemory-core/src/store/memory_trait.rs @@ -9,9 +9,12 @@ //! the implementation falls back to `GLOBAL_NAMESPACE` (legacy behavior), which //! Phase B/C will tighten once the memory tools pass namespace explicitly. +use std::sync::Arc; + use async_trait::async_trait; use chrono::{TimeZone, Utc}; -use rusqlite::{params, OptionalExtension}; +use parking_lot::Mutex; +use rusqlite::{params, Connection, OptionalExtension}; use serde_json::json; use crate::store::namespace_store::fts5; @@ -134,7 +137,18 @@ impl UnifiedMemory { } if let Some(sid) = opts.session_id { - let episodic_entries = match fts5::episodic_session_entries(&self.conn, sid) { + // Synchronous SQL behind the connection mutex — run it on the + // blocking pool rather than an executor thread. A join failure is + // folded into the same non-fatal arm as a query failure below. + let fetched = { + let conn = Arc::clone(&self.conn); + let session = sid.to_owned(); + tokio::task::spawn_blocking(move || fts5::episodic_session_entries(&conn, &session)) + .await + .context("join episodic session entries") + .and_then(|entries| entries) + }; + let episodic_entries = match fetched { Ok(entries) => { tracing::debug!( "[memory-trait] loaded {} episodic entries for session={sid}", @@ -192,9 +206,19 @@ impl UnifiedMemory { // already came in via the same-session path above. if opts.cross_session { let exclude = opts.session_id; - let cross_entries = match fts5::episodic_cross_session_search( - &self.conn, query, limit, exclude, - ) { + // Same blocking-pool hop as the same-session fetch above. + let fetched = { + let conn = Arc::clone(&self.conn); + let query = query.to_owned(); + let exclude = exclude.map(str::to_owned); + tokio::task::spawn_blocking(move || { + fts5::episodic_cross_session_search(&conn, &query, limit, exclude.as_deref()) + }) + .await + .context("join cross-session episodic search") + .and_then(|entries| entries) + }; + let cross_entries = match fetched { Ok(entries) => { tracing::debug!( "[memory-trait] cross-session episodic recall returned {} entries (exclude={:?})", @@ -266,6 +290,150 @@ impl UnifiedMemory { } } +// ── Blocking SQL bodies ────────────────────────────────────────────────────── +// +// The connection is a `parking_lot::Mutex`: every SQL +// call is synchronous and holds the lock for its duration, so running one on +// an executor thread stalls every task scheduled there. Each `Memory` method +// below owns its parameters, hops to `spawn_blocking`, and runs its body here; +// the bodies are associated fns (not `&self` methods) because the closure must +// be `'static` and cannot borrow the store. + +/// One `memory_docs` row as `get` selects it: +/// `(document_id, key, content, updated_at, category, taint, session_id)`. +type MemoryDocRow = (String, String, String, f64, String, String, Option); + +impl UnifiedMemory { + fn get_blocking( + conn: &Arc>, + ns: &str, + key: &str, + ) -> anyhow::Result> { + let conn = conn.lock(); + // `session_id` is selected here for the same reason `list` selects it: + // it is a column on this row, and a `get` that dropped it made the two + // readers disagree about one record. The contract's round-trip + // assertion catches exactly that (`tinymemory_conformance`), and it was + // invisible until #18 §A3 let this store be bound as a driver at all. + let row: Option = conn + .query_row( + "SELECT document_id, key, content, updated_at, category, taint, session_id + FROM memory_docs WHERE namespace = ?1 AND key = ?2 LIMIT 1", + params![ns, key], + |row| { + Ok(( + row.get(0)?, + row.get(1)?, + row.get(2)?, + row.get(3)?, + row.get(4)?, + row.get(5)?, + row.get(6)?, + )) + }, + ) + .optional()?; + Ok(row.map( + |(id, key, content, updated_at, category, taint_str, session_id)| MemoryEntry { + id, + key, + content, + namespace: Some(ns.to_string()), + category: memory_category_from_stored(&category), + timestamp: timestamp_to_rfc3339(updated_at), + session_id, + score: None, + taint: crate::MemoryTaint::from_db_str(&taint_str), + }, + )) + } + + fn list_blocking( + conn: &Arc>, + ns: &str, + category: Option<&MemoryCategory>, + session_id: Option<&str>, + ) -> anyhow::Result> { + let conn = conn.lock(); + let mut stmt = conn.prepare( + "SELECT document_id, key, content, category, session_id, updated_at, taint + FROM memory_docs WHERE namespace = ?1 ORDER BY updated_at DESC", + )?; + let rows = stmt.query_map(params![ns], |row| { + let stored_category: String = row.get(3)?; + Ok(MemoryEntry { + id: row.get(0)?, + key: row.get(1)?, + content: row.get(2)?, + namespace: Some(ns.to_string()), + category: memory_category_from_stored(&stored_category), + session_id: row.get(4)?, + timestamp: timestamp_to_rfc3339(row.get(5)?), + score: None, + taint: crate::MemoryTaint::from_db_str(&row.get::<_, String>(6)?), + }) + })?; + let mut entries = rows.collect::>>()?; + if let Some(category) = category { + entries.retain(|entry| &entry.category == category); + } + if let Some(session_id) = session_id { + entries.retain(|entry| entry.session_id.as_deref() == Some(session_id)); + } + Ok(entries) + } + + fn forget_lookup_blocking( + conn: &Arc>, + ns: &str, + key: &str, + ) -> anyhow::Result> { + let conn = conn.lock(); + Ok(conn + .query_row( + "SELECT document_id FROM memory_docs WHERE namespace = ?1 AND key = ?2 LIMIT 1", + params![ns, key], + |row| row.get(0), + ) + .optional()?) + } + + fn namespace_summaries_blocking( + conn: &Arc>, + ) -> anyhow::Result> { + let conn = conn.lock(); + let mut stmt = conn.prepare( + "SELECT namespace, COUNT(*) AS n, MAX(updated_at) AS last + FROM memory_docs + GROUP BY namespace + ORDER BY namespace", + )?; + let rows = stmt.query_map([], |row| { + let ns: String = row.get(0)?; + let count: i64 = row.get(1)?; + let last: Option = row.get(2)?; + Ok((ns, count, last)) + })?; + let mut out = Vec::new(); + for r in rows { + let (ns, count, last) = r?; + out.push(NamespaceSummary { + namespace: ns, + count: usize::try_from(count).unwrap_or(0), + last_updated: last.map(timestamp_to_rfc3339), + }); + } + Ok(out) + } + + fn count_blocking(conn: &Arc>) -> anyhow::Result { + let conn = conn.lock(); + let count: i64 = + conn.query_row("SELECT COUNT(*) FROM memory_docs", [], |row| row.get(0))?; + usize::try_from(count).context("negative count") + } +} + #[async_trait] impl Memory for UnifiedMemory { fn name(&self) -> &str { @@ -369,43 +537,10 @@ impl Memory for UnifiedMemory { // it again, which is the retry loop behind #5164. let ns = UnifiedMemory::sanitize_namespace(namespace); let key = crate::store::safety::canonical_document_key(key); - let conn = self.conn.lock(); - // `session_id` is selected here for the same reason `list` selects it: - // it is a column on this row, and a `get` that dropped it made the two - // readers disagree about one record. The contract's round-trip - // assertion catches exactly that (`tinymemory_conformance`), and it was - // invisible until #18 §A3 let this store be bound as a driver at all. - let row: Option<(String, String, String, f64, String, String, Option)> = conn - .query_row( - "SELECT document_id, key, content, updated_at, category, taint, session_id - FROM memory_docs WHERE namespace = ?1 AND key = ?2 LIMIT 1", - params![ns, key], - |row| { - Ok(( - row.get(0)?, - row.get(1)?, - row.get(2)?, - row.get(3)?, - row.get(4)?, - row.get(5)?, - row.get(6)?, - )) - }, - ) - .optional()?; - Ok(row.map( - |(id, key, content, updated_at, category, taint_str, session_id)| MemoryEntry { - id, - key, - content, - namespace: Some(ns.clone()), - category: memory_category_from_stored(&category), - timestamp: timestamp_to_rfc3339(updated_at), - session_id, - score: None, - taint: crate::MemoryTaint::from_db_str(&taint_str), - }, - )) + let conn = Arc::clone(&self.conn); + tokio::task::spawn_blocking(move || Self::get_blocking(&conn, &ns, &key)) + .await + .context("join Memory::get")? } async fn list( @@ -415,33 +550,14 @@ impl Memory for UnifiedMemory { session_id: Option<&str>, ) -> anyhow::Result> { let ns = UnifiedMemory::sanitize_namespace(normalize_namespace(namespace)); - let conn = self.conn.lock(); - let mut stmt = conn.prepare( - "SELECT document_id, key, content, category, session_id, updated_at, taint - FROM memory_docs WHERE namespace = ?1 ORDER BY updated_at DESC", - )?; - let rows = stmt.query_map(params![ns], |row| { - let stored_category: String = row.get(3)?; - Ok(MemoryEntry { - id: row.get(0)?, - key: row.get(1)?, - content: row.get(2)?, - namespace: Some(ns.clone()), - category: memory_category_from_stored(&stored_category), - session_id: row.get(4)?, - timestamp: timestamp_to_rfc3339(row.get(5)?), - score: None, - taint: crate::MemoryTaint::from_db_str(&row.get::<_, String>(6)?), - }) - })?; - let mut entries = rows.collect::>>()?; - if let Some(category) = category { - entries.retain(|entry| &entry.category == category); - } - if let Some(session_id) = session_id { - entries.retain(|entry| entry.session_id.as_deref() == Some(session_id)); - } - Ok(entries) + let category = category.cloned(); + let session_id = session_id.map(str::to_owned); + let conn = Arc::clone(&self.conn); + tokio::task::spawn_blocking(move || { + Self::list_blocking(&conn, &ns, category.as_ref(), session_id.as_deref()) + }) + .await + .context("join Memory::list")? } async fn forget(&self, namespace: &str, key: &str) -> anyhow::Result { @@ -451,17 +567,17 @@ impl Memory for UnifiedMemory { let ns = UnifiedMemory::sanitize_namespace(namespace); let key = crate::store::safety::canonical_document_key(key); let row: Option = { - let conn = self.conn.lock(); - conn.query_row( - "SELECT document_id FROM memory_docs WHERE namespace = ?1 AND key = ?2 LIMIT 1", - params![ns, key], - |row| row.get(0), - ) - .optional()? + let conn = Arc::clone(&self.conn); + let ns = ns.clone(); + tokio::task::spawn_blocking(move || Self::forget_lookup_blocking(&conn, &ns, &key)) + .await + .context("join Memory::forget")?? }; let Some(document_id) = row else { return Ok(false); }; + // `delete_document` awaits internally (graph upkeep, sidecar removal), + // so only the synchronous lookup above runs on the blocking pool. self.delete_document(&ns, &document_id) .await .map_err(anyhow::Error::msg)?; @@ -469,36 +585,17 @@ impl Memory for UnifiedMemory { } async fn namespace_summaries(&self) -> anyhow::Result> { - let conn = self.conn.lock(); - let mut stmt = conn.prepare( - "SELECT namespace, COUNT(*) AS n, MAX(updated_at) AS last - FROM memory_docs - GROUP BY namespace - ORDER BY namespace", - )?; - let rows = stmt.query_map([], |row| { - let ns: String = row.get(0)?; - let count: i64 = row.get(1)?; - let last: Option = row.get(2)?; - Ok((ns, count, last)) - })?; - let mut out = Vec::new(); - for r in rows { - let (ns, count, last) = r?; - out.push(NamespaceSummary { - namespace: ns, - count: usize::try_from(count).unwrap_or(0), - last_updated: last.map(timestamp_to_rfc3339), - }); - } - Ok(out) + let conn = Arc::clone(&self.conn); + tokio::task::spawn_blocking(move || Self::namespace_summaries_blocking(&conn)) + .await + .context("join Memory::namespace_summaries")? } async fn count(&self) -> anyhow::Result { - let conn = self.conn.lock(); - let count: i64 = - conn.query_row("SELECT COUNT(*) FROM memory_docs", [], |row| row.get(0))?; - usize::try_from(count).context("negative count") + let conn = Arc::clone(&self.conn); + tokio::task::spawn_blocking(move || Self::count_blocking(&conn)) + .await + .context("join Memory::count")? } async fn health_check(&self) -> bool { diff --git a/crates/tinymemory-tinycortex/Cargo.toml b/crates/tinymemory-tinycortex/Cargo.toml index efa35e2..22ae6ef 100644 --- a/crates/tinymemory-tinycortex/Cargo.toml +++ b/crates/tinymemory-tinycortex/Cargo.toml @@ -22,14 +22,16 @@ tinymemory-api = { path = "../tinymemory-api" } # `Memory` traits. tinycortex = { version = "0.1", default-features = false } # The optional capability families lifted here in issue #18 §C3 delegate to -# `tinymemory-core` on a blocking thread — that is where the summary tree, -# chunk store, entities, graph and diff ledger actually live. Depending on it +# `tinymemory-core` — that is where the summary tree, chunk store, entities, +# graph and diff ledger actually live. The synchronous-SQL families run on a +# blocking thread; the KV/graph accessors go through the client shim inline. Depending on it # makes this adapter heavier than the mandatory-only version it replaces; §D # feature-gates that weight once §A3 has removed the direct engine call sites # core still has. tinymemory-core = { path = "../tinymemory-core" } -# `spawn_blocking`: every family method runs synchronous engine work off the -# async executor rather than blocking it. +# `spawn_blocking`: the families that run synchronous engine work (profile, +# episodic, goals — and, via `tinymemory-core`, the mandatory Memory methods) +# hop off the async executor rather than blocking it. tokio = { version = "1", features = ["rt"] } # Timestamps on ingest and diff records. chrono = { version = "0.4", features = ["serde"] } diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index 9a1ae2f..7772f55 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -56,6 +56,9 @@ use tinymemory_api::types::{ NamespaceDocumentInput, NamespaceMemoryHit, NamespaceRetrievalContext, NamespaceSummary, StoredMemoryDocument, }; +// The KV read paths must address rows by the same canonical form the +// write-path shim stores them under — see `MemoryGraph::kv_get` below. +use tinymemory_core::store::safety::canonical_identifier; use tinymemory_core::store::{MemoryClient, MemoryClientRef}; /// The concrete, credential-free host configuration available inside a module. @@ -538,6 +541,12 @@ impl MemoryGraph for TinycortexProvider { namespace: Option<&str>, key: &str, ) -> Result, MemoryError> { + // The write path stores `canonical_identifier(key)` (the KV shim in + // `tinymemory-core` canonicalizes on the way in), so the stored keys + // are canonical and a raw-key comparison misses every rewritten key. + // `kv_delete` already goes through the shim; this lookup has to apply + // the same transform or put→get misses while put→delete works. + let key = canonical_identifier(key); let record = self .client .kv_records(namespace) @@ -579,7 +588,13 @@ impl MemoryGraph for TinycortexProvider { .await .map_err(|error| Self::other("kv_list", error))?; if let Some(prefix) = prefix { - records.retain(|record| record.key.starts_with(prefix)); + // Stored keys are canonical (see `kv_get`), so prefix matching is + // over canonical stored keys: the caller's prefix is canonicalized + // before comparing. A prefix that truncates a PII pattern + // mid-match stays raw (the transform is a no-op on it) and will + // not reach a rewritten key — the placeholder is the stored form. + let prefix = canonical_identifier(prefix); + records.retain(|record| record.key.starts_with(&prefix)); } records.truncate(limit); Self::cross(&records, "convert key/value records") diff --git a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs index e871856..38865e0 100644 --- a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs +++ b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs @@ -118,3 +118,106 @@ async fn the_full_provider_actually_retains() { almost nothing" ); } + +/// The KV write path canonicalizes identifiers (the shim in `tinymemory-core` +/// routes every `set_*`/`delete_*` through `canonical_identifier`), so a read +/// path that compares the raw caller key misses every rewritten key: put→get +/// answered `None` while put→delete answered `true`. `kv_get` and `kv_list` +/// must apply the same transform the write path did. +#[tokio::test(flavor = "multi_thread")] +async fn kv_reads_find_a_key_the_canonicalizer_rewrites() { + use tinymemory_api::provider::MemoryProvider; + use tinymemory_core::store::safety::canonical_identifier; + + let workspace = tempfile::tempdir().expect("workspace"); + let provider = provider_over(workspace.path()); + let graph = provider.as_graph().expect("the full provider serves Graph"); + + // A formatted national ID is strict-gated PII, so the write path rewrites + // it. (A bare Luhn-valid digit run would NOT do here: the strict gate + // deliberately ignores bare-numeric shapes so scanner-built identifiers — + // timestamps, phone-shaped JIDs — keep their identity.) + let key = "ssn-123-45-6789"; + let canonical = canonical_identifier(key); + assert_ne!( + canonical, key, + "fixture must be a key the canonicalizer rewrites" + ); + + let value = serde_json::json!({"ticket": 42}); + graph + .kv_put(None, key, value.clone()) + .await + .expect("kv_put"); + + let record = graph + .kv_get(None, key) + .await + .expect("kv_get") + .expect("kv_get must find the key it just put under the same raw key"); + assert_eq!(record.value, value, "kv_get surfaced another record"); + assert_eq!( + record.key, canonical, + "the stored key is the canonical form, and reads surface it as stored" + ); + + // Prefix matching is over canonical stored keys, so the raw caller key + // works as a prefix of its own record. + let listed = graph.kv_list(None, Some(key), 16).await.expect("kv_list"); + assert!( + listed.iter().any(|r| r.key == canonical), + "kv_list under the raw-key prefix must reach the rewritten record, got {listed:?}" + ); + + // Delete already routed through the canonicalizing shim; the fix must not + // break that half of the symmetry. + assert!( + graph.kv_delete(None, key).await.expect("kv_delete"), + "kv_delete must find the rewritten key" + ); + assert!( + graph + .kv_get(None, key) + .await + .expect("kv_get after delete") + .is_none(), + "the record must be gone after kv_delete reported true" + ); +} + +/// The same symmetry holds for namespaced KV rows: namespace and key are both +/// canonicalized on write, so both must be canonicalized on read. +#[tokio::test(flavor = "multi_thread")] +async fn namespaced_kv_reads_apply_the_write_path_canonicalization() { + use tinymemory_api::provider::MemoryProvider; + + let workspace = tempfile::tempdir().expect("workspace"); + let provider = provider_over(workspace.path()); + let graph = provider.as_graph().expect("the full provider serves Graph"); + + // A namespace the canonicalizer rewrites, guarded like the key leg — a + // no-op namespace would prove only key symmetry under a namespace. + let ns = "ssn-123-45-6789"; + assert_ne!( + tinymemory_core::store::safety::canonical_identifier(ns), + ns, + "the fixture namespace must be one the canonicalizer rewrites" + ); + let key = "cliente-RFC-VECJ880326XK4"; + let value = serde_json::json!("rewritten"); + graph + .kv_put(Some(ns), key, value.clone()) + .await + .expect("kv_put"); + + let record = graph + .kv_get(Some(ns), key) + .await + .expect("kv_get") + .expect("a namespaced put must be readable back under the same raw key"); + assert_eq!(record.value, value); + assert!( + graph.kv_delete(Some(ns), key).await.expect("kv_delete"), + "namespaced kv_delete must stay symmetric with kv_put" + ); +}