diff --git a/crates/tinymemory-core/src/global.rs b/crates/tinymemory-core/src/global.rs index 99fd1ea..012333e 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 9797979..75b6fdc 100644 --- a/crates/tinymemory-core/src/global_tests.rs +++ b/crates/tinymemory-core/src/global_tests.rs @@ -118,6 +118,119 @@ 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 46b7066..0dd8b54 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 da3cd8f..4506648 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 6c646c3..9c43c9b 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 c8e630b..e74cbf0 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 62715f9..74130f9 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 231ae94..0411ee7 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 278a000..a48734b 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 62b8676..14e7336 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 5ebc398..db4e404 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 c62aed4..78e1b0b 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 e7f6cd0..2b27606 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 e648343..4117ccc 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