Merge pull request #105 from YellowSnnowmann/feat/5560-connection-created-hook

Serve the Composio connection bootstrap over the bus

Author: YellowSnnowmann

crates/tinymemory-api/src/null_tests.rs

@@ -514,6 +514,10 @@ fn every_optional_method_fails_with_its_advertised_family_name() {
         block_on(driver.run_connection_sync("gmail", "conn-1")),
         Capability::SourceSync,
     );
+    assert_unsupported(
+        block_on(driver.bootstrap_connection("gmail", "conn-1")),
+        Capability::SourceSync,
+    );
     assert_unsupported(
         block_on(driver.source_sync_state("gmail", "conn-1")),
         Capability::SourceSync,

crates/tinymemory-api/src/provider/sync.rs

@@ -155,6 +155,50 @@ pub trait MemorySourceSync: Send + Sync {
         Err(MemoryError::unsupported(Capability::SourceSync))
     }
 
+    /// Run one connection's first-time bootstrap.
+    ///
+    /// What a host's "this connection was just authorised" event reaches. The
+    /// driver resolves the provider for `toolkit` and runs its bootstrap: at
+    /// minimum fetching and persisting the account profile, and for providers
+    /// that override it, registering triggers or seeding labels as well.
+    ///
+    /// # Why this is not part of [`Self::run_connection_sync`]
+    ///
+    /// A sync moves items and is expected to run many times; a bootstrap
+    /// establishes the things a sync then assumes and is expected to run once.
+    /// Folding them together would either re-register triggers on every sync
+    /// or leave a connection whose first sync silently has no profile behind
+    /// it — and the two also fail differently, which is the more practical
+    /// reason: a bootstrap that fails should not stop items from syncing, and
+    /// a caller can only make that choice if it can tell the two apart.
+    ///
+    /// # Not idempotent, and the caller owns that
+    ///
+    /// Calling it twice runs the provider's bootstrap twice. Providers whose
+    /// bootstrap is a trigger registration should make that registration
+    /// idempotent themselves; the contract does not promise it, because a
+    /// driver cannot know whether a second call means "retry the one that
+    /// failed" or "the connection was re-authorised".
+    ///
+    /// # Errors
+    ///
+    /// [`MemoryError::Invalid`] for a toolkit the driver has no provider for,
+    /// or a connection it cannot resolve — the same rule
+    /// [`Self::run_connection_sync`] follows, and for the same reason: a
+    /// silent success over a connection that can never bootstrap is worse than
+    /// an error.
+    ///
+    /// [`MemoryError::Unsupported`] from a driver that serves this family but
+    /// not this member. Otherwise the provider's own failure.
+    async fn bootstrap_connection(
+        &self,
+        toolkit: &str,
+        connection_id: &str,
+    ) -> Result<(), MemoryError> {
+        let _ = (toolkit, connection_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 121 members on it, built as a `cdylib`. A host that
+//! exports one object with 122 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

@@ -294,6 +294,8 @@ pub mod methods {
     /// `RunSourceSync` — run one configured memory source's sync now,
     /// whatever kind it is.
     pub const RUN_SOURCE_SYNC: &str = "RunSourceSync";
+    /// `BootstrapConnection` — run one connection's first-time bootstrap.
+    pub const BOOTSTRAP_CONNECTION: &str = "BootstrapConnection";
     /// `SourceSyncState` — the persisted cursor and budget for one connection.
     pub const SOURCE_SYNC_STATE: &str = "SourceSyncState";
     /// `SyncAuditLog` — past sync runs, newest first.
@@ -319,7 +321,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; 121] = [
+pub const METHODS: [&str; 122] = [
     methods::DRIVER_ID,
     methods::CAPABILITIES,
     methods::HEALTH,
@@ -433,6 +435,7 @@ pub const METHODS: [&str; 121] = [
     methods::DIAGNOSE,
     methods::RUN_CONNECTION_SYNC,
     methods::RUN_SOURCE_SYNC,
+    methods::BOOTSTRAP_CONNECTION,
     methods::SOURCE_SYNC_STATE,
     methods::SYNC_AUDIT_LOG,
     methods::ESTIMATE_SYNC_COST_USD,

crates/tinymemory-module/src/lib.rs

@@ -170,6 +170,15 @@ async fn setup(connection: Connection, mut config: ModuleConfig) -> BusResult<()
     // credential strip above, because this is the seam that hands the config
     // back out to the engine repeatedly.
     tinymemory_core::config_loader::set_config_loader(Arc::new(ModuleConfigLoader::new(&config)));
+    // The Composio provider registry is a process-global too, and it is the one
+    // the host used to fill on its own boot. This process has its own statics,
+    // so without this line `get_provider` answers `None` for every toolkit
+    // inside the module — and it answers `None` rather than failing to build,
+    // which is why nothing above catches it. `BootstrapConnection` is the
+    // member that reads it; the sync pipeline resolves its provider a different
+    // way and is unaffected either way. Idempotent by the registry's own
+    // contract, so a second call from a host that also inits is harmless.
+    tinymemory_core::sync::composio::providers::init_default_providers();
     host::install(connection.clone());
     // The two seams no bus interface serves, and no local answer can honestly
     // stand in for. Both degraded in silence rather than with a named cause;
@@ -763,6 +772,7 @@ mod exports {
             // live here; these are the on-demand half plus what past runs cost.
             "RunConnectionSync",
             "RunSourceSync",
+            "BootstrapConnection",
             "SourceSyncState",
             "SyncAuditLog",
             "EstimateSyncCostUsd",

crates/tinymemory-module/src/service/mod.rs

@@ -65,6 +65,7 @@
 //! Diagnose()                                        -> Diagnosis
 //!
 //! RunConnectionSync(toolkit, connection_id)         -> SyncRunOutcome
+//! BootstrapConnection(toolkit, connection_id)       -> ()
 //! RunSourceSync(source_id)                          -> SyncRunOutcome
 //! SourceSyncState(toolkit, connection_id)           -> Option<SourceSyncState>
 //! SyncAuditLog(limit)                               -> [SyncAuditEntry]
@@ -1750,6 +1751,19 @@ impl MemoryService {
             .map_err(|error| into_bus_error(&error))
     }
 
+    /// Run one connection's first-time bootstrap.
+    ///
+    /// Beside `RunConnectionSync` rather than inside it: a sync moves items and
+    /// runs many times, a bootstrap establishes what a sync then assumes and
+    /// runs once. They also fail differently, and a caller can only decline to
+    /// stop syncing over a failed bootstrap if it can tell the two apart.
+    async fn bootstrap_connection(&self, toolkit: String, connection_id: String) -> BusResult<()> {
+        require_family!(self, as_source_sync, Capability::SourceSync)
+            .bootstrap_connection(&toolkit, &connection_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/src/service/test.rs

@@ -743,6 +743,12 @@ async fn the_two_new_families_are_gated_on_their_own_capability() {
         .expect_err("a driver without the source-sync family must refuse");
     assert_eq!(refusal(error), wire::UNSUPPORTED);
 
+    let error = service
+        .bootstrap_connection("gmail".to_string(), "conn-1".to_string())
+        .await
+        .expect_err("a driver without the source-sync family must refuse");
+    assert_eq!(refusal(error), wire::UNSUPPORTED);
+
     let error = service
         .coding_session_status()
         .await
@@ -763,3 +769,35 @@ async fn the_two_new_families_are_gated_on_their_own_capability() {
         .expect_err("a driver without the maintenance family must refuse");
     assert_eq!(refusal(error), wire::UNSUPPORTED);
 }
+
+/// The Composio provider registry is filled by this process, not by the host.
+///
+/// It is a process-global, and before the memory engine moved into a module the
+/// host's own boot was what called `init_default_providers`. A `cdylib` has its
+/// own statics, so that call does nothing for this process — and the failure is
+/// silent in the worst way: `get_provider` answers `None` rather than erroring,
+/// so `BootstrapConnection` would report "no composio provider registered for
+/// 'gmail'" on a perfectly good connection, and nothing in a build or a type
+/// check would have said so.
+///
+/// This pins the call the module's startup makes. It is deliberately asserting
+/// a toolkit the registry's own `init_default_providers` registers rather than
+/// an arbitrary string, so that a rename upstream fails here instead of in the
+/// field.
+#[test]
+fn the_default_composio_providers_populate_the_registry() {
+    use tinymemory_core::sync::composio::providers::{get_provider, init_default_providers};
+
+    init_default_providers();
+
+    assert!(
+        get_provider("gmail").is_some(),
+        "init_default_providers must register the gmail provider; BootstrapConnection \
+         resolves through this registry and answers Invalid when it is empty"
+    );
+    assert!(
+        get_provider("__definitely_not_a_real_toolkit__").is_none(),
+        "an unregistered toolkit must stay unregistered — otherwise the assertion above \
+         would pass against a registry that returns something for everything"
+    );
+}

crates/tinymemory-module/tests/module_e2e.rs

@@ -695,6 +695,7 @@ const EXPECTED_METHODS: &[&str] = &[
     "Diagnose",
     "RunConnectionSync",
     "RunSourceSync",
+    "BootstrapConnection",
     "SourceSyncState",
     "SyncAuditLog",
     "EstimateSyncCostUsd",
@@ -1671,3 +1672,46 @@ async fn portability_and_lifecycle_round_trip(bus: &tinybus::Proxy) {
     .expect("Shutdown timed out")
     .expect("Shutdown");
 }
+
+#[tokio::test]
+#[ignore = "drives a real dlopen'ed module; must be the only such test in the process — see the module docs"]
+async fn bootstrap_connection_finds_its_provider_registry_inside_the_module() {
+    // The Composio provider registry is a process-global that the *host's* boot
+    // used to fill. This module is a `cdylib` with its own statics, so unless
+    // its startup calls `init_default_providers` the registry here is empty —
+    // and `get_provider` answers `None` rather than erroring, so every
+    // `BootstrapConnection` would report "no composio provider registered" over
+    // a perfectly good connection. Nothing in a build, a type check or a unit
+    // test in the module's own workspace sees that, because they all run in a
+    // process the host has already initialised.
+    //
+    // So this asserts against the *loaded artifact*, and it asserts the
+    // distinction rather than the outcome. `Ok` means the provider resolved
+    // and its bootstrap ran; any other error means it resolved and the run
+    // failed on its own terms. Exactly one result says the registry was never
+    // populated, and that is the regression.
+    //
+    // An earlier revision required failure here, on the assumption that a temp
+    // workspace has no Composio — and the call succeeded, because the module's
+    // proxied `ComposioHost` answers through the test harness and the default
+    // bootstrap is content with that. Asserting the symptom instead of the
+    // mechanism made the test wrong about the one thing it exists to pin.
+    let workspace = tempfile::tempdir().expect("tempdir");
+    let (client, _host, _task) = admit_module(workspace.path()).await;
+
+    let result: Result<(), _> = proxy(&client)
+        .call(
+            "BootstrapConnection",
+            ("gmail".to_string(), "conn-1".to_string()),
+        )
+        .await;
+
+    if let Err(error) = result {
+        let rendered = format!("{error:?}");
+        assert!(
+            !rendered.contains("no composio provider registered"),
+            "the module's provider registry is empty — its startup did not call \
+             init_default_providers. Error was: {rendered}"
+        );
+    }
+}

crates/tinymemory-tinycortex/src/engine/mod.rs

@@ -2502,6 +2502,50 @@ impl MemorySourceSync for TinycortexProvider {
         })
     }
 
+    async fn bootstrap_connection(
+        &self,
+        toolkit: &str,
+        connection_id: &str,
+    ) -> Result<(), MemoryError> {
+        use tinymemory_core::sync::composio::providers::{get_provider, ProviderContext};
+
+        // Same gate as `run_connection_sync`, and deliberately before the
+        // provider lookup: a toolkit with no pipeline cannot bootstrap into
+        // anything a later sync would read, so reporting it here names the
+        // real problem rather than "no provider".
+        ensure_syncable_toolkit(toolkit)?;
+
+        let provider = get_provider(toolkit).ok_or_else(|| {
+            MemoryError::Invalid(format!("no composio provider registered for '{toolkit}'"))
+        })?;
+
+        // `from_config` answers `None` when no Composio client resolves in
+        // either mode — the not-signed-in case. That is `Invalid` rather than a
+        // silent `Ok`: a caller that just authorised a connection and gets a
+        // success back would believe the profile was fetched.
+        let ctx = ProviderContext::from_config(
+            self.config.to_arc(),
+            toolkit,
+            Some(connection_id.to_string()),
+        )
+        .ok_or_else(|| {
+            MemoryError::Invalid(format!(
+                "no viable composio client for '{toolkit}'; connection {connection_id} \
+                 cannot bootstrap"
+            ))
+        })?;
+
+        // `max_items` / `sync_depth_days` are left at their defaults on
+        // purpose. They cap how much a *sync* walks; a bootstrap fetches one
+        // profile and registers what the provider needs, and giving it a walk
+        // budget would imply it walks.
+        provider.on_connection_created(&ctx).await.map_err(|error| {
+            MemoryError::Other(anyhow::anyhow!(
+                "bootstrap {toolkit} connection {connection_id}: {error}"
+            ))
+        })
+    }
+
     async fn source_sync_state(
         &self,
         toolkit: &str,