Run the periodic memory sync loops inside the module

crates/tinymemory-core/src/global.rs

@@ -15,6 +15,13 @@
 //! let client = memory::global::client()?;
 //! client.put_doc(input).await?;
 //! ```
+//!
+//! There are two ways in, and which one a caller wants depends on whether it
+//! already holds a client. [`init`] builds one from a workspace directory;
+//! [`bind`] publishes a client the caller built itself, which is what a host
+//! that constructs its store through `store::factories` needs — calling [`init`]
+//! there would put a second client, and a second ingestion worker, over the same
+//! SQLite file.
 
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
@@ -293,6 +300,126 @@ pub fn client_if_ready() -> Option<MemoryClientRef> {
         .map(|entry| Arc::clone(&entry.client))
 }
 
+/// Register an **already-built** client as the one for `workspace_dir`.
+///
+/// # Why this exists beside [`init`]
+///
+/// [`init`] *constructs* the client, which is right for a caller that owns the
+/// workspace and wants whatever client it implies. It is wrong for a caller
+/// that has already built one, and that caller now exists: the loadable
+/// TinyMemory module builds its store through
+/// `store::factories::create_memory_client_with_local_ai` — it has to, because
+/// only that entry point takes the module's own embedding routes, storage
+/// provider and workspace — and *then* finds that every runner in
+/// `sync::pipelines::host` begins with [`client_if_ready`].
+///
+/// Reaching for [`init`] there would build a **second** [`MemoryClient`] over
+/// the same SQLite file: two ingestion workers, duplicate graph extraction and
+/// duplicate embedding work, which is precisely the hazard the per-workspace
+/// cache and [`init`]'s reuse checks exist to prevent. The fix is to publish the
+/// client that already exists rather than to construct another one.
+///
+/// Writes into **both** resolution paths — the global slot and the
+/// per-workspace cache — so [`client_if_ready`], [`client`] and
+/// [`client_for_workspace`] converge on the one client. That convergence is the
+/// invariant [`init`] already works to preserve; a `bind` that wrote only the
+/// slot would leave `client_for_workspace` free to build a second client for the
+/// same workspace, which is the same hazard by another route.
+///
+/// A workspace that differs from the one currently bound *rebinds*, with the
+/// same log [`init`] emits, because a caller that hands over a client for
+/// another workspace is making the same active-user-switch statement.
+///
+/// # A different client for the same workspace is refused
+///
+/// The one case that must not pass silently. `cache_client`'s rule is that a
+/// racing caller's client wins and the loser uses the returned handle — free for
+/// [`init`], whose caller only wanted *a* client. A `bind` caller is different:
+/// it is already using the client it passed, so quietly handing back somebody
+/// else's would neither retire the caller's client nor stop its worker. Two
+/// clients already exist at that point; the honest report is an error naming it,
+/// and the global slot is left as it was rather than repointed at a client the
+/// caller is not the one using.
+///
+/// # Errors
+///
+/// Lock poisoning, or a *different* client already bound for `workspace_dir`.
+pub fn bind(workspace_dir: PathBuf, client: MemoryClientRef) -> Result<MemoryClientRef, String> {
+    bind_in_slot(global_slot(), workspace_dir, client)
+}
+
+/// Implementation backing [`bind`] — extracted for the same reason
+/// [`client_from`] is, so the refusal and the rebind can be asserted against a
+/// local slot instead of racing the process-global singleton.
+fn bind_in_slot(
+    slot: &GlobalClientSlot,
+    workspace_dir: PathBuf,
+    client: MemoryClientRef,
+) -> Result<MemoryClientRef, String> {
+    // Global slot first, then the workspace cache. `init` and
+    // `client_for_workspace` both take the two in that order — `init` calls
+    // `cache_client` while holding the slot's write guard — and a third entry
+    // point taking them the other way round is an ABBA deadlock against a
+    // concurrent init.
+    let mut guard = slot
+        .write()
+        .map_err(|e| format!("[memory:global] write lock poisoned: {e}"))?;
+
+    let published = cache_client(&workspace_dir, &client)?;
+    if !Arc::ptr_eq(&published, &client) {
+        return Err(already_bound(&workspace_dir));
+    }
+
+    if let Some(existing) = guard.as_ref() {
+        if existing.workspace_dir == workspace_dir {
+            // The same client bound twice: idempotent, and the shape a retried
+            // setup produces.
+            if Arc::ptr_eq(&existing.client, &published) {
+                log::debug!(
+                    "[memory:global] MemoryClient already bound for {}",
+                    workspace_dir.display()
+                );
+                return Ok(published);
+            }
+            // Reachable only if something published to the slot without
+            // publishing to the cache — no path in this module does — so this is
+            // a contract violation rather than a race. It is the double-client
+            // hazard either way, so it gets the same refusal.
+            return Err(already_bound(&workspace_dir));
+        }
+
+        log::info!(
+            "[memory:global] rebinding MemoryClient workspace {} -> {}",
+            existing.workspace_dir.display(),
+            workspace_dir.display()
+        );
+    }
+
+    log::info!(
+        "[memory:global] binding a caller-built MemoryClient workspace={}",
+        workspace_dir.display()
+    );
+    *guard = Some(GlobalMemoryClient {
+        workspace_dir,
+        client: Arc::clone(&published),
+    });
+    Ok(published)
+}
+
+/// The refusal [`bind`] returns when a second client already owns a workspace.
+///
+/// Names the hazard rather than the symptom: the caller's next question is
+/// always "so which client is the store actually using?", and the answer is that
+/// two of them are.
+fn already_bound(workspace_dir: &Path) -> String {
+    format!(
+        "[memory:global] a different MemoryClient is already bound for {} — binding this one \
+         would leave two clients, and two ingestion workers, over the same store; build the \
+         client once and bind that",
+        workspace_dir.display()
+    )
+}
+
 #[cfg(test)]
 #[path = "global_tests.rs"]
 mod tests;

crates/tinymemory-core/src/global_tests.rs

@@ -118,6 +118,119 @@ async fn client_returns_a_handle_after_explicit_init() {
     let _arc: Arc<MemoryClient> = c;
 }
 
+/// The whole point of `bind`: the client the caller already built becomes the
+/// one every resolution path answers with, without a second one being built.
+#[tokio::test]
+async fn bind_publishes_a_caller_built_client_to_both_resolution_paths() {
+    crate::test_seams::init();
+    let slot = GlobalClientSlot::default();
+    let tmp = TempDir::new().unwrap();
+    let workspace = tmp.path().join("ws-bound");
+    let client: MemoryClientRef =
+        Arc::new(MemoryClient::from_workspace_dir(workspace.clone()).unwrap());
+
+    let bound = bind_in_slot(&slot, workspace.clone(), Arc::clone(&client)).unwrap();
+
+    assert!(
+        Arc::ptr_eq(&bound, &client),
+        "bind must not swap the client"
+    );
+    assert!(Arc::ptr_eq(&client_from(&slot).unwrap(), &client));
+    // The per-workspace cache is the half a slot-only bind would miss, and
+    // missing it lets `client_for_workspace` build a second engine over the
+    // same store.
+    assert!(Arc::ptr_eq(
+        &client_for_workspace(&workspace).unwrap(),
+        &client
+    ));
+}
+
+/// Re-binding the same client is what a retried setup produces, and must not
+/// read as the double-client hazard.
+#[tokio::test]
+async fn binding_the_same_client_twice_is_idempotent() {
+    crate::test_seams::init();
+    let slot = GlobalClientSlot::default();
+    let tmp = TempDir::new().unwrap();
+    let workspace = tmp.path().join("ws-bound-twice");
+    let client: MemoryClientRef =
+        Arc::new(MemoryClient::from_workspace_dir(workspace.clone()).unwrap());
+
+    let first = bind_in_slot(&slot, workspace.clone(), Arc::clone(&client)).unwrap();
+    let second = bind_in_slot(&slot, workspace, Arc::clone(&client)).unwrap();
+
+    assert!(Arc::ptr_eq(&first, &second));
+}
+
+/// The case that would reintroduce the hazard `bind` exists to avoid: a second
+/// client over one workspace must be named, not absorbed.
+#[tokio::test]
+async fn binding_a_different_client_for_one_workspace_is_refused() {
+    crate::test_seams::init();
+    let slot = GlobalClientSlot::default();
+    let tmp = TempDir::new().unwrap();
+    let workspace = tmp.path().join("ws-two-clients");
+    let first: MemoryClientRef =
+        Arc::new(MemoryClient::from_workspace_dir(workspace.clone()).unwrap());
+    let second: MemoryClientRef =
+        Arc::new(MemoryClient::from_workspace_dir(workspace.clone()).unwrap());
+
+    bind_in_slot(&slot, workspace.clone(), Arc::clone(&first)).unwrap();
+    let error = match bind_in_slot(&slot, workspace.clone(), Arc::clone(&second)) {
+        Ok(_) => panic!("a second client over one workspace must not bind"),
+        Err(error) => error,
+    };
+
+    assert!(error.contains("already bound"), "{error}");
+    // And the refusal leaves the binding alone rather than repointing it at a
+    // client the caller that owns the slot is not the one using.
+    assert!(Arc::ptr_eq(&client_from(&slot).unwrap(), &first));
+    assert!(Arc::ptr_eq(
+        &client_for_workspace(&workspace).unwrap(),
+        &first
+    ));
+}
+
+/// A bind for another workspace is the active-user-switch shape `init` already
+/// handles, so it rebinds rather than refusing.
+#[tokio::test]
+async fn bind_rebinds_when_the_workspace_changes() {
+    crate::test_seams::init();
+    let slot = GlobalClientSlot::default();
+    let tmp = TempDir::new().unwrap();
+    let workspace_a = tmp.path().join("ws-bind-a");
+    let workspace_b = tmp.path().join("ws-bind-b");
+    let client_a: MemoryClientRef =
+        Arc::new(MemoryClient::from_workspace_dir(workspace_a.clone()).unwrap());
+    let client_b: MemoryClientRef =
+        Arc::new(MemoryClient::from_workspace_dir(workspace_b.clone()).unwrap());
+
+    bind_in_slot(&slot, workspace_a, Arc::clone(&client_a)).unwrap();
+    bind_in_slot(&slot, workspace_b, Arc::clone(&client_b)).unwrap();
+
+    assert!(Arc::ptr_eq(&client_from(&slot).unwrap(), &client_b));
+}
+
+/// `init` and `bind` must not disagree about which client owns a workspace,
+/// whichever ran first.
+#[tokio::test]
+async fn init_after_bind_reuses_the_bound_client() {
+    crate::test_seams::init();
+    let slot = GlobalClientSlot::default();
+    let tmp = TempDir::new().unwrap();
+    let workspace = tmp.path().join("ws-bind-then-init");
+    let client: MemoryClientRef =
+        Arc::new(MemoryClient::from_workspace_dir(workspace.clone()).unwrap());
+
+    bind_in_slot(&slot, workspace.clone(), Arc::clone(&client)).unwrap();
+    let from_init = init_in_slot(&slot, workspace).unwrap();
+
+    assert!(
+        Arc::ptr_eq(&from_init, &client),
+        "init must reuse the bound client rather than construct a second one"
+    );
+}
+
 #[tokio::test]
 async fn client_errs_clearly_when_not_initialised() {
     crate::test_seams::init();

crates/tinymemory-module/src/config.rs

@@ -24,6 +24,15 @@
 //! split is not a hard isolation boundary and is not claimed as one — it is a
 //! refusal to widen what crosses a boundary that already exists.
 //!
+//! The Composio fields are where that line is easiest to misread, so it is drawn
+//! explicitly: [`ModuleConfig::composio_mode`] and
+//! [`ModuleConfig::composio_entity_id`] are *routing*, not access. The mode says
+//! which branch the sync pipelines take and the entity says whose connected
+//! accounts a call addresses; neither authorises anything. The direct-mode API
+//! key and the backend session bearer both stay out — the first is fetched over
+//! the bus per call ([`crate::composio`]), and the second is refused outright,
+//! which is why backend-mode Composio sync cannot run inside this module.
+//!
 //! # `MemoryConfig` travels whole
 //!
 //! The engine's own configuration is `tinymemory_api::host::MemoryConfig`,
@@ -122,6 +131,67 @@ pub struct ModuleConfig {
     /// never embed a URL or a token — it appears in status output and audit
     /// events.
     pub driver_id: String,
+
+    /// The user's global memory-sync cadence, in seconds.
+    ///
+    /// `None` means the host stated no choice, and the engine falls back to
+    /// `DEFAULT_MEMORY_SYNC_INTERVAL_SECS` — 24h, floored at each provider's own
+    /// minimum. `Some(0)` is "Manual only" and stops the periodic loops from
+    /// firing any source. Anything else is the user's own cadence.
+    ///
+    /// # Why the default is `None` and not `Some(0)`
+    ///
+    /// A host too old to send this field is deserialized through the struct's
+    /// `#[serde(default)]`, so whatever [`Default`] says here is what an older
+    /// host silently means. The two candidates fail in opposite directions and
+    /// they are not symmetrical:
+    ///
+    /// - `Some(0)` reads as manual-only, which is the exact failure this field
+    ///   exists to remove: every source skipped on every tick, with no error, no
+    ///   warning, and nothing to distinguish it from a sync that ran and found
+    ///   nothing new. A memory that has quietly stopped updating looks identical
+    ///   to one that is up to date.
+    /// - `None` reads as "the user chose nothing", which is *true* of a host
+    ///   that sent nothing, and lands on the same 24h default the host applies
+    ///   to a user who never set one.
+    ///
+    /// So this defaults to `None`. The cost of getting that wrong is bounded and
+    /// visible — a user who picked "Manual only" gets a 24h background sync
+    /// until their host learns to send the field, and every one of those syncs
+    /// is still gated by the per-source `enabled` toggle, which *does* travel
+    /// here in [`Self::memory_sources`]. The cost of getting `Some(0)` wrong is
+    /// invisible by construction. Between a bounded over-sync a user can see and
+    /// a no-sync nobody can, this picks the one that can be noticed.
+    pub memory_sync_interval_secs: Option<u64>,
+
+    /// How the host routes Composio calls: `backend` or `direct`.
+    ///
+    /// Empty means the host stated no mode — an older host, or one with no
+    /// Composio integration configured — and is treated exactly as `backend` is:
+    /// not direct.
+    ///
+    /// # Only `direct` can be served from inside the module
+    ///
+    /// `sync::pipelines::host::composio_config` selects its direct branch on
+    /// this value, and its other branch needs a backend session bearer. This
+    /// struct has no field for one and deliberately never will: a bearer is a
+    /// credential, and a load-time snapshot could not follow one the host
+    /// refreshes mid-session in any case. See `EngineRuntimeConfig`'s
+    /// `session_token`, which names that refusal rather than reporting a
+    /// signed-out user.
+    ///
+    /// This is a *mode*, not a credential, and the distinction is load-bearing:
+    /// the direct-mode API key still does not travel here. It is fetched from
+    /// the host over the bus for the duration of one call — see
+    /// [`crate::composio`] — which is why this field can exist at all.
+    pub composio_mode: String,
+
+    /// The Composio entity the host authenticates as.
+    ///
+    /// An identifier rather than a credential: it selects whose connected
+    /// accounts a direct-mode call addresses, and holding it grants nothing on
+    /// its own. Empty is sent as no entity at all rather than as an empty one.
+    pub composio_entity_id: String,
 }
 
 impl Default for ModuleConfig {
@@ -145,6 +215,13 @@ impl Default for ModuleConfig {
             cloud_embedding_dimensions: 0,
             models_supporting_dimensions: Vec::new(),
             driver_id: tinymemory::registry::TINYCORTEX_DRIVER_ID.to_string(),
+            // The two below are what an older host means, and both are argued
+            // for on their own fields. In short: an absent cadence is "no
+            // choice", never "manual only"; an absent Composio mode is "not
+            // direct".
+            memory_sync_interval_secs: None,
+            composio_mode: String::new(),
+            composio_entity_id: String::new(),
         }
     }
 }

crates/tinymemory-module/src/config_loader.rs

@@ -42,21 +42,27 @@
 //! `signals = []`), not a bus *pull*: a pull would re-introduce the two-answers
 //! problem above while still being stale between ticks.
 //!
-//! # One gap this loader cannot paper over
+//! # The gap this loader used to have, and how it was closed
 //!
-//! `EngineRuntimeConfig::memory_sync_interval_secs` answers `Some(0)`, and
-//! the contract reads `Some(0)` as **manual only**. So a periodic sync loop
-//! started inside this process would consider every source manual and skip it —
-//! silently, which is the failure class this migration keeps producing.
+//! `EngineRuntimeConfig::memory_sync_interval_secs` answered the constant
+//! `Some(0)`, and the contract reads `Some(0)` as **manual only**. A periodic
+//! sync loop started inside this process therefore considered every source
+//! manual and skipped it — silently, which is the failure class this migration
+//! keeps producing.
 //!
-//! It is left as it is on purpose. `ModuleConfig` carries no cadence field, so
-//! answering anything else would mean this module *guessing* at a user setting
-//! it was never told — the same argument `crate::host` gives for refusing to
-//! synthesise a scheduler-gate policy from `ModuleConfig::scheduler_gate`, and
-//! the same conclusion: guessing is worse than not answering. The honest fix is
-//! for the host to send the cadence in `ModuleConfig`, at which point this
-//! loader answers it without further change. Until then, nothing in this
-//! process starts a periodic sync loop, and this note is why.
+//! The fix was not for this loader to invent a better number. Guessing at a user
+//! setting the module was never told is the same thing `crate::host` refuses to
+//! do when it declines to synthesise a scheduler-gate policy from
+//! `ModuleConfig::scheduler_gate`, and it has the same answer: guessing is worse
+//! than not answering. So the *host* now sends the cadence, as
+//! `ModuleConfig::memory_sync_interval_secs`, and this loader hands it back
+//! along with everything else. Nothing here needed changing, which is the point
+//! — the snapshot answers whatever the host put in it.
+//!
+//! What is left is the staleness above, and it now bites one more setting: a
+//! user who changes their sync cadence, or switches Composio between backend and
+//! direct mode, after this module loaded keeps the old value in this process
+//! until the host reloads the module.
 
 use std::sync::atomic::AtomicBool;
 use std::sync::Arc;