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,