From 2920bbcaff8392fe0dccf7e297044830cd9f4525 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 25 Aug 2026 19:31:18 +0530 Subject: [PATCH] Let the module resolve a backend session, so proxied Composio can run in it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Composio sync has two credential paths and only one of them worked inside a loaded module. The direct branch reads its API key through `ComposioHost::api_key`, a seam the host answers per call. The proxied branch read `Config::session_token`, which inside a module is a load-time snapshot carrying no bearer — so `EngineRuntimeConfig` refused it by design and every proxied user fell out of the branch. The consequence was quiet and large. A host whose Composio mode is backend — which is OpenHuman's default — got a periodic sync loop that never started, because `composio_sync_can_run` gated on direct mode, and neither the host nor the module reported it, since neither believed it was responsible. `ComposioHost` gains `session_bearer`, beside the `api_key` that already works. `composio_config`'s proxied branch consults the seam first and falls back to the config exactly as before, so a host running the engine in-process behaves identically — no seam is installed there, the accessor answers `None`, and the old config read still happens. ## Why a seam rather than a field on ModuleConfig The bearer is an app-session JWT the host refreshes. A value captured at module load works until it expires and then makes every sync fail with an auth error that reads as the user being signed out — the silent-staleness failure this whole migration keeps having to design against. Asking per call means the answer is always the one that is valid now. It is the same reasoning `api_key` already carries, and the reason that member is fetched per call too. ## What the gate now excludes Both modes qualify: direct resolves a key, backend resolves a bearer. What is still refused is a host that resolved to *neither* — an empty or unrecognised mode string, which has no credential path at all, so starting the loop would fail on every tick and append a failed audit row each time. The trait member is defaulted to `None`, so a host that predates it compiles unchanged and falls back to the config read it always did. `session_bearer` deliberately does NOT copy `is_available`'s optimism. That probe answers `true` when it cannot reach the host, because a wrong `false` there reads as "not signed in" and hides a broken sync. A credential is not something to be optimistic about: an unreachable host yields `None`, which lets `composio_config` refuse by name rather than send an empty bearer at the backend. --- crates/tinymemory-core/src/composio_host.rs | 32 +++++++++++++++++ .../src/sync/pipelines/host.rs | 10 ++++-- crates/tinymemory-module/src/composio.rs | 19 ++++++++++ crates/tinymemory-module/src/composio_test.rs | 36 +++++++++++++++++++ crates/tinymemory-module/src/lib.rs | 29 +++++++++------ crates/tinymemory-module/src/service/test.rs | 24 ++++++++----- 6 files changed, 129 insertions(+), 21 deletions(-) diff --git a/crates/tinymemory-core/src/composio_host.rs b/crates/tinymemory-core/src/composio_host.rs index 3258901e..71337830 100644 --- a/crates/tinymemory-core/src/composio_host.rs +++ b/crates/tinymemory-core/src/composio_host.rs @@ -75,6 +75,27 @@ pub trait ComposioHost: Send + Sync + std::fmt::Debug { /// `None` when direct mode is not configured. fn api_key(&self, config: &Config) -> Option; + /// The OpenHuman backend bearer for proxied ("backend") mode. + /// + /// A seam rather than a config field, and that is the whole point of it. + /// The bearer is an app-session JWT the host refreshes; a value captured + /// once — at module load, say — works until it expires and then makes every + /// sync fail with an auth error that reads as the user being signed out. + /// Asking per call means the answer is always the one that is valid now. + /// + /// `None` means the host has no session to lend, which is a signed-out user + /// rather than a broken one. The caller must not read that as "nothing to + /// sync": [`composio_config`](crate::sync::pipelines::host::composio_config) + /// turns it into a named refusal instead. + /// + /// Defaulted to `None` so a host that predates this member still compiles + /// and simply falls back to whatever `Config::session_token` answers, which + /// is exactly the behaviour it had before the member existed. + fn session_bearer(&self, config: &Config) -> Option { + let _ = config; + None + } + /// Whether *some* viable client resolves for the current config. /// /// The sync layer uses this as its "is the user signed in?" probe. It must @@ -146,6 +167,17 @@ pub fn api_key(config: &Config) -> Option { composio_host()?.api_key(config) } +/// The backend bearer from the installed host, or `None` when no host is +/// installed or the host has no session. +/// +/// The two are deliberately not distinguished here: both mean "this process +/// cannot authenticate a proxied Composio call right now", and the caller's +/// fallback and error message are the same either way. +#[must_use] +pub fn session_bearer(config: &Config) -> Option { + composio_host()?.session_bearer(config) +} + /// Whether a viable Composio client resolves. `false` when unwired. #[must_use] pub fn is_available(config: &Config) -> bool { diff --git a/crates/tinymemory-core/src/sync/pipelines/host.rs b/crates/tinymemory-core/src/sync/pipelines/host.rs index dbc91e91..bb0abf3e 100644 --- a/crates/tinymemory-core/src/sync/pipelines/host.rs +++ b/crates/tinymemory-core/src/sync/pipelines/host.rs @@ -259,8 +259,14 @@ pub fn composio_config(config: &Config) -> Result { entity_id: Some(config.composio().entity_id.clone()), }) } else { - let bearer = config - .session_token()? + // The seam first, the config second — the mirror of the direct branch + // above. Inside a loaded module `session_token` cannot answer (the + // module holds a load-time snapshot with no bearer in it), so without + // the seam this branch refuses for every proxied user. Outside a module + // no host is installed, the seam answers `None`, and this falls through + // to exactly the config read it always did. + let bearer = crate::composio_host::session_bearer(config) + .or_else(|| config.session_token().ok().flatten()) .ok_or_else(|| "OpenHuman backend bearer token is not configured".to_string())?; Ok(ComposioSyncConfig { mode: ComposioMode::Proxied, diff --git a/crates/tinymemory-module/src/composio.rs b/crates/tinymemory-module/src/composio.rs index c843c1e6..b735dff0 100644 --- a/crates/tinymemory-module/src/composio.rs +++ b/crates/tinymemory-module/src/composio.rs @@ -84,6 +84,13 @@ pub const API_KEY_METHOD: &str = "ApiKey"; /// Whether *some* viable Composio client resolves host-side right now. pub const IS_AVAILABLE_METHOD: &str = "IsAvailable"; +/// The OpenHuman backend bearer for proxied mode, or `None` when signed out. +/// +/// Asked per call rather than carried in `ModuleConfig` because it is a session +/// JWT the host refreshes: a snapshot works until it expires and then reads as +/// a signed-out user on every subsequent sync. +pub const SESSION_BEARER_METHOD: &str = "SessionBearer"; + /// Latched so the gap is reported once per process rather than once per sync /// tick — the periodic scheduler consults this seam on every tick, and an /// unlatched report would page on every one of them. Same guard the scheduler @@ -353,6 +360,18 @@ impl ComposioHost for BusComposioHost { self.probe::>(API_KEY_METHOD).flatten() } + /// The proxied-mode bearer, fetched per call for the reason on + /// [`SESSION_BEARER_METHOD`]. + /// + /// An unreachable host flattens to `None`, which the caller turns into a + /// named refusal rather than silence — the opposite of `is_available`'s + /// optimistic answer below, and deliberately so: a bearer this process + /// cannot obtain is not a credential it may guess at. + fn session_bearer(&self, _config: &tinymemory_core::Config) -> Option { + self.probe::>(SESSION_BEARER_METHOD) + .flatten() + } + /// Whether the sync layer should treat the user as signed in. /// /// # An unreachable host answers *yes*, deliberately diff --git a/crates/tinymemory-module/src/composio_test.rs b/crates/tinymemory-module/src/composio_test.rs index 85f8f27e..406e3b1a 100644 --- a/crates/tinymemory-module/src/composio_test.rs +++ b/crates/tinymemory-module/src/composio_test.rs @@ -30,6 +30,7 @@ struct Executed { struct FakeComposioHost { executed: tokio::sync::mpsc::UnboundedSender, api_key: Option, + session_bearer: Option, available: bool, } @@ -76,6 +77,11 @@ impl FakeComposioHost { Ok(self.api_key.clone()) } + async fn session_bearer(&self) -> BusResult> { + std::future::ready(()).await; + Ok(self.session_bearer.clone()) + } + async fn is_available(&self) -> BusResult { std::future::ready(()).await; Ok(self.available) @@ -98,6 +104,7 @@ async fn bus_with_composio_host( FakeComposioHost { executed, api_key: api_key.map(str::to_string), + session_bearer: Some("bearer-from-the-host".to_string()), available, }, ) @@ -263,3 +270,32 @@ fn only_the_nobody_is_listening_family_reads_as_unserved() { assert!(!unserved("ai.tinyhumans.tinybus.Error.Failed")); assert!(!unserved("ai.tinyhumans.tinybus.Error.Timeout")); } + +/// The proxied-mode bearer crosses the bus, and an unreachable host answers +/// `None` rather than guessing. +/// +/// The second half is the half worth pinning. `is_available` deliberately +/// answers `true` when it cannot reach the host, because a wrong `false` there +/// reads as "not signed in" and hides a sync that is actually broken. This +/// member must not copy that: a credential is not something to be optimistic +/// about, and `None` is what lets `composio_config` refuse by name instead of +/// sending an empty bearer at the backend. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_session_bearer_crosses_the_bus_and_is_never_guessed() { + let (connection, _executed) = bus_with_composio_host(None, true).await; + let bridge = BusComposioHost::new(connection); + let config = tinymemory_tinycortex::engine::EngineRuntimeConfig::from(&ModuleConfig::default()); + + assert_eq!( + bridge.session_bearer(&config).as_deref(), + Some("bearer-from-the-host"), + "the host's live session must reach the engine unchanged" + ); + + let unserved = BusComposioHost::new(bus_without_composio_host().await); + assert_eq!( + unserved.session_bearer(&config), + None, + "an unreachable host must not be optimistic about a credential" + ); +} diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index fb3ea52c..67443450 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -378,18 +378,25 @@ fn start_composio_periodic_sync(config: &ModuleConfig) { } 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`" + "[tinymemory:module] periodic Composio sync is NOT started: this host resolved Composio \ + to neither direct nor backend mode, so no credential can be obtained for it. \ + Composio-connected sources will not update in this process" ); } /// 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. +/// Both modes qualify now. Direct mode reads its API key through +/// `ComposioHost::api_key`, and backend mode reads its bearer through +/// `ComposioHost::session_bearer` — the seam added precisely so this gate could +/// stop excluding the mode most hosts actually run. It used to be direct-only, +/// which meant the loop silently did not start for a host whose default is +/// backend, and neither side reported it because neither thought it was +/// responsible. +/// +/// What is still excluded is a host that resolved to *neither* — an empty or +/// unrecognised mode string. There is no credential path for that, so starting +/// the loop would fail on every tick and append a failed audit row each time. /// /// Asked of the *same* `EngineRuntimeConfig` the loop's own ticks will be handed /// and through the same `MemoryHostConfig::composio` accessor `composio_config` @@ -404,9 +411,11 @@ fn start_composio_periodic_sync(config: &ModuleConfig) { /// 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() + let composio = tinymemory_tinycortex::engine::EngineRuntimeConfig::from(config).composio(); + // Mirrors `composio_config`'s own branch: direct, else anything that names + // a mode at all takes the proxied path. An unset mode names neither and is + // the one case with no credential to reach for. + composio.is_direct() || !composio.mode.trim().is_empty() } /// The workspace whose queue this process's worker pool drains. diff --git a/crates/tinymemory-module/src/service/test.rs b/crates/tinymemory-module/src/service/test.rs index 9e1640b0..5a7b6b5c 100644 --- a/crates/tinymemory-module/src/service/test.rs +++ b/crates/tinymemory-module/src/service/test.rs @@ -530,18 +530,24 @@ fn the_sync_loops_are_claimed_once_and_a_foreign_workspace_is_refused() { /// 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. +/// that started the loop for a mode with no credential path would list the +/// user's connections every 20 minutes and fail every due one, appending a +/// failed row to the sync audit each time; a gate that refused a mode that CAN +/// resolve one would leave a host whose Composio sources simply stop updating, +/// with a single line at boot to explain it. +/// +/// Backend mode moved from the second category to the first when +/// `ComposioHost::session_bearer` landed. It used to be excluded because +/// `EngineRuntimeConfig::session_token` refuses by design — which meant the +/// loop did not start for a host whose default mode is backend, and neither the +/// host nor the module reported it, because neither thought it was responsible. /// /// 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() { +fn composio_periodic_sync_starts_for_any_mode_that_can_resolve_a_credential() { let mut config = test_config(std::path::Path::new("/tinymemory-module/composio-gate")); assert!( @@ -551,14 +557,14 @@ fn composio_periodic_sync_starts_only_when_the_host_resolved_direct_mode() { 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" + crate::composio_sync_can_run(&config), + "backend mode resolves its bearer through ComposioHost::session_bearer" ); 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" + "direct mode resolves its key through ComposioHost::api_key" ); // The pipeline's own branch test is case-insensitive. If the gate were not,