Sync active TinyMemory pin with async resolver

crates/tinymemory-api/src/error/mod.rs

@@ -7,6 +7,10 @@
 
 use serde::Serialize;
 
+/// The suffix of a hosted engine's codes for a feature the account's plan
+/// does not include (`MEMORY_PLAN_REQUIRED`).
+pub const PLAN_REQUIRED_SUFFIX: &str = "_PLAN_REQUIRED";
+
 /// Every way a memory operation can fail.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, thiserror::Error)]
 #[serde(rename_all = "snake_case")]
@@ -62,15 +66,36 @@ impl Error {
         }
     }
 
-    /// Whether the failure is known to affect every request on the account.
+    /// Whether the failure is known to affect every request on the account,
+    /// including a plan refusal ([`Self::is_plan_required`]).
     /// `Unavailable` is deliberately excluded: it describes a transient
     /// request failure and carries no typed proof of an account-wide outage.
     #[must_use]
     pub fn is_account_wide(&self) -> bool {
         matches!(
             self,
             Self::Off(_) | Self::Unauthorized(_) | Self::InsufficientCredits(_)
-        )
+        ) || self.is_plan_required()
+    }
+
+    /// Whether the engine refused because the account's plan does not include
+    /// the feature: an [`Self::Engine`] message led by a well-formed
+    /// `[CODE] ` whose code ends in [`PLAN_REQUIRED_SUFFIX`]. Not a credential
+    /// fault, so a host keeps the session and shows an upgrade path.
+    #[must_use]
+    pub fn is_plan_required(&self) -> bool {
+        let Self::Engine(message) = self else {
+            return false;
+        };
+        message
+            .strip_prefix('[')
+            .and_then(|rest| rest.split_once("] "))
+            .is_some_and(|(code, _)| {
+                code.ends_with(PLAN_REQUIRED_SUFFIX)
+                    && code
+                        .chars()
+                        .all(|c| c.is_ascii_uppercase() || c.is_ascii_digit() || c == '_')
+            })
     }
 
     /// Whether retrying the same call later may succeed.

crates/tinymemory-api/src/error/mod_tests.rs

@@ -68,3 +68,19 @@ fn request_timeouts_and_rate_limits_are_transient_not_account_wide() {
         assert!(!error.is_account_wide(), "{error:?}");
     }
 }
+
+#[test]
+fn a_plan_refusal_is_account_wide_only_with_a_well_formed_code() {
+    let plan = Error::Engine("[MEMORY_PLAN_REQUIRED] memory API recall: not in the plan".into());
+    assert!(plan.is_plan_required());
+    assert!(plan.is_account_wide());
+    for other in [
+        Error::Engine("saw [MEMORY_PLAN_REQUIRED] mid-message".into()),
+        Error::Engine("[not a code_PLAN_REQUIRED] memory API x".into()),
+        Error::Engine("[USER_INSUFFICIENT_CREDITS] top up".into()),
+        Error::Unavailable("[MEMORY_PLAN_REQUIRED] x".into()),
+    ] {
+        assert!(!other.is_plan_required(), "{other:?}");
+        assert!(!other.is_account_wide(), "{other:?}");
+    }
+}

crates/tinymemory-integrations/src/cortex/engine/mod_hosted_tests.rs

@@ -521,3 +521,44 @@ async fn a_visibility_timeout_says_whether_the_listing_was_rate_limited() {
         "{missing}"
     );
 }
+
+#[tokio::test]
+async fn a_forget_is_sent_in_batches_and_a_retry_that_finds_them_gone_succeeds() {
+    use axum::http::StatusCode;
+    use axum::routing::post;
+    use axum::{Json, Router};
+    use std::sync::Mutex;
+    let bodies: Arc<Mutex<Vec<serde_json::Value>>> = Arc::default();
+    let seen = bodies.clone();
+    // Each batch's first send fails transiently; its retry finds the
+    // events already gone (404), which is success for an idempotent removal.
+    let app = Router::new().route(
+        "/memory/forget",
+        post(move |Json(body): Json<serde_json::Value>| {
+            let seen = seen.clone();
+            async move {
+                let mut seen = seen.lock().unwrap();
+                seen.push(body);
+                if seen.len() % 2 == 1 {
+                    (StatusCode::SERVICE_UNAVAILABLE, Json(serde_json::json!({})))
+                } else {
+                    (StatusCode::NOT_FOUND, Json(serde_json::json!({})))
+                }
+            }
+        }),
+    );
+    let engine = hosted_engine(&serve(app).await);
+    let ids: Vec<String> = (0..250).map(|i| format!("e{i}")).collect();
+    engine
+        .log
+        .forget_events("user:alice/app:learnings", &ids)
+        .await
+        .unwrap();
+    let sizes: Vec<usize> = bodies
+        .lock()
+        .unwrap()
+        .iter()
+        .map(|body| body["selector"]["memory_ids"].as_array().unwrap().len())
+        .collect();
+    assert_eq!(sizes, vec![100, 100, 100, 100, 50, 50]);
+}

crates/tinymemory-integrations/src/cortex/engine/mod_proxied_tests.rs

@@ -184,6 +184,28 @@ async fn a_narrower_erasure_forgets_each_scopes_events() {
     all_named(&proxy, "alice");
 }
 
