From 829add8485e2583ee69057374afc53233c54d218 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 25 Aug 2026 12:16:39 +0530 Subject: [PATCH 1/2] Run the periodic memory sync loops inside the module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The host starts `sync::composio::start_periodic_sync` and `start_workspace_periodic_sync` against the second, in-process engine it also boots. openhuman#5560 deletes that engine, so the loops move in here beside the queue worker pool. Three things had to be closed first, each of which failed quietly rather than loudly. The cadence read as manual-only: `EngineRuntimeConfig` answered the constant `Some(0)`, which `effective_interval_secs` maps to `None`, so both loops skipped every source on every tick with nothing logged. `ModuleConfig` now carries `memory_sync_interval_secs` and the accessor answers it. An absent field defaults to `None` — "no explicit choice", the 24h fallback — not to `Some(0)`, because an over-sync is bounded and visible while a no-sync is invisible by construction, and an older host's payload means whatever the default says. The Composio branch was never selected: `composio()` answered an empty mode, so `composio_config` fell to its backend branch and failed on a session bearer. `composio_mode` and `composio_entity_id` now cross as routing — the direct-mode key still comes from `ComposioHost` per call, and there is no field for a bearer. `session_token` therefore returns a named refusal instead of `Ok(None)`, which used to surface as "not configured" and send a reader after a sign-in that cannot help, and the loop is gated to direct mode rather than started to fail every tick forever. The module's client was not in the global slot: every pipeline run opens with `global::client_if_ready()`, and the module builds its store through `store::factories`, which never touches it. `global::bind` publishes the already-built client into the slot and the per-workspace cache; `init` would have built a second `MemoryClient` over the same SQLite file. A different client for one workspace is refused rather than absorbed. One degradation is documented, not fixed: the module's scheduler-gate stub always answers `Normal`, so neither loop honours the "Memory Tree off" and "signed out" pauses, and a re-enable does not wake them early. Co-Authored-By: Claude Opus 5 --- crates/tinymemory-core/src/global.rs | 127 +++++++ crates/tinymemory-core/src/global_tests.rs | 104 ++++++ crates/tinymemory-module/src/config.rs | 77 +++++ crates/tinymemory-module/src/config_loader.rs | 32 +- .../src/config_loader_test.rs | 44 +++ crates/tinymemory-module/src/config_test.rs | 79 +++++ crates/tinymemory-module/src/host.rs | 34 +- crates/tinymemory-module/src/lib.rs | 327 ++++++++++++++---- crates/tinymemory-module/src/provider.rs | 7 + crates/tinymemory-module/src/service/test.rs | 89 ++++- .../tinymemory-tinycortex/src/engine/mod.rs | 79 ++++- .../tinymemory-tinycortex/src/engine/test.rs | 84 ++++- .../tests/full_provider_conformance.rs | 7 + docs/specs/tinybus-module.md | 53 +++ 14 files changed, 1050 insertions(+), 93 deletions(-) diff --git a/crates/tinymemory-core/src/global.rs b/crates/tinymemory-core/src/global.rs index 99fd1ead..012333e5 100644 --- a/crates/tinymemory-core/src/global.rs +++ b/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 { .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 { + 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 { + // 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; diff --git a/crates/tinymemory-core/src/global_tests.rs b/crates/tinymemory-core/src/global_tests.rs index 97979799..680c99fe 100644 --- a/crates/tinymemory-core/src/global_tests.rs +++ b/crates/tinymemory-core/src/global_tests.rs @@ -118,6 +118,110 @@ async fn client_returns_a_handle_after_explicit_init() { let _arc: Arc = 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(); diff --git a/crates/tinymemory-module/src/config.rs b/crates/tinymemory-module/src/config.rs index 46b70661..0dd8b541 100644 --- a/crates/tinymemory-module/src/config.rs +++ b/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, + + /// 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(), } } } diff --git a/crates/tinymemory-module/src/config_loader.rs b/crates/tinymemory-module/src/config_loader.rs index da3cd8f8..45066485 100644 --- a/crates/tinymemory-module/src/config_loader.rs +++ b/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; diff --git a/crates/tinymemory-module/src/config_loader_test.rs b/crates/tinymemory-module/src/config_loader_test.rs index 6c646c3b..9c43c9bf 100644 --- a/crates/tinymemory-module/src/config_loader_test.rs +++ b/crates/tinymemory-module/src/config_loader_test.rs @@ -44,6 +44,50 @@ async fn load_answers_from_the_module_config() { assert_eq!(sources[0]["id"], "gmail:1"); } +/// The two settings the periodic sync loops gate on reach them through here. +/// +/// This is the end-to-end shape of the first two blockers: both loops reload +/// config on every tick through this loader, and both used to receive constants +/// instead of the host's answers — a cadence of `Some(0)`, which the contract +/// reads as manual-only and which skips every source with nothing logged, and an +/// empty Composio mode, which never selects the one branch a module can serve. +/// A regression here is invisible at runtime, so it is pinned at the seam. +#[tokio::test] +async fn the_loader_answers_the_hosts_cadence_and_composio_mode() { + let mut config = module_config("/tmp/module-workspace"); + config.memory_sync_interval_secs = Some(3_600); + config.composio_mode = "direct".to_string(); + config.composio_entity_id = "entity-7".to_string(); + + let answered = ModuleConfigLoader::new(&config) + .load() + .await + .expect("the module always has a config"); + + assert_eq!(answered.memory_sync_interval_secs(), Some(3_600)); + assert!(answered.composio().is_direct()); + assert_eq!(answered.composio().entity_id, "entity-7"); +} + +/// "Manual only" has to survive as itself. +/// +/// The cadence is the one field where two different values produce the same +/// observable behaviour — a loop that fires nothing — so a bug that turned a +/// user's `Some(0)` into a default cadence, or a default into `Some(0)`, would +/// be found only by a user noticing their data was wrong weeks later. +#[tokio::test] +async fn a_manual_only_cadence_survives_the_loader_intact() { + let mut config = module_config("/tmp/module-workspace"); + config.memory_sync_interval_secs = Some(0); + + let answered = ModuleConfigLoader::new(&config) + .load() + .await + .expect("the module always has a config"); + + assert_eq!(answered.memory_sync_interval_secs(), Some(0)); +} + /// The one field that would smuggle a credential back out. #[tokio::test] async fn the_loader_hands_back_no_carried_credential() { diff --git a/crates/tinymemory-module/src/config_test.rs b/crates/tinymemory-module/src/config_test.rs index c8e630b0..e74cbf05 100644 --- a/crates/tinymemory-module/src/config_test.rs +++ b/crates/tinymemory-module/src/config_test.rs @@ -132,6 +132,85 @@ fn stripping_is_idempotent() { assert!(!config.strip_host_credentials()); } +/// The older-host case, stated as a test rather than as a hope. +/// +/// The host and the module are compiled and released separately, so a host that +/// predates these three fields sends JSON without them. The struct's +/// `#[serde(default)]` fills them from [`ModuleConfig::default`], and what that +/// resolves to is a product decision argued on each field — so it is pinned +/// here, where changing it fails a test instead of changing behaviour quietly. +#[test] +fn a_host_that_predates_the_sync_fields_gets_the_documented_defaults() { + // Every other key present, the three new ones absent: exactly the payload an + // older host sends. + let json = serde_json::json!({ + "workspace_dir": "/tmp/w", + "driver_id": "tinymemory", + }); + let config: ModuleConfig = serde_json::from_value(json).expect("an older host's config loads"); + + // `None`, not `Some(0)`. `Some(0)` is manual-only, which skips every source + // on every tick with nothing logged; `None` is "no explicit choice" and + // lands on the same 24h default the host applies to a user who set none. + assert_eq!( + config.memory_sync_interval_secs, None, + "an absent cadence must not read as manual-only" + ); + // Not direct, which is what an unconfigured Composio integration should look + // like, and exactly what the engine answered before this field existed. + assert!(config.composio_mode.is_empty()); + assert!(config.composio_entity_id.is_empty()); +} + +/// The cadence is a wire value with three meanings, and all three have to +/// survive the trip. +#[test] +fn every_cadence_the_host_can_state_round_trips() { + for cadence in [None, Some(0), Some(86_400)] { + let config = ModuleConfig { + workspace_dir: "/tmp/w".into(), + memory_sync_interval_secs: cadence, + ..ModuleConfig::default() + }; + let json = serde_json::to_string(&config).expect("serializes"); + let back: ModuleConfig = serde_json::from_str(&json).expect("deserializes"); + + assert_eq!( + back.memory_sync_interval_secs, cadence, + "the host's cadence must reach the module unchanged" + ); + } +} + +/// Routing crosses; access does not. +/// +/// The Composio fields are the closest this struct comes to the credential line, +/// so the distinction is asserted rather than described: a mode and an entity +/// travel, and neither the direct-mode key nor a backend bearer has anywhere to +/// travel in. +#[test] +fn the_composio_fields_carry_routing_and_not_access() { + let config = ModuleConfig { + workspace_dir: "/tmp/w".into(), + composio_mode: "direct".to_string(), + composio_entity_id: "entity-42".to_string(), + ..ModuleConfig::default() + }; + + let json = serde_json::to_string(&config).expect("serializes"); + let back: ModuleConfig = serde_json::from_str(&json).expect("deserializes"); + + assert_eq!(back.composio_mode, "direct"); + assert_eq!(back.composio_entity_id, "entity-42"); + // The structural credential check above scans field *names*; this is the + // other half — there is no key at all for either Composio secret, so a + // direct key or a session bearer has nowhere to be put. + let value: serde_json::Value = serde_json::from_str(&json).expect("valid json"); + let object = value.as_object().expect("config is a json object"); + assert!(!object.contains_key("composio_api_key")); + assert!(!object.contains_key("session_token")); +} + #[test] fn a_populated_config_round_trips() { let config = ModuleConfig { diff --git a/crates/tinymemory-module/src/host.rs b/crates/tinymemory-module/src/host.rs index 62715f93..74130f92 100644 --- a/crates/tinymemory-module/src/host.rs +++ b/crates/tinymemory-module/src/host.rs @@ -172,8 +172,30 @@ pub(crate) fn install(connection: Connection) { // the first time anything actually consults it. The queue worker pool now runs // in here (`crate::start_queue_pool`) and consults both, so the scheduler-gate // report fires on every boot that drains a job — which is the honest signal that -// the throttle is not in effect. The two periodic sync loops are still started -// host-side, so nothing in this process reaches them through that path yet. +// the throttle is not in effect. +// +// # The scheduler-gate stub now also silences two user-visible pauses +// +// `crate::start_sync_loops` moved the two periodic sync loops in here as well, +// and they consult this gate for something the queue pool does not: not a +// throttle but a *stop*. Step 0 of every tick in both loops is +// `sync::composio::periodic::periodic_pause_reason`, which exists to honour two +// states — `PauseReason::UserDisabled`, the user switching Memory Tree off in +// Settings, and `PauseReason::SignedOut`, no live session. It reads them off +// `current_policy()`, which is the stub, which always answers `Policy::Normal`. +// So in module mode both loops tick straight through both pauses: a user who +// switched memory off still gets background fetches, and a signed-out user still +// gets a Composio connection walk every 20 minutes. +// +// The per-source `enabled` toggle is unaffected — that is read from the source +// registry inside the tick, not from the gate — so switching off one source +// still works. It is the two *global* pauses that do not. +// +// The same stub's `resume_notify` hands back a `Notify` nobody fires, so the +// other half of that design is gone too: re-enabling sync no longer wakes the +// loops early, and the user waits out the remaining 20-minute tick instead of +// syncing within seconds. That half is benign; the paragraph above is not, and +// closing it needs the same `SchedulerGate` bus interface named below. // // Note also what is deliberately *not* built here: a module-local registry that // banked shutdown hooks and drained them on the module's own `Shutdown` method. @@ -193,7 +215,9 @@ static SHUTDOWN_REPORTED: AtomicBool = AtomicBool::new(false); /// What the missing scheduler gate costs, in the terms a reader of the log needs. const GATE_UNSERVED: &str = "scheduler gate unserved in module mode: background memory work in \ this process runs ungated, ignoring the host's background-AI \ - throttle (user toggle, AC power, CPU pressure, signed-out)"; + throttle (user toggle, AC power, CPU pressure, signed-out) — and \ + the periodic sync loops here therefore also ignore the \ + \"Memory Tree off\" and \"signed out\" pauses that would stop them"; /// What the missing shutdown host costs. const SHUTDOWN_UNSERVED: &str = "shutdown host unserved in module mode: a memory shutdown hook \ @@ -295,8 +319,8 @@ pub(crate) fn install_unserved_seams() { log::warn!( "[tinymemory:module] two host seams are unserved in module mode: scheduler_gate and \ shutdown are stubs that keep the unwired behaviour and report once when consulted. \ - Background-AI throttling and graceful queue-lock release are not honoured inside this \ - process" + Background-AI throttling, the \"Memory Tree off\" and \"signed out\" pauses on periodic \ + sync, and graceful queue-lock release are not honoured inside this process" ); } diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index 231ae94d..0411ee77 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -99,6 +99,11 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, OnceLock}; use tinybus::{Connection, Error as BusError, Result as BusResult}; +// The trait, not only its methods: `composio` is reached as a method on +// `EngineRuntimeConfig` in `composio_sync_can_run`, and without the trait in +// scope rustc points at the struct's `composio_mode` field instead. +use tinymemory_api::host::MemoryHostConfig; +use tinymemory_core::store::MemoryClientRef; /// The module refused its configuration or could not bring up a store. const SETUP_FAILED_ERROR: &str = "ai.tinyhumans.tinymemory.Error.SetupFailed"; @@ -190,59 +195,220 @@ async fn setup(connection: Connection, mut config: ModuleConfig) -> BusResult<() log::error!("[tinymemory:module] create memory store failed: {error}"); setup_error("create memory store") })?; + let client: MemoryClientRef = Arc::new(client); // After the store, never before: `queue::start` recovers stale locks as its // first act, which opens the queue database, and the factory above is what // creates the workspace it lives in. start_queue_pool(&config); - // ── The periodic sync loops are deliberately NOT started here ─────────── - // - // This is the obvious next line to write — the queue pool moved in here for - // exactly the reason the sync loops would, and `composio_host` and - // `config_loader` are now installed above, which is what a reader would - // check first. It does not work yet, and it would fail *quietly*, so the - // reasons are written down rather than left to be rediscovered. - // - // `tinymemory_core::sync::composio::start_periodic_sync` dispatches through - // `sync::pipelines::host::run_composio_connection_with_caps`, and three - // separate things in that path have no answer in this process: - // - // 1. **The pipeline reads credentials off the `Config`, not off the seam.** - // `composio_config` takes the direct-mode branch only when - // `config.composio().mode == "direct"` and otherwise needs - // `config.session_token()`. `EngineRuntimeConfig` answers - // `ComposioMode::default()` (mode `""`) and `Ok(None)`, so backend mode - // fails with "backend bearer token is not configured" and direct mode is - // never selected at all. `ComposioHost::api_key` cannot rescue this: the - // seam is consulted *inside* the direct branch that is not taken. The - // real fix is to route the pipeline's own HTTP client through - // `ComposioHost::execute`, which is a change to the engine's contract. - // - // 2. **`crate::global::client_if_ready()` is `None` here.** That is the - // first line of every pipeline run. This module builds its store through - // `create_memory_client_with_local_ai`, which does not touch the global - // slot, and calling `global::init` would build a *second* client via - // `MemoryClient::from_workspace_dir` — different embedding routes, a - // second ingestion worker over the same SQLite file. - // - // 3. **The cadence reads as "manual only".** - // `EngineRuntimeConfig::memory_sync_interval_secs()` is `Some(0)`, which - // the contract defines as manual-only, so both loops would skip every - // source on every tick. This is the one that would be invisible: no - // error, no warning, just a sync that never fires. See - // `config_loader`'s module docs for why the loader does not invent a - // different number. - // - // A fourth consequence is worth knowing even once those are fixed: this - // module's scheduler gate is a stub that always reads `Normal`, so a sync - // loop in here would not honour the "signed out" and "user disabled" pauses - // that `periodic_pause_reason` exists to apply. + // Also after the store, and for a second reason on top of that one: what is + // published is the client just built, and there is nothing to publish until + // it exists. The sync loops follow the bind rather than the other way round + // — every runner in `sync::pipelines::host` opens with + // `global::client_if_ready()`, so a loop started before this would fail + // every run. + if bind_memory_client(&config, &client) { + start_sync_loops(&config); + } - let provider = provider::provider(&config, Arc::new(client)); + let provider = provider::provider(&config, client); service::serve(&connection, Arc::new(provider), config).await } +/// Publish the store this process just built as the client for its workspace. +/// +/// # Why this is a `bind` and not `global::init` +/// +/// Everything in `tinymemory_core::sync` resolves its store through +/// `global::client_if_ready()`, which is `None` in this process: the module +/// builds its store through `create_memory_client_with_local_ai` — the only +/// entry point that takes this module's embedding routes, storage provider and +/// workspace — and that factory never touches the global slot. +/// +/// The obvious repair, `global::init(workspace)`, is the wrong one and quietly +/// so. It constructs a *second* `MemoryClient` over the same SQLite file, with +/// the host's default routes rather than this module's, and each client owns an +/// ingestion worker: duplicate graph extraction and duplicate embedding work +/// against one store, which `global`'s own comments call out as the hazard its +/// per-workspace cache exists to prevent. `global::bind` publishes the client +/// that already exists instead, into both the global slot and the per-workspace +/// cache, so all three resolution paths converge on it. +/// +/// # Which slot this writes +/// +/// This module's own. The `cdylib` carries its own compiled copy of +/// `tinymemory-core`, so the slot filled here is the static that *this +/// process's module-side* loops read through `client_if_ready`, and not the one +/// a host still booting an in-process engine fills with `global::init`. That is +/// what makes binding safe to do before that host's engine is deleted: this +/// cannot repoint the host's engine at this client, and the host's `init` +/// cannot make this bind refuse. +/// +/// The refusal below therefore means one specific thing — a second +/// `MemoryClient` was built for this workspace *inside this module* — which is +/// the hazard the whole function exists to keep from happening quietly. +/// +/// # Returns +/// +/// Whether the client is bound. A failure is reported and the caller starts no +/// sync loops: with no client resolvable, every run in both loops would fail on +/// its first line with "memory client is not ready" — a named cause, but a loop +/// that can only fail is not worth the ticks, the Composio list call every 20 +/// minutes, or the failed-sync audit rows it would append forever. +fn bind_memory_client(config: &ModuleConfig, client: &MemoryClientRef) -> bool { + match tinymemory_core::global::bind(config.workspace_dir.clone(), Arc::clone(client)) { + Ok(_) => true, + Err(error) => { + // The path in `error` stays in this module's log, like the factory + // failure above; nothing here crosses the bus. + log::error!( + "[tinymemory:module] could not publish the memory client for this workspace, so \ + periodic memory sync will not run in this process: {error}" + ); + false + } + } +} + +/// Start the engine's two periodic sync loops for this process. +/// +/// # Why the module has to own these +/// +/// The same reason [`start_queue_pool`] does. Both loops are engine code — +/// `tinymemory_core::sync::composio::periodic` and +/// `tinymemory_core::sync::workspace::periodic` — and until now the only calls +/// to them in any tree were the host's, made against the second, in-process +/// engine the host also booted. A host that deletes that engine, which is the +/// entire point of loading this module, is left with two loops it can no longer +/// start and a memory that stops updating: Composio connections stop pulling +/// mail, issues and documents, and registered repos, folders, RSS feeds and web +/// pages go stale. The sync layer reports "no connections", which is +/// indistinguishable from a user who has none. +/// +/// # The host must stop starting them in the same change +/// +/// Not "should" — this is the one part the module cannot guard. The `cdylib` +/// carries its own copy of `tinymemory-core`, so the `OnceLock` each loop +/// guards itself with is a *different* static from the host's: a host that +/// still calls `start_periodic_sync` while loading this module gets two pairs +/// of loops, neither of which can see the other, both walking the same source +/// registry into the same store. [`claim_sync_loops`] catches only the +/// in-process case. So the host's call site goes in the same change that +/// deletes the engine it was calling against. +/// +/// # What they do not get in module mode +/// +/// Stated rather than hidden, in the same terms [`start_queue_pool`] states its +/// own two: +/// +/// - **Neither loop honours the scheduler-gate pauses.** Both call +/// `periodic_pause_reason` as step 0 of every tick, precisely so a user who +/// switched Memory Tree off, or who is signed out, gets no background fetch. +/// This module serves no scheduler gate — see the section comment on +/// `host::install_unserved_seams` for why it cannot — and the stub in its +/// place always answers `Policy::Normal`, so `periodic_pause_reason` is always +/// `None` and both loops tick straight through both pauses. The per-source +/// `enabled` toggle still applies; the two *global* pauses do not. +/// - **Their resume wake never fires.** The stub's `resume_notify` hands back a +/// `Notify` nobody signals, so a user who re-enables sync waits out the +/// remaining 20-minute tick instead of syncing within seconds. That is the +/// benign half of the same gap. +/// +/// # Backend-mode Composio sync cannot run here, so it is not started +/// +/// `composio_config` takes its direct branch on `config.composio().mode` and +/// otherwise needs a backend session bearer, which this module holds no field +/// for and refuses to hold — see `ModuleConfig::composio_mode` and +/// `EngineRuntimeConfig::session_token` for that decision in full. Starting the +/// Composio loop under any other mode would list the user's connections every 20 +/// minutes and fail every due one with the same named cause, appending a failed +/// row to the sync audit each time, forever. So it is gated, and the gate says +/// so out loud once instead. +/// +/// The workspace loop is started either way: it drives repos, folders, RSS and +/// web pages through `sources::sync::sync_source`, which never touches Composio. +/// +/// Both the gate and the pipeline read the load-time snapshot, so a user who +/// switches Composio mode after this module loaded is not picked up until the +/// host reloads it — `config_loader`'s documented staleness, and not new here. +/// The gate itself is [`start_composio_periodic_sync`]. +fn start_sync_loops(config: &ModuleConfig) { + match claim_sync_loops(&config.workspace_dir) { + WorkspaceClaim::Start => { + // Warn, not debug: it is true on every boot in module mode, and a + // reader of the log should not have to know which seams are stubbed + // to find out that the pauses are not in effect. + log::warn!( + "[tinymemory:module] starting the periodic memory sync loops in this process. \ + They do not honour the scheduler gate — it is unserved here, so the \ + \"Memory Tree off\" and \"signed out\" pauses are ignored and a re-enable is \ + not woken early — though each source's own enabled toggle still applies" + ); + // Workspace sources first, because this one runs in every mode. + tinymemory_core::sync::workspace::start_workspace_periodic_sync(); + start_composio_periodic_sync(config); + } + WorkspaceClaim::AlreadyRunning => { + log::debug!( + "[tinymemory:module] the periodic memory sync loops for this workspace are \ + already running" + ); + } + WorkspaceClaim::Foreign => { + log::error!( + "[tinymemory:module] the periodic memory sync loops are already running for a \ + different workspace in this process, and both guard themselves process-wide, \ + so this store gets no periodic sync: Composio connections and registered \ + sources will not update. One module process serves one workspace" + ); + } + } +} + +/// Start the Composio half of [`start_sync_loops`], if this host's mode allows. +/// +/// Split out so the gate is one readable decision rather than a conditional +/// buried in a match arm, and so the refusal branch has somewhere to explain +/// itself. The decision itself is [`composio_sync_can_run`]. +fn start_composio_periodic_sync(config: &ModuleConfig) { + if composio_sync_can_run(config) { + tinymemory_core::sync::composio::start_periodic_sync(); + return; + } + + log::warn!( + "[tinymemory:module] periodic Composio sync is NOT started: this host did not resolve \ + Composio to direct mode, and backend mode needs a session bearer this module holds no \ + field for and will not carry. Composio-connected sources will not update in this \ + process until the sync client routes through `ComposioHost::execute`" + ); +} + +/// Whether the Composio pipelines can resolve a credential in this process. +/// +/// True only for direct mode, which is the whole of the gate: the other branch +/// of `sync::pipelines::host::composio_config` needs a backend session bearer, +/// and `EngineRuntimeConfig::session_token` refuses to answer one by design. +/// +/// Asked of the *same* `EngineRuntimeConfig` the loop's own ticks will be handed +/// and through the same `MemoryHostConfig::composio` accessor `composio_config` +/// reads, so the two cannot disagree about which host this is. What is left that +/// could drift is the comparison — this side calls `ComposioMode::is_direct`, +/// the pipeline inlines the same case-insensitive test against +/// `COMPOSIO_MODE_DIRECT` — so a host that spells its mode `"Direct"` is either +/// started and served or neither, never started and then failed on every tick. +/// +/// A predicate rather than a condition inside its one caller, for the reason +/// [`claim_workspace`] is one: this is the whole of what is worth asserting, and +/// asserting it through the caller would spawn a real 20-minute tick loop into +/// the test binary. +pub(crate) fn composio_sync_can_run(config: &ModuleConfig) -> bool { + tinymemory_tinycortex::engine::EngineRuntimeConfig::from(config) + .composio() + .is_direct() +} + /// The workspace whose queue this process's worker pool drains. /// /// The pool is bound to one workspace — every `queue::store` entry point @@ -254,40 +420,62 @@ async fn setup(connection: Connection, mut config: ModuleConfig) -> BusResult<() /// which workspace won makes that case loud instead of invisible. static QUEUE_POOL_WORKSPACE: OnceLock = OnceLock::new(); -/// What [`claim_queue_pool`] found when asked to start a pool. +/// The workspace whose periodic sync loops this process drives. +/// +/// A separate cell from [`QUEUE_POOL_WORKSPACE`] because they are separate +/// services that can each be claimed or not, but the trap is identical and so +/// is the reasoning: `start_periodic_sync` and `start_workspace_periodic_sync` +/// each guard themselves with a process-global `OnceLock<()>`, which makes a +/// second call a no-op that is indistinguishable from a first that worked, while +/// what each loop actually syncs is rooted at whatever workspace the installed +/// `config_loader` answers for. +static SYNC_LOOPS_WORKSPACE: OnceLock = OnceLock::new(); + +/// What a claim on one of this process's workspace-bound background services +/// found. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum QueuePoolClaim { - /// Nothing had claimed the pool; this caller starts it. +pub(crate) enum WorkspaceClaim { + /// Nothing had claimed the service; this caller starts it. Start, - /// A pool is already draining this workspace's queue, so there is nothing - /// to do and nothing wrong. - AlreadyDraining, - /// A pool is running, but rooted somewhere else. This store's queue has - /// nothing draining it and cannot be given a pool of its own. + /// It is already running for this workspace, so there is nothing to do and + /// nothing wrong. + AlreadyRunning, + /// It is running, but rooted somewhere else. This store cannot be given one + /// of its own, and goes without. Foreign, } -/// Decide whether this caller is the one that starts the pool. -/// -/// Split out from [`start_queue_pool`] so the decision can be asserted without -/// spawning four job workers and a daily scheduler into a test process, and -/// because `queue::start`'s own `Once` is not observable from here at all — a -/// second call to it is indistinguishable from a first that worked. -pub(crate) fn claim_queue_pool(workspace: &Path) -> QueuePoolClaim { - match QUEUE_POOL_WORKSPACE.set(workspace.to_path_buf()) { - Ok(()) => QueuePoolClaim::Start, +/// Decide whether this caller is the one that starts `cell`'s service. +/// +/// Split out from the two `start_*` functions so each decision can be asserted +/// without spawning real workers and tick loops into a test process, and because +/// the guards inside `tinymemory-core` are not observable from here at all — a +/// second call to any of them is indistinguishable from a first that worked. +fn claim_workspace(cell: &OnceLock, workspace: &Path) -> WorkspaceClaim { + match cell.set(workspace.to_path_buf()) { + Ok(()) => WorkspaceClaim::Start, // `set` hands the rejected value back, so the comparison needs no // second read and cannot race with a concurrent claim. Err(rejected) => { - if QUEUE_POOL_WORKSPACE.get() == Some(&rejected) { - QueuePoolClaim::AlreadyDraining + if cell.get() == Some(&rejected) { + WorkspaceClaim::AlreadyRunning } else { - QueuePoolClaim::Foreign + WorkspaceClaim::Foreign } } } } +/// Claim the queue worker pool for `workspace`. See [`start_queue_pool`]. +pub(crate) fn claim_queue_pool(workspace: &Path) -> WorkspaceClaim { + claim_workspace(&QUEUE_POOL_WORKSPACE, workspace) +} + +/// Claim the periodic sync loops for `workspace`. See [`start_sync_loops`]. +pub(crate) fn claim_sync_loops(workspace: &Path) -> WorkspaceClaim { + claim_workspace(&SYNC_LOOPS_WORKSPACE, workspace) +} + /// Start the engine's queue worker pool for this process. /// /// # Why the module has to own this @@ -329,7 +517,7 @@ pub(crate) fn claim_queue_pool(workspace: &Path) -> QueuePoolClaim { /// once per process the first time the pool consults them. fn start_queue_pool(config: &ModuleConfig) { match claim_queue_pool(&config.workspace_dir) { - QueuePoolClaim::Start => { + WorkspaceClaim::Start => { // Warn, not debug: it is true on every boot in module mode, and a // reader of the log should not have to know which seams are stubbed // to find out that the throttle is not in effect. @@ -344,12 +532,12 @@ fn start_queue_pool(config: &ModuleConfig) { tinymemory_tinycortex::engine::EngineRuntimeConfig::from(config), )); } - QueuePoolClaim::AlreadyDraining => { + WorkspaceClaim::AlreadyRunning => { log::debug!( "[tinymemory:module] the queue worker pool for this workspace is already running" ); } - QueuePoolClaim::Foreign => { + WorkspaceClaim::Foreign => { log::error!( "[tinymemory:module] a queue worker pool is already running for a different \ workspace in this process, and `queue::start` is guarded process-wide, so the \ @@ -421,6 +609,13 @@ mod exports { // is what drives a job's own outbound embed while the rest are busy. At // two, a draining queue would starve inbound dispatch and the module // would stop answering recalls until the queue emptied. + // + // The two periodic sync loops `setup` also starts do not move the + // number. They sleep on a 20-minute `interval` and yield across every + // fetch, so they hold no worker between ticks; the one moment they do is + // `BusComposioHost::probe`, which blocks its caller for one bus round + // trip and is bounded at twice per tick — see the note on `probe` for + // why that bridge blocks at all. worker_threads = 8, provides = ["ai.tinyhumans.tinymemory.Memory"], methods = [ diff --git a/crates/tinymemory-module/src/provider.rs b/crates/tinymemory-module/src/provider.rs index 278a0002..a48734b1 100644 --- a/crates/tinymemory-module/src/provider.rs +++ b/crates/tinymemory-module/src/provider.rs @@ -27,6 +27,13 @@ impl From<&ModuleConfig> for EngineRuntimeConfig { default_temperature: config.default_temperature, output_language: config.output_language.clone(), memory_sources: config.memory_sources.clone(), + // The three the periodic sync loops read. They cross as data rather + // than being answered by the engine config's own constants, because + // the constants were `Some(0)` — manual-only — and an empty Composio + // mode, and both of those skip work rather than fail it. + memory_sync_interval_secs: config.memory_sync_interval_secs, + composio_mode: config.composio_mode.clone(), + composio_entity_id: config.composio_entity_id.clone(), } } } diff --git a/crates/tinymemory-module/src/service/test.rs b/crates/tinymemory-module/src/service/test.rs index 62b86764..14e73365 100644 --- a/crates/tinymemory-module/src/service/test.rs +++ b/crates/tinymemory-module/src/service/test.rs @@ -472,22 +472,105 @@ fn the_queue_pool_is_claimed_once_and_a_foreign_workspace_is_refused() { assert_eq!( crate::claim_queue_pool(workspace), - crate::QueuePoolClaim::Start, + crate::WorkspaceClaim::Start, "the first claim must be the one that starts the pool" ); assert_eq!( crate::claim_queue_pool(workspace), - crate::QueuePoolClaim::AlreadyDraining, + crate::WorkspaceClaim::AlreadyRunning, "a second claim for the same workspace must not start a second pool" ); assert_eq!( crate::claim_queue_pool(elsewhere), - crate::QueuePoolClaim::Foreign, + crate::WorkspaceClaim::Foreign, "a claim for another workspace must be named, not silently swallowed — \ `queue::start` would no-op and that store's queue would never drain" ); } +/// The periodic sync loops are claimed the same way, and for the same reason. +/// +/// Asserted through `claim_sync_loops` rather than `start_sync_loops` for the +/// reason above and one more: starting them for real spawns two 20-minute tick +/// loops that reload config and walk the source registry for the rest of the +/// test binary's life. +/// +/// This also pins that the two services claim *independent* cells, without a +/// fourth test that would have to assume an execution order. The workspace here +/// differs from the queue pool's, so a single shared cell would make whichever +/// of these two tests ran second read `Foreign` where it expects `Start`. +/// +/// The `Foreign` outcome is what a second module setup in one process would hit. +/// `claim_process_setup` already refuses that, so this is a second guard on a +/// case the first one covers — kept because the cost is one `OnceLock` and the +/// failure it guards is a store that silently never syncs. +#[test] +fn the_sync_loops_are_claimed_once_and_a_foreign_workspace_is_refused() { + let workspace = std::path::Path::new("/tinymemory-module/sync-loops-claim"); + let elsewhere = std::path::Path::new("/tinymemory-module/sync-loops-elsewhere"); + + assert_eq!( + crate::claim_sync_loops(workspace), + crate::WorkspaceClaim::Start, + "the first claim must be the one that starts the loops" + ); + assert_eq!( + crate::claim_sync_loops(workspace), + crate::WorkspaceClaim::AlreadyRunning, + "a second claim for the same workspace must not start a second pair" + ); + assert_eq!( + crate::claim_sync_loops(elsewhere), + crate::WorkspaceClaim::Foreign, + "a claim for another workspace must be named, not silently swallowed — \ + both loops guard themselves process-wide and that store would never sync" + ); +} + +/// The Composio gate answers for exactly the branch the pipeline would take. +/// +/// Worth pinning because the two ways it can be wrong are both quiet. A gate +/// that started the loop in backend mode would list the user's connections +/// every 20 minutes and fail every due one on `session_token`'s refusal, +/// appending a failed row to the sync audit each time; a gate that refused +/// direct mode would leave a host that could sync perfectly well with Composio +/// sources that simply stop updating, and one line at boot to explain it. +/// +/// Asserted through `composio_sync_can_run` rather than +/// `start_composio_periodic_sync` for the reason the claim tests above give: +/// the decision is the whole of what is worth checking, and the call after it +/// spawns a real 20-minute tick loop for the life of the test binary. +#[test] +fn composio_periodic_sync_starts_only_when_the_host_resolved_direct_mode() { + let mut config = test_config(std::path::Path::new("/tinymemory-module/composio-gate")); + + assert!( + !crate::composio_sync_can_run(&config), + "a host that states no mode is not direct — and has no bearer either" + ); + + config.composio_mode = tinymemory_api::host::COMPOSIO_MODE_BACKEND.to_string(); + assert!( + !crate::composio_sync_can_run(&config), + "backend mode needs a session bearer this module refuses to hold" + ); + + config.composio_mode = tinymemory_api::host::COMPOSIO_MODE_DIRECT.to_string(); + assert!( + crate::composio_sync_can_run(&config), + "direct mode is the one branch that resolves its credential in here" + ); + + // The pipeline's own branch test is case-insensitive. If the gate were not, + // this host would be started and would then fail every tick — the exact + // shape the gate exists to prevent. + config.composio_mode = "Direct".to_string(); + assert!( + crate::composio_sync_can_run(&config), + "the gate must match `composio_config` on case, or it starts a loop that cannot work" + ); +} + /// A second store opens normally, and needs no pool of its own to do it. /// /// The pairing with the test above is the point. `queue::start` is guarded by a diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index 5ebc3982..db4e404b 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -93,8 +93,52 @@ pub struct EngineRuntimeConfig { pub output_language: Option, /// Opaque source configuration, passed through verbatim. pub memory_sources: serde_json::Value, + /// The user's global memory-sync cadence in seconds, as the host resolved + /// it. + /// + /// `None` is "no explicit choice" and callers fall back to + /// [`DEFAULT_MEMORY_SYNC_INTERVAL_SECS`](tinymemory_api::host::DEFAULT_MEMORY_SYNC_INTERVAL_SECS); + /// `Some(0)` is manual-only. + /// + /// Carried rather than answered with a constant because both periodic sync + /// loops read it as their gate, and the constant they used to get was + /// `Some(0)` — manual-only, which skips every source on every tick with no + /// error, no warning and nothing in the log. A cadence a host cannot state + /// is the one field where a wrong constant is invisible. + pub memory_sync_interval_secs: Option, + /// Composio routing mode: + /// [`COMPOSIO_MODE_BACKEND`](tinymemory_api::host::COMPOSIO_MODE_BACKEND) or + /// [`COMPOSIO_MODE_DIRECT`](tinymemory_api::host::COMPOSIO_MODE_DIRECT). + /// + /// Empty means the host stated no mode, and reads as "not direct" — the same + /// answer backend mode gets, which is what an unset Composio integration + /// should look like. + pub composio_mode: String, + /// The Composio entity the host authenticates as. + /// + /// An identifier, not a credential: it selects whose connected accounts a + /// direct-mode call addresses. Empty is sent as no entity at all rather than + /// as an empty one — see `ComposioClient::execute_direct`. + pub composio_entity_id: String, } +/// Why a module-side engine configuration can never answer a backend session +/// token, said once so every path that hits it reports the same cause. +/// +/// The bare "not configured" this used to produce is the wrong story. It reads +/// as "the user is signed out", which a reader then tries to fix by signing in; +/// the truth is structural and no sign-in changes it. This configuration is +/// built from a `ModuleConfig`, which has no field for a bearer and deliberately +/// never will — and even if it did, a load-time snapshot could not follow a +/// token the host refreshes mid-session, so the module would authenticate with +/// an expired bearer until the next module load. +const NO_BACKEND_SESSION: &str = + "this engine configuration carries no backend session token, by design: a loaded memory \ + module holds no credentials, and a bearer the host refreshes mid-session could not be \ + answered from a load-time snapshot anyway. Backend-mode Composio memory sync therefore \ + cannot run inside the module — only direct mode resolves here — and closing that means \ + routing the sync client through `ComposioHost::execute`, not handing the module a token"; + #[async_trait] impl MemoryHostConfig for EngineRuntimeConfig { fn workspace_dir(&self) -> &PathBuf { @@ -155,8 +199,17 @@ impl MemoryHostConfig for EngineRuntimeConfig { fn effective_backend_api_url(&self) -> String { String::new() } + /// Always the named refusal, never `Ok(None)`. + /// + /// The contract distinguishes the two: `Ok(None)` is "read fine, not signed + /// in", `Err` is "could not be read". This configuration is neither — there + /// is no store to read, and there never will be. `Ok(None)` would send + /// `sync::pipelines::host::composio_config` down its backend branch to fail + /// with "OpenHuman backend bearer token is not configured", which points a + /// reader at a sign-in that would not help. See `NO_BACKEND_SESSION`, + /// which is where the whole reason is written down. fn session_token(&self) -> Result, String> { - Ok(None) + Err(NO_BACKEND_SESSION.to_string()) } fn default_model(&self) -> Option<&str> { self.default_model.as_deref() @@ -168,7 +221,7 @@ impl MemoryHostConfig for EngineRuntimeConfig { self.output_language.as_deref() } fn memory_sync_interval_secs(&self) -> Option { - Some(0) + self.memory_sync_interval_secs } fn onboarding_completed(&self) -> bool { true @@ -176,8 +229,28 @@ impl MemoryHostConfig for EngineRuntimeConfig { fn secrets_encrypt(&self) -> bool { false } + /// The routing mode and entity the host resolved, and nothing else. + /// + /// Rebuilt from two scalar fields rather than stored as a [`ComposioMode`] + /// so the two remaining members can only ever hold what this type promises: + /// + /// - `api_key` stays `None` because the direct-mode key is not configuration + /// here. `composio_config` asks the `ComposioHost` seam for it first and + /// falls back to this field second, so leaving it empty routes the key + /// through the seam — fetched for the duration of one call, held in no + /// field, which is the property that lets this struct call itself + /// credential-free. + /// - `triage_disabled` stays `false` because nothing in the memory layer + /// reads it; it gates the host's LLM triage of Composio *triggers*, a path + /// that never enters this crate. Carrying a value nobody reads would + /// invite a reader to believe it does something here. fn composio(&self) -> ComposioMode { - ComposioMode::default() + ComposioMode { + mode: self.composio_mode.clone(), + entity_id: self.composio_entity_id.clone(), + api_key: None, + triage_disabled: false, + } } fn memory_sources_json(&self) -> anyhow::Result { Ok(self.memory_sources.clone()) diff --git a/crates/tinymemory-tinycortex/src/engine/test.rs b/crates/tinymemory-tinycortex/src/engine/test.rs index c62aed45..78e1b0b7 100644 --- a/crates/tinymemory-tinycortex/src/engine/test.rs +++ b/crates/tinymemory-tinycortex/src/engine/test.rs @@ -64,6 +64,9 @@ fn runtime_config() -> EngineRuntimeConfig { default_temperature: 0.3, output_language: Some("en".to_string()), memory_sources: serde_json::json!([{"id": "source-1"}]), + memory_sync_interval_secs: Some(14_400), + composio_mode: tinymemory_api::host::COMPOSIO_MODE_DIRECT.to_string(), + composio_entity_id: "entity-1".to_string(), } } @@ -182,11 +185,20 @@ async fn runtime_config_routes_models_and_round_trips_source_configuration() { assert!(config.to_arc().as_any().is::()); assert_eq!(config.api_url(), None); assert!(config.effective_backend_api_url().is_empty()); - assert_eq!(config.session_token().expect("session token"), None); - assert_eq!(config.memory_sync_interval_secs(), Some(0)); + // Not `Ok(None)`: see the accessor's own doc. `Ok(None)` reads as "signed + // out" and sends a reader after a sign-in that cannot help. + let session = config + .session_token() + .expect_err("a module-side config must refuse rather than report signed-out"); + assert!(session.contains("no backend session token"), "{session}"); + assert_eq!(config.memory_sync_interval_secs(), Some(14_400)); assert!(config.onboarding_completed()); assert!(!config.secrets_encrypt()); - assert!(!config.composio().is_direct()); + assert!(config.composio().is_direct()); + assert_eq!(config.composio().entity_id, "entity-1"); + // The key never rides in the config — `composio_config` resolves it through + // the `ComposioHost` seam, per call. + assert!(config.composio().api_key.is_none()); assert_eq!(config.composio_source_caps_migration_version(), 0); config.set_composio_source_caps_migration_version(2); config.apply_env_overrides(); @@ -208,6 +220,72 @@ async fn runtime_config_routes_models_and_round_trips_source_configuration() { .expect("the in-memory adapter save is a no-op"); } +/// The cadence is answered from the field, including the two values that mean +/// something other than a number of seconds. +/// +/// This is the blocker the periodic loops could not see past: the accessor used +/// to answer the constant `Some(0)`, which +/// `sync::composio::periodic::effective_interval_secs` maps to `None` — the +/// contract's manual-only — so every source was skipped on every tick with +/// nothing logged. A cadence that reads as a *setting* has to come from the +/// host, and the only wrong answer that is silent is this one. +#[test] +fn the_sync_cadence_is_answered_from_the_host_and_not_from_a_constant() { + let mut config = runtime_config(); + + // No explicit user choice: callers fall back to the 24h default. + config.memory_sync_interval_secs = None; + assert_eq!(config.memory_sync_interval_secs(), None); + + // "Manual only", which the host can now actually express. + config.memory_sync_interval_secs = Some(0); + assert_eq!(config.memory_sync_interval_secs(), Some(0)); + + config.memory_sync_interval_secs = Some(3_600); + assert_eq!(config.memory_sync_interval_secs(), Some(3_600)); +} + +/// A host that states no Composio mode reads as "not direct", which is what an +/// unconfigured integration should look like — and is exactly what the accessor +/// answered before the field existed, so an older host's behaviour is unchanged. +#[test] +fn an_unstated_composio_mode_is_not_direct() { + let config = EngineRuntimeConfig { + composio_mode: String::new(), + composio_entity_id: String::new(), + ..runtime_config() + }; + + assert!(!config.composio().is_direct()); + assert!(config.composio().entity_id.is_empty()); +} + +/// Backend mode fails with the structural cause, not with a sign-in prompt. +/// +/// `composio_config` reaches `session_token` only on its backend branch, so this +/// is the message a backend-mode Composio sync inside a module actually +/// produces. It has to say *why* — no sign-in fixes a config that has no field +/// for a bearer. +#[test] +fn the_backend_branch_names_why_a_module_cannot_serve_it() { + let config = EngineRuntimeConfig { + composio_mode: tinymemory_api::host::COMPOSIO_MODE_BACKEND.to_string(), + ..runtime_config() + }; + + let error = config + .session_token() + .expect_err("backend mode must fail, and say why"); + + assert!(error.contains("no backend session token"), "{error}"); + assert!( + error.contains("ComposioHost::execute"), + "the message must name what would close the gap: {error}" + ); + // The old wording sent readers to a sign-in that cannot help. + assert!(!error.contains("not configured"), "{error}"); +} + #[test] fn people_profile_and_scope_boundary_conversions_are_total_and_fail_closed() { use tinymemory_api::provider::types::SourceScope; diff --git a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs index e7f6cd0c..2b276062 100644 --- a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs +++ b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs @@ -112,6 +112,13 @@ fn provider_config( default_temperature: 0.2, output_language: None, memory_sources, + // No cadence and no Composio mode: the conformance suite drives the + // provider directly and starts no periodic loop, so the values that + // matter to those loops are left at what a host that states nothing + // sends. + memory_sync_interval_secs: None, + composio_mode: String::new(), + composio_entity_id: String::new(), } } diff --git a/docs/specs/tinybus-module.md b/docs/specs/tinybus-module.md index e6483430..4117ccc5 100644 --- a/docs/specs/tinybus-module.md +++ b/docs/specs/tinybus-module.md @@ -222,6 +222,59 @@ bind that driver directly. The general lesson: **"carried verbatim" carries credentials verbatim too.** +### Three fields the periodic sync loops need + +`memory_sync_interval_secs`, `composio_mode` and `composio_entity_id` are the +module's answer to settings the engine used to read off a host `Config` it no +longer has. All three are optional on the wire, like every other field. + +**The cadence is the one that failed silently.** `EngineRuntimeConfig` answered +the constant `Some(0)`, which the contract defines as *manual only*, so both +loops skipped every source on every tick — no error, no warning, nothing in the +log. It now answers the host's value, and an absent field defaults to `None` +("the user chose nothing", so the 24h fallback) rather than to `Some(0)`. The +two are not symmetrical: an over-sync is bounded and a user can see it, a +no-sync is invisible by construction. Host and module are separately released, +so whatever the default says is what an older host silently means. + +**The Composio pair is routing, not access.** The mode picks which branch +`sync::pipelines::host::composio_config` takes, and the entity says whose +connected accounts a call addresses; neither authorises anything. The +direct-mode API key still does not travel — it is fetched from the host per call +— and there is no field for a backend session bearer, which is the whole reason +only direct mode can run in here. + +## The two periodic sync loops, and what they lose + +`setup` starts `sync::workspace::start_workspace_periodic_sync`, and in direct +mode `sync::composio::start_periodic_sync`, for the reason it starts the queue +worker pool: a host that deletes its in-process engine can start neither, and a +memory that stops updating reports "no connections" — indistinguishable from a +user who has none. + +Three things had to become true first, and all three were false. The cadence is +one (above). The second is that `composio_config`'s direct branch was never +selected, because `EngineRuntimeConfig` answered an empty mode; `ComposioHost` +cannot rescue that, because its key is consulted *inside* the branch not taken. +The third is that `global::client_if_ready()` — the first line of every pipeline +run — was `None`, because this module builds its store through +`store::factories`, which never touches the global slot. + +The third is closed by `global::bind`, which publishes the **already-built** +client into the global slot *and* the per-workspace cache, so all three +resolution paths converge on it. `global::init` would have built a second +`MemoryClient` over the same SQLite file — two ingestion workers, duplicate +graph extraction, duplicate embedding — which is why `bind` refuses a different +client for a workspace rather than quietly absorbing it. + +**What the loops lose here.** The scheduler gate is a stub that always answers +`Normal`, so neither honours `periodic_pause_reason`'s two pauses — "Memory Tree +off" and "signed out" — and re-enabling sync no longer wakes them early instead +of waiting out the tick. Each source's own `enabled` toggle still applies. +**Backend-mode Composio sync is not started at all**, and says so once at boot, +rather than listing the user's connections every 20 minutes and failing every +due one forever. + ## Two operational constraints **Two worker threads, not one.** A recall that triggers an embed makes an From 841b318098b79e97aa40bd4273908be19e390184 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 25 Aug 2026 12:51:00 +0530 Subject: [PATCH 2/2] Format the bind tests to rustfmt's line breaking --- crates/tinymemory-core/src/global_tests.rs | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/crates/tinymemory-core/src/global_tests.rs b/crates/tinymemory-core/src/global_tests.rs index 680c99fe..75b6fdc0 100644 --- a/crates/tinymemory-core/src/global_tests.rs +++ b/crates/tinymemory-core/src/global_tests.rs @@ -131,12 +131,18 @@ async fn bind_publishes_a_caller_built_client_to_both_resolution_paths() { 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(&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)); + assert!(Arc::ptr_eq( + &client_for_workspace(&workspace).unwrap(), + &client + )); } /// Re-binding the same client is what a retried setup produces, and must not @@ -179,7 +185,10 @@ async fn binding_a_different_client_for_one_workspace_is_refused() { // 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)); + assert!(Arc::ptr_eq( + &client_for_workspace(&workspace).unwrap(), + &first + )); } /// A bind for another workspace is the active-user-switch shape `init` already