Add a member for syncing one memory source, whatever kind it is

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<SyncRunOutcome, MemoryError> {
+        let _ = source_id;
+        Err(MemoryError::unsupported(Capability::SourceSync))
+    }
+
     async fn source_sync_state(
         &self,
         toolkit: &str,

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.

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,

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",

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<SourceSyncState>
 //! 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<SyncRunOutcome> {
+        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

crates/tinymemory-module/tests/module_e2e.rs

@@ -694,6 +694,7 @@ const EXPECTED_METHODS: &[&str] = &[
     "FlushSourceTree",
     "Diagnose",
     "RunConnectionSync",
+    "RunSourceSync",
     "SourceSyncState",
     "SyncAuditLog",
     "EstimateSyncCostUsd",

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<SyncRunOutcome, MemoryError> {
+        // 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,