+#[tokio::test]
+async fn erasing_only_the_tenant_node_does_not_purge_the_tenant() {
+    let (endpoint, proxy) = proxied_double().await;
+    let engine = proxied_engine(&endpoint);
+    engine.store(learning("user:alice")).await.unwrap();
+    engine
+        .store(learning("user:alice/agent:helper"))
+        .await
+        .unwrap();
+    let alice = proxy.tenant("alice").unwrap();
+    assert_eq!(alice.event_count(), 2);
+    engine
+        .erase(EraseRequest::new(Reach::exact(
+            "user:alice".parse().unwrap(),
+        )))
+        .await
+        .unwrap();
+    assert!(proxy.purged.lock().unwrap().is_empty(), "no tenant purge");
+    assert_eq!(alice.event_count(), 1, "the agent's learning below is kept");
+    all_named(&proxy, "alice");
+}
+
 #[tokio::test]
 async fn erasing_the_whole_tree_names_no_tenant_and_is_refused() {
     let (endpoint, proxy) = proxied_double().await;

crates/tinymemory-integrations/src/cortex/error/mod.rs

@@ -60,9 +60,7 @@ pub type Result<T> = std::result::Result<T, Error>;
 /// The TinyHumans backend's code for an exhausted credit balance (HTTP 402).
 pub(crate) const INSUFFICIENT_CREDITS_CODE: &str = "USER_INSUFFICIENT_CREDITS";
 
-/// The suffix of the TinyHumans backend's codes for a feature the account's
-/// plan does not include (`MEMORY_PLAN_REQUIRED`).
-pub(crate) const PLAN_REQUIRED_SUFFIX: &str = "_PLAN_REQUIRED";
+pub(crate) use tinymemory_api::error::PLAN_REQUIRED_SUFFIX;
 
 /// The message every variant carries.
 fn message_of(error: &Error) -> &str {
@@ -101,8 +99,7 @@ pub fn error_code(error: &Error) -> Option<&str> {
 /// and shows an upgrade path.
 #[must_use]
 pub fn is_plan_required(error: &Error) -> bool {
-    matches!(error, Error::Engine(_))
-        && error_code(error).is_some_and(|code| code.ends_with(PLAN_REQUIRED_SUFFIX))
+    error.is_plan_required()
 }
 
 #[cfg(test)]

crates/tinymemory-integrations/src/sources/sync/mod.rs

@@ -39,10 +39,6 @@ pub const MIN_SCHEDULE_MINS: u32 = 15;
 /// `ceil(N / 100)` instead of `N`.
 pub const STORE_CHUNK: usize = MAX_STORE_MANY;
 
-/// The suffix of the hosted engine's codes for a feature the account's plan
-/// does not include (`MEMORY_PLAN_REQUIRED`).
-const PLAN_REQUIRED_SUFFIX: &str = "_PLAN_REQUIRED";
-
 /// Why a sync failed.
 #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
 pub enum SyncError {
@@ -64,21 +60,12 @@ impl SyncError {
     #[must_use]
     pub fn is_account_wide(&self) -> bool {
         match self {
-            Self::Engine(error) => error.is_account_wide() || is_plan_refusal(error),
+            Self::Engine(error) => error.is_account_wide(),
             _ => false,
         }
     }
 }
 
-/// Whether the hosted engine refused because the account's plan lacks the
-/// feature: an `Engine` error led by a `[*_PLAN_REQUIRED] ` code.
-fn is_plan_refusal(error: &tinymemory_api::Error) -> bool {
-    matches!(error, tinymemory_api::Error::Engine(message) if message
-        .strip_prefix('[')
-        .and_then(|rest| rest.split_once("] "))
-        .is_some_and(|(code, _)| code.ends_with(PLAN_REQUIRED_SUFFIX)))
-}
-
 /// A source as the host's registry holds it.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RegisteredSource {

crates/tinymemory-integrations/tests/legacy_importer.rs

@@ -33,12 +33,16 @@ enum Mode {
     Credits,
     /// Refuses these legacy ids as invalid, taking the rest.
     Refuse(HashSet<String>),
+    /// Refuses every bulk store as invalid, so items go one at a time, and
+    /// is unavailable after this many single stores.
+    DownAfter(usize),
 }
 
 struct Scripted {
     inner: ReferenceEngine,
     mode: Mutex<Mode>,
     calls: AtomicUsize,
+    singles: AtomicUsize,
 }
 
 impl Scripted {
@@ -47,6 +51,7 @@ impl Scripted {
             inner: ReferenceEngine::new(),
             mode: Mutex::new(mode),
             calls: AtomicUsize::new(0),
+            singles: AtomicUsize::new(0),
         })
     }
 
@@ -72,6 +77,16 @@ impl Scripted {
                     Ok(())
                 }
             }
+            Mode::DownAfter(_) if items.len() > 1 => {
+                Err(Error::InvalidRequest("batch too large".into()))
+            }
+            Mode::DownAfter(limit) => {
+                if self.singles.fetch_add(1, Ordering::SeqCst) < limit {
+                    Ok(())
+                } else {
+                    Err(Error::Unavailable("engine down".into()))
+                }
+            }
         }
     }
 }
@@ -524,3 +539,17 @@ fn the_persisted_file_keeps_its_wire_shape() {
         serde_json::from_str(r#"{"state":{"phase":"done","imported":2,"total":2}}"#).unwrap();
     assert_eq!(old.state.phase, ImportPhase::Done);
 }
+
+#[tokio::test]
+async fn a_failure_in_the_one_by_one_fallback_checkpoints_the_last_stored_item() {
+    let dir = legacy();
+    let engine = Scripted::new(Mode::DownAfter(5));
+    let rig = rig(dir.path(), engine, true);
+    let file = import(&rig).await;
+    assert_eq!(file.state.imported, 5);
+    assert_eq!(
+        file.checkpoint.documents.as_deref(),
+        Some("d004"),
+        "the checkpoint must not pass an item that was never stored"
+    );
+}