From a3fec1c51ebadec579d4c359475b188428c7dfc8 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 25 Aug 2026 21:46:56 +0530 Subject: [PATCH] Add a member for syncing one memory source, whatever kind it is MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `MemorySourceSync` could run a Composio connection and nothing else. A host with a folder, repository, RSS or web-page source and a "sync now" button had no member to call, so those paths stayed on the engine — and the engine function behind them opens with `global::client_if_ready()`. That made this the last thing standing between OpenHuman and deleting its in-process engine. Without the member, a host that stops booting one loses source sync for EVERY kind, and loses it quietly: the failure is "memory client is not ready", raised inside a spawned task that logs rather than propagates, so neither a compile nor a green test suite sees it. ## A source id, not a source The driver already reads the source registry — it has to, because the per-source budgets the pipeline applies live there. Passing the whole entry would put a second copy of those caps on the wire and invite the two to disagree about limits that cost money when they are wrong. ## NotFound is not an empty sync An unregistered id answers `NotFound` rather than a successful run that moved nothing. A caller retrying a source that was deleted underneath it should learn that, and the two are indistinguishable if both report zero records. Registered in all four places in declaration order, so the drift assertion holds: the interface impl, the module manifest, `METHODS` with its length and the members count in the bus crate's docs, and `EXPECTED_METHODS` in the loader E2E. --- crates/tinymemory-api/src/provider/sync.rs | 41 +++++++++++++++++++ crates/tinymemory-bus/src/lib.rs | 2 +- crates/tinymemory-bus/src/names.rs | 7 +++- crates/tinymemory-module/src/lib.rs | 1 + crates/tinymemory-module/src/service/mod.rs | 14 +++++++ crates/tinymemory-module/tests/module_e2e.rs | 1 + .../tinymemory-tinycortex/src/engine/mod.rs | 35 ++++++++++++++++ 7 files changed, 99 insertions(+), 2 deletions(-) diff --git a/crates/tinymemory-api/src/provider/sync.rs b/crates/tinymemory-api/src/provider/sync.rs index 144353d..185b26f 100644 --- a/crates/tinymemory-api/src/provider/sync.rs +++ b/crates/tinymemory-api/src/provider/sync.rs @@ -43,6 +43,7 @@ use async_trait::async_trait; +use crate::capabilities::Capability; use crate::error::MemoryError; // The value types this family exchanges. They are defined in `tinymemory-bus` @@ -114,6 +115,46 @@ pub trait MemorySourceSync: Send + Sync { /// # Errors /// /// Backend failures only. + /// Run one configured memory source through its pipeline, whatever kind it + /// is — a folder, a repository, an RSS feed, a web page, or a Composio + /// connection. + /// + /// # Why this exists beside [`Self::run_connection_sync`] + /// + /// That member is Composio-shaped: it takes a toolkit and a connection id, + /// which the other source kinds do not have. A host with a folder source + /// and a "sync now" button had nothing to call, and the engine function + /// behind it reaches a process-global memory client — so a host that stopped + /// embedding an engine lost source sync entirely, for every kind, with the + /// failure landing as "memory client is not ready" inside a spawned task. + /// + /// # A source id, not a source + /// + /// The driver already reads the source registry — it has to, to know the + /// per-source budgets the pipeline applies — so passing the whole entry + /// would put a second copy on the wire and invite the two to disagree about + /// caps that cost money when they are wrong. The id is the smaller and the + /// more honest argument. + /// + /// # This is the manual path, and it is not idempotent + /// + /// Same contract as [`Self::run_connection_sync`]: calling it twice runs the + /// pipeline twice, the cursor making the second run cheap rather than free, + /// and both runs append an audit row. + /// + /// # Errors + /// + /// [`MemoryError::NotFound`] when no source is registered under `source_id` + /// — deliberately distinct from a sync that ran and found nothing, because + /// a caller retrying a deleted source should learn that rather than see an + /// empty success. [`MemoryError::Unsupported`] from a driver that serves + /// this family but not this member. Otherwise the pipeline's own failure, + /// with whatever usage it incurred named in the message. + async fn run_source_sync(&self, source_id: &str) -> Result { + let _ = source_id; + Err(MemoryError::unsupported(Capability::SourceSync)) + } + async fn source_sync_state( &self, toolkit: &str, diff --git a/crates/tinymemory-bus/src/lib.rs b/crates/tinymemory-bus/src/lib.rs index 4e743ac..983abcd 100644 --- a/crates/tinymemory-bus/src/lib.rs +++ b/crates/tinymemory-bus/src/lib.rs @@ -2,7 +2,7 @@ //! the members that carry them. //! //! TinyMemory ships as a loadable `TinyBus` module: `crates/tinymemory-module` -//! exports one object with 120 members on it, built as a `cdylib`. A host that +//! exports one object with 121 members on it, built as a `cdylib`. A host that //! loads it — OpenHuman — can call into it but cannot `use` anything out of it, //! so the payload vocabulary has to be published as an ordinary library. This //! is that library. diff --git a/crates/tinymemory-bus/src/names.rs b/crates/tinymemory-bus/src/names.rs index 79fd7f6..cb57347 100644 --- a/crates/tinymemory-bus/src/names.rs +++ b/crates/tinymemory-bus/src/names.rs @@ -290,6 +290,10 @@ pub mod methods { // and what past runs cost. /// `RunConnectionSync` — run one connection's sync now. pub const RUN_CONNECTION_SYNC: &str = "RunConnectionSync"; + + /// `RunSourceSync` — run one configured memory source's sync now, + /// whatever kind it is. + pub const RUN_SOURCE_SYNC: &str = "RunSourceSync"; /// `SourceSyncState` — the persisted cursor and budget for one connection. pub const SOURCE_SYNC_STATE: &str = "SourceSyncState"; /// `SyncAuditLog` — past sync runs, newest first. @@ -315,7 +319,7 @@ pub mod methods { /// The order matters: `tinybus`'s `Interface::members()` returns declaration /// order, and the module compares the two sequences directly rather than as /// sets, so a reordering is caught alongside an addition or a removal. -pub const METHODS: [&str; 120] = [ +pub const METHODS: [&str; 121] = [ methods::DRIVER_ID, methods::CAPABILITIES, methods::HEALTH, @@ -428,6 +432,7 @@ pub const METHODS: [&str; 120] = [ methods::FLUSH_SOURCE_TREE, methods::DIAGNOSE, methods::RUN_CONNECTION_SYNC, + methods::RUN_SOURCE_SYNC, methods::SOURCE_SYNC_STATE, methods::SYNC_AUDIT_LOG, methods::ESTIMATE_SYNC_COST_USD, diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index 6744345..f430a7c 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -762,6 +762,7 @@ mod exports { // Source sync this process runs itself. The periodic loops already // live here; these are the on-demand half plus what past runs cost. "RunConnectionSync", + "RunSourceSync", "SourceSyncState", "SyncAuditLog", "EstimateSyncCostUsd", diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index 5ae63de..a92c33a 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -65,6 +65,7 @@ //! Diagnose() -> Diagnosis //! //! RunConnectionSync(toolkit, connection_id) -> SyncRunOutcome +//! RunSourceSync(source_id) -> SyncRunOutcome //! SourceSyncState(toolkit, connection_id) -> Option //! SyncAuditLog(limit) -> [SyncAuditEntry] //! EstimateSyncCostUsd(input_tokens, output_tokens) -> f64 @@ -1736,6 +1737,19 @@ impl MemoryService { .map_err(|error| into_bus_error(&error)) } + /// Run one configured memory source through its pipeline, whatever kind. + /// + /// Beside `RunConnectionSync` rather than replacing it: that member is + /// Composio-shaped and this one covers every kind, including the folder, + /// repository, feed and web-page sources that have no toolkit or connection + /// id to name. + async fn run_source_sync(&self, source_id: String) -> BusResult { + require_family!(self, as_source_sync, Capability::SourceSync) + .run_source_sync(&source_id) + .await + .map_err(|error| into_bus_error(&error)) + } + /// The persisted cursor, dedup and budget state for one connection. /// /// `None` is "never synced", which is a state and not an error — a status diff --git a/crates/tinymemory-module/tests/module_e2e.rs b/crates/tinymemory-module/tests/module_e2e.rs index 99837fb..f9f8456 100644 --- a/crates/tinymemory-module/tests/module_e2e.rs +++ b/crates/tinymemory-module/tests/module_e2e.rs @@ -694,6 +694,7 @@ const EXPECTED_METHODS: &[&str] = &[ "FlushSourceTree", "Diagnose", "RunConnectionSync", + "RunSourceSync", "SourceSyncState", "SyncAuditLog", "EstimateSyncCostUsd", diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index 7f9a55a..895fb26 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -2467,6 +2467,41 @@ impl MemorySourceSync for TinycortexProvider { }) } + async fn run_source_sync(&self, source_id: &str) -> Result { + // Resolved here rather than passed in: the registry is the driver's own + // and carries the per-source budgets the pipeline applies, so a caller + // supplying the entry would put a second copy of those caps on the wire. + let source = tinymemory_core::sources::registry::get_source_in(&self.config, source_id) + .map_err(|error| MemoryError::Other(anyhow::anyhow!("read source registry: {error}")))? + .ok_or_else(|| { + // Distinct from an empty sync on purpose: a caller retrying a + // source that was deleted underneath it should learn that, not + // read a successful run that moved nothing. + MemoryError::NotFound(format!("no memory source registered as {source_id}")) + })?; + + let outcome = tinymemory_core::engine::run_source_pipeline(&source, &self.config) + .await + .map_err(|failure| { + // Same shape as `run_connection_sync`: the usage travels in the + // message because an error path has nowhere structured to put it. + MemoryError::Other(anyhow::anyhow!( + "sync source {source_id}: {} (actions_called={}, provider_cost_usd={})", + failure.message, + failure.actions_called, + failure.provider_cost_usd + )) + })?; + + Ok(SyncRunOutcome { + records_ingested: outcome.records_ingested, + more_pending: outcome.more_pending, + actions_called: outcome.actions_called, + provider_cost_usd: outcome.provider_cost_usd, + note: outcome.note, + }) + } + async fn source_sync_state( &self, toolkit: &str,