diff --git a/src/openhuman/agent/debug/mod.rs b/src/openhuman/agent/debug/mod.rs index 3abc19cca..8db39f828 100644 --- a/src/openhuman/agent/debug/mod.rs +++ b/src/openhuman/agent/debug/mod.rs @@ -218,7 +218,10 @@ async fn render_via_session(config: &Config, agent_id: &str) -> Result` + // surface on `self.tools` / `self.tool_specs` and letting + // the next provider call carry the new schema. KV cache stays + // intact; the system prompt's `## Connected Integrations` + // block goes mildly stale until the next session, but the + // schema is the source of truth the model actually routes + // against. + // + // The signal we react to is the process-wide + // [`crate::openhuman::composio::INTEGRATIONS_CACHE`], kept + // current by (a) the desktop UI's 5 s + // `composio_list_connections` poll, (b) the post-OAuth + // `ComposioConnectionCreatedSubscriber` invalidation, and + // (c) the 60 s TTL fallback. We read it via the read-only + // [`crate::openhuman::composio::cached_active_integrations`] + // helper — never trigger a backend fetch ourselves, never + // block on a writer. + // + // We need a `Config` to key into `INTEGRATIONS_CACHE`. The + // `Config::load_or_init()` call is cached internally so this + // is cheap on the hot path. On config-load failure we skip + // the refresh — no signal we can safely act on, same as + // when the cache itself is empty. + if let Ok(cfg) = crate::openhuman::config::Config::load_or_init().await { + if let Some(cache_view) = + crate::openhuman::composio::cached_active_integrations(&cfg) + { + let new_hash = crate::openhuman::composio::connected_set_hash(&cache_view); + if new_hash != self.last_seen_integrations_hash { + log::info!( + "[agent_loop] connection set changed mid-session (hash {:x} -> {:x}); refreshing tool schema (system prompt left intact for KV cache)", + self.last_seen_integrations_hash, + new_hash + ); + // Snapshot the previous integration list so we + // can roll back if `refresh_delegation_tools` + // bails on a shared `Arc` — otherwise the + // agent's `connected_integrations` and its + // schema would disagree until the next + // event-driven refresh, and the + // `last_seen_integrations_hash` advance below + // would suppress retries. + let prev_integrations = + std::mem::replace(&mut self.connected_integrations, cache_view); + if self.refresh_delegation_tools() { + self.last_seen_integrations_hash = new_hash; + } else { + // Reconcile aborted (shared Arc) — restore + // the previous integration list so the + // next turn re-detects the same change and + // retries cleanly. We deliberately do NOT + // advance `last_seen_integrations_hash`. + self.connected_integrations = prev_integrations; + } + } + } + } + // Cache empty/expired or config unavailable => no signal. + // We leave the current tool surface alone and pick up any + // real change on the next turn after the UI's 5 s poll has + // repopulated [`INTEGRATIONS_CACHE`]. + log::trace!( "[agent_loop] system prompt reused (history_len={}) — KV cache prefix preserved", self.history.len() @@ -1260,100 +1338,134 @@ impl Agent { } /// Re-synthesise `delegate_*` tools for the orchestrator's `subagents` - /// declaration using the live `connected_integrations` slice. + /// declaration using the live `connected_integrations` slice, and + /// reconcile the resulting set into `self.tools` / `self.tool_specs` / + /// `self.visible_tool_specs` / `self.visible_tool_names`. /// - /// The synchronous `AgentBuilder` path cannot do this — it has no - /// async runtime handle to reach Composio — so it passes an empty - /// integrations slice to `collect_orchestrator_tools` and produces - /// zero skill delegation tools. That leaves the orchestrator without - /// `delegate_gmail` / `delegate_notion` / … even when those toolkits - /// are actually authorised. Call this *after* - /// [`Agent::fetch_connected_integrations`] (and before - /// [`Agent::build_system_prompt`]) on turn 1 to add the missing - /// delegation tools to the live tool registry, tool-spec list, and - /// visible-tool whitelist so the LLM sees them as first-class tools. + /// **Reconciliation strategy** — full rebuild of the synthesised + /// subset: /// - /// No-op when: - /// * the agent definition isn't in the global registry, - /// * the definition has no `subagents` entries, - /// * no new tools would be added (every synthesised name already - /// exists in `self.tools`), - /// * the `tools` / `tool_specs` Arc is shared (e.g. a sub-agent has - /// already captured a snapshot — should never happen on turn 1). - pub fn refresh_delegation_tools(&mut self) { + /// 1. Drop every tool whose name was in [`Self::synthesized_tool_names`] + /// from the previous synthesis. Direct tools (`query_memory`, + /// `cron_add`, …) are untouched because their names are not in + /// that set. + /// 2. Append the freshly collected synthesis output verbatim. + /// 3. Replace `synthesized_tool_names` with the new set so the + /// next refresh has a clean mask to undo. + /// + /// This is safer than appending-only or strict-diff reconcile: + /// + /// * Stale tools after a revoke can never leak — anything from the + /// previous synthesis is unconditionally dropped, the new set is + /// authoritative. + /// * Direct tools can never be accidentally removed — only names + /// in `synthesized_tool_names` are touched. + /// * Duplicate registration is impossible — retain+extend + /// guarantees every final entry is either a non-synthesised + /// direct tool or a member of the fresh `synthed` set. + /// + /// **When to call**: on turn 1 (after [`Agent::fetch_connected_integrations`] + /// populates `self.connected_integrations` for the first time) and + /// on any subsequent turn where the connection set has changed + /// since the last reconcile (detected via + /// [`Self::last_seen_integrations_hash`] vs. + /// [`crate::openhuman::composio::cached_active_integrations`]). + /// + /// **Atomicity**: when `Arc::get_mut` fails (a sub-agent or other + /// caller has already captured a clone of the tool list), we restore + /// the previous `synthesized_tool_names` and bail. The next refresh + /// attempt will re-apply the full transition cleanly rather than + /// resuming from a partial state. This should never happen on a turn + /// boundary in production — sub-agents always drop their snapshots + /// before the parent's next turn — but it's defended against anyway. + /// + /// **Return value** — `true` when the agent's tool surface is now + /// consistent with `self.connected_integrations` (either because a + /// successful reconcile applied, or because no reconcile was needed). + /// `false` only when the Arc was shared and the reconcile was + /// aborted; callers should treat this as "retry next turn" and + /// **not** advance any signal they use to gate future refreshes + /// (e.g. `last_seen_integrations_hash`) — otherwise a one-shot + /// shared-Arc collision could suppress further reconciliation until + /// another integration event happened to bump the hash again. + pub fn refresh_delegation_tools(&mut self) -> bool { use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry; use crate::openhuman::tools::orchestrator_tools::collect_orchestrator_tools; let Some(reg) = AgentDefinitionRegistry::global() else { - return; + // No registry — there's nothing we can do until the + // registry is initialised. The agent's surface stays at + // whatever the builder produced; callers can safely treat + // this as "no reconcile needed right now". + return true; }; let Some(def) = reg.get(&self.agent_definition_id) else { log::debug!( "[agent] refresh_delegation_tools: definition '{}' not in registry — skipping", self.agent_definition_id ); - return; + return true; }; if def.subagents.is_empty() { - return; + return true; } let synthed = collect_orchestrator_tools(def, reg, &self.connected_integrations); - if synthed.is_empty() { - return; + let synthed_names: std::collections::HashSet = + synthed.iter().map(|t| t.name().to_string()).collect(); + let synthed_specs: Vec = + synthed.iter().map(|t| t.spec()).collect(); + + // Skip the Arc mutation entirely when neither the previous nor + // the next synthesis produced any names — saves an + // `Arc::get_mut` probe on agents that don't declare `subagents`. + if self.synthesized_tool_names.is_empty() && synthed_names.is_empty() { + return true; } - let existing: std::collections::HashSet = - self.tools.iter().map(|t| t.name().to_string()).collect(); - let new_tools: Vec> = synthed - .into_iter() - .filter(|t| !existing.contains(t.name())) - .collect(); - if new_tools.is_empty() { - return; - } + // Mask used to drop the previous synthesis. We `take` it now so + // the restore-on-failure path below can put it back if the Arc + // turns out to be shared. + let old_synth = std::mem::take(&mut self.synthesized_tool_names); - let new_names: Vec = new_tools.iter().map(|t| t.name().to_string()).collect(); - let new_specs: Vec = - new_tools.iter().map(|t| t.spec()).collect(); - - // The tools / tool_specs Arcs are uniquely owned at turn-1 build - // time (no sub-agent has captured a snapshot yet). If a caller - // managed to clone them out before the first turn fired, the - // mutation below would silently no-op — log that case so it's - // visible in diagnostics. match ( Arc::get_mut(&mut self.tools), Arc::get_mut(&mut self.tool_specs), ) { (Some(tools_vec), Some(specs_vec)) => { - tools_vec.extend(new_tools); - specs_vec.extend(new_specs); + tools_vec.retain(|t| !old_synth.contains(t.name())); + specs_vec.retain(|s| !old_synth.contains(&s.name)); + tools_vec.extend(synthed); + specs_vec.extend(synthed_specs); } _ => { log::warn!( "[agent] refresh_delegation_tools: tools/tool_specs Arc is shared — \ - cannot inject delegation tools ({} would have been added: {:?})", - new_names.len(), - new_names + cannot reconcile delegation surface (would have produced {} synthesised tool(s)). \ + Restoring previous synthesized_tool_names so the next refresh retries cleanly.", + synthed_names.len() ); - return; + self.synthesized_tool_names = old_synth; + return false; } } - // Definitions with a Named ToolScope carry an explicit visible - // whitelist; the synthesised delegation tools must opt in to that - // whitelist or the LLM never sees them. Wildcard-scope agents - // keep `visible_tool_names` empty ("no filter"), so we skip the - // insert there. + // `visible_tool_names` carries an explicit allowlist for + // [`ToolScope::Named`] agents. Drop the previously-synthesised + // names and add the new ones so the visible set tracks the + // tool list. Wildcard-scope agents keep this empty ("no + // filter") and never need touching. if !self.visible_tool_names.is_empty() { - for name in &new_names { + for name in &old_synth { + self.visible_tool_names.remove(name); + } + for name in &synthed_names { self.visible_tool_names.insert(name.clone()); } } - // Rebuild the visible-spec cache so the next prompt build picks - // up the new tools. + // Rebuild the visible-spec cache from the new tool_specs so the + // next provider call carries the reconciled schema. let visible_specs: Vec = if self.visible_tool_names.is_empty() { (*self.tool_specs).clone() @@ -1366,13 +1478,33 @@ impl Agent { }; self.visible_tool_specs = Arc::new(visible_specs); + // Compute add/remove deltas for the log line — useful when + // diagnosing a Composio connect/revoke that should have rebuilt + // the surface but didn't. Materialise to owned `Vec` + // so we can move `synthed_names` into `self.synthesized_tool_names` + // below without the log-statement reborrow blocking the move. + let added: Vec = synthed_names + .iter() + .filter(|n| !old_synth.contains(n.as_str())) + .cloned() + .collect(); + let removed: Vec = old_synth + .iter() + .filter(|n| !synthed_names.contains(n.as_str())) + .cloned() + .collect(); + + self.synthesized_tool_names = synthed_names; + log::info!( - "[agent] refresh_delegation_tools: added {} delegation tool(s) for agent '{}' (display='{}'): {:?}", - new_names.len(), + "[agent] refresh_delegation_tools: reconciled delegation surface for agent '{}' (display='{}'); now {} synthesised tool(s); added={:?} removed={:?}", self.agent_definition_id, self.agent_definition_name, - new_names + self.synthesized_tool_names.len(), + added, + removed ); + true } /// Builds the system prompt for the current turn, including tool diff --git a/src/openhuman/agent/harness/session/types.rs b/src/openhuman/agent/harness/session/types.rs index f2696e000..5bade6c71 100644 --- a/src/openhuman/agent/harness/session/types.rs +++ b/src/openhuman/agent/harness/session/types.rs @@ -156,6 +156,33 @@ pub struct Agent { /// summarizer sub-agent before they enter agent history. pub(super) payload_summarizer: Option>, + /// Hash of the Composio connection set this Agent last reconciled + /// against. Compared at top-of-turn to a fresh hash computed from + /// [`crate::openhuman::composio::cached_active_integrations`]; on + /// diff, [`Agent::refresh_delegation_tools`] re-synthesises the + /// `delegate_` surface to match the live connected set. + /// + /// Initialised to `0` at construction. Turn 1's existing refresh + /// path (gated by `history.is_empty()`) writes the first real hash + /// after [`Agent::fetch_connected_integrations`] populates + /// [`Agent::connected_integrations`], so the per-turn check is + /// dormant on session startup and only fires when integrations + /// actually change mid-conversation. + pub(super) last_seen_integrations_hash: u64, + /// Names of every tool currently in [`Agent::tools`] that was + /// produced by [`crate::openhuman::tools::orchestrator_tools::collect_orchestrator_tools`] + /// (i.e. `delegate_` skill tools and archetype-delegation + /// tools like `delegate_archivist`). Tracked so + /// [`Agent::refresh_delegation_tools`] can drop the entire + /// previously-synthesised subset on each refresh and append the + /// fresh set — without that mask we'd risk either leaking stale + /// `delegate_` entries on revoke or accidentally removing + /// direct tools (`query_memory`, `cron_add`, …) that share a name + /// prefix. + /// + /// Populated by `refresh_delegation_tools` itself; empty at + /// construction time. + pub(super) synthesized_tool_names: std::collections::HashSet, } /// A builder for creating `Agent` instances with custom configuration. diff --git a/src/openhuman/composio/bus.rs b/src/openhuman/composio/bus.rs index 95743643f..9280b108b 100644 --- a/src/openhuman/composio/bus.rs +++ b/src/openhuman/composio/bus.rs @@ -368,14 +368,19 @@ impl EventHandler for ComposioConnectionCreatedSubscriber { "[composio:bus] connection_created" ); - let Some(provider) = get_provider(toolkit) else { - tracing::debug!( - toolkit = %toolkit, - "[composio:bus] no provider registered, skipping connection_created hook" - ); - return; - }; - + // Run the post-active cache refresh for EVERY toolkit, not just + // ones with a registered provider. Earlier shape gated the + // entire spawn block on `get_provider(toolkit)` — that meant + // toolkits without a provider (most of the 119 Composio + // toolkits, e.g. `googlecalendar`) bypassed the eager cache + // warm and had to wait for the desktop UI's 5 s + // `composio_list_connections` diff-poll to invalidate the + // stale cache. The chat-runtime then missed the new connection + // on any turn that fell inside that window. Decoupling the + // cache refresh from provider routing fixes it: every + // connect → invalidate + eager warm, provider hook becomes a + // downstream optional step gated on its own `get_provider` + // lookup. let toolkit = toolkit.clone(); let connection_id = connection_id.clone(); @@ -385,8 +390,8 @@ impl EventHandler for ComposioConnectionCreatedSubscriber { // has actually clicked through. Resolve the config + client // first, then poll the backend for the connection record // until we observe ACTIVE/CONNECTED (or hit the timeout). - // Only then do we run the provider hook, so the very first - // provider call doesn't race the OAuth handshake. + // Only then do we invalidate + warm the cache so we never + // surface a half-finished connection to the chat runtime. // // NOTE: Future improvement — listen for an explicit // "connection_active" backend event instead of polling. @@ -419,12 +424,48 @@ impl EventHandler for ComposioConnectionCreatedSubscriber { toolkit = %toolkit, connection_id = %connection_id, status = %status, - "[composio:bus] connection observed active, dispatching on_connection_created" + "[composio:bus] connection observed active; invalidating + eagerly warming integrations cache" ); // Bust the prompt-level integrations cache now that // the connection is confirmed ACTIVE, so the next // agent session picks up the newly connected toolkit. super::ops::invalidate_connected_integrations_cache(); + // Eagerly warm the cache from the backend so the + // very next `cached_active_integrations` read + // (typically the orchestrator's next-turn refresh, + // or the desktop UI's 5 s `composio_list_connections` + // poll — whichever fires first) returns the new + // toolkit immediately instead of waiting for a + // cache-miss round trip on the hot path. Cost: one + // background `list_connections` call per OAuth + // completion. Best-effort — on backend failure the + // UI poll will repopulate within ~5 s as a safety + // net. + // + // Use the status-distinguishing fetcher so we log + // `Authoritative(empty)` and backend unavailability + // differently — `fetch_connected_integrations` + // collapses both to `Vec::new()` and would + // otherwise hide auth/backend failures from + // incident triage. + match super::ops::fetch_connected_integrations_status(ctx.config.as_ref()).await + { + super::ops::FetchConnectedIntegrationsStatus::Authoritative(entries) => { + tracing::debug!( + toolkit = %toolkit, + connection_id = %connection_id, + cached_entries = entries.len(), + "[composio:bus] eagerly warmed integrations cache after connection became active" + ); + } + super::ops::FetchConnectedIntegrationsStatus::Unavailable => { + tracing::warn!( + toolkit = %toolkit, + connection_id = %connection_id, + "[composio:bus] eager cache warm after connection became active skipped: backend unavailable" + ); + } + } } Err(WaitError::Timeout { last_status }) => { tracing::warn!( @@ -432,7 +473,7 @@ impl EventHandler for ComposioConnectionCreatedSubscriber { connection_id = %connection_id, last_status = ?last_status, timeout_secs = CONNECTION_READY_TIMEOUT.as_secs(), - "[composio:bus] timed out waiting for connection to become active; aborting on_connection_created" + "[composio:bus] timed out waiting for connection to become active; skipping cache refresh + provider hook" ); return; } @@ -441,12 +482,24 @@ impl EventHandler for ComposioConnectionCreatedSubscriber { toolkit = %toolkit, connection_id = %connection_id, error = %error, - "[composio:bus] backend lookup failed while waiting for connection; aborting on_connection_created" + "[composio:bus] backend lookup failed while waiting for connection; skipping cache refresh + provider hook" ); return; } } + // Optional provider-specific post-OAuth hook (e.g. gmail's + // inbox ingest). Only fires for toolkits that registered a + // provider; the rest just got the cache refresh above and + // we're done. + let Some(provider) = get_provider(&toolkit) else { + tracing::debug!( + toolkit = %toolkit, + "[composio:bus] no provider registered for toolkit; cache refreshed, no provider hook to dispatch" + ); + return; + }; + if let Err(e) = provider.on_connection_created(&ctx).await { tracing::warn!( toolkit = %toolkit, diff --git a/src/openhuman/composio/mod.rs b/src/openhuman/composio/mod.rs index 3d4ea1b3e..49b52e477 100644 --- a/src/openhuman/composio/mod.rs +++ b/src/openhuman/composio/mod.rs @@ -50,7 +50,8 @@ pub use action_tool::ComposioActionTool; pub use bus::{register_composio_trigger_subscriber, ComposioTriggerSubscriber}; pub use client::{build_composio_client, ComposioClient}; pub use ops::{ - fetch_connected_integrations, fetch_connected_integrations_status, fetch_toolkit_actions, + cached_active_integrations, connected_set_hash, fetch_connected_integrations, + fetch_connected_integrations_status, fetch_toolkit_actions, invalidate_connected_integrations_cache, FetchConnectedIntegrationsStatus, }; pub use periodic::{record_sync_success, start_periodic_sync}; diff --git a/src/openhuman/composio/ops.rs b/src/openhuman/composio/ops.rs index 942778742..18c045abe 100644 --- a/src/openhuman/composio/ops.rs +++ b/src/openhuman/composio/ops.rs @@ -165,6 +165,36 @@ pub async fn composio_delete_connection( ); // Bust the integrations cache so the next prompt reflects the removal. invalidate_connected_integrations_cache(); + // Eagerly warm the cache from the backend so the very next + // `cached_active_integrations` read (typically the orchestrator's + // next-turn refresh, or the desktop UI's 5 s + // `composio_list_connections` poll) sees the removal immediately + // instead of waiting for a cache-miss round trip on the hot path. + // Symmetric with the connect-side eager warm in + // [`super::bus::ComposioConnectionCreatedSubscriber`]. Best-effort — + // on backend failure the UI poll repopulates within ~5 s as a + // safety net. + // + // Use the status-distinguishing fetcher so we log + // `Authoritative(empty)` and backend unavailability differently — + // `fetch_connected_integrations` collapses both to `Vec::new()` + // and would otherwise hide auth/backend failures from incident + // triage. + match fetch_connected_integrations_status(config).await { + FetchConnectedIntegrationsStatus::Authoritative(entries) => { + tracing::debug!( + connection_id = %connection_id, + cached_entries = entries.len(), + "[composio] eagerly warmed integrations cache after connection deletion" + ); + } + FetchConnectedIntegrationsStatus::Unavailable => { + tracing::warn!( + connection_id = %connection_id, + "[composio] eager cache warm after connection deletion skipped: backend unavailable" + ); + } + } Ok(RpcOutcome::new( resp, vec![format!("composio: connection {connection_id} deleted")], @@ -701,6 +731,101 @@ pub fn invalidate_connected_integrations_cache() { } } +/// Read-only snapshot of the currently cached connected integrations for +/// the given config, or [`None`] when the cache is empty, expired, or +/// the lock is held by a writer. +/// +/// Designed for hot-path callers that want a cheap "what does the cache +/// already say?" probe without triggering a backend fetch. The agent +/// harness uses this on every turn to detect mid-session connection +/// changes — it relies on the desktop UI's 5 s `composio_list_connections` +/// poll (which calls into [`fetch_connected_integrations`] and +/// repopulates this cache) plus the event-driven invalidation path to +/// keep the cache current. +/// +/// `try_read` (not `read`) so a writer in progress — e.g. the UI poll +/// repopulating the cache — never blocks a turn. Worst case the agent +/// sees `None` for one turn while the writer holds the lock; the next +/// turn picks up the value naturally. +/// +/// TTL is enforced defensively: entries older than [`CACHE_TTL`] are +/// treated as missing even though they're still in the map (a stale +/// entry would otherwise pin the agent to a frozen view if every +/// invalidation path silently failed). +pub fn cached_active_integrations(config: &Config) -> Option> { + let key = cache_key(config); + let guard = match INTEGRATIONS_CACHE.try_read() { + Ok(g) => g, + Err(_) => { + tracing::trace!( + key = %key, + "[composio][integrations_cache] cached_active_integrations:lock_contended" + ); + return None; + } + }; + let Some(cached) = guard.get(&key) else { + tracing::trace!( + key = %key, + "[composio][integrations_cache] cached_active_integrations:miss" + ); + return None; + }; + let age = cached.cached_at.elapsed(); + if age > CACHE_TTL { + tracing::trace!( + key = %key, + age_ms = age.as_millis() as u64, + ttl_ms = CACHE_TTL.as_millis() as u64, + "[composio][integrations_cache] cached_active_integrations:expired" + ); + return None; + } + tracing::trace!( + key = %key, + entries = cached.entries.len(), + age_ms = age.as_millis() as u64, + "[composio][integrations_cache] cached_active_integrations:hit" + ); + Some(cached.entries.clone()) +} + +/// Stable hash of the *routing-relevant* slice of a connected-integrations +/// snapshot. +/// +/// Two snapshots produce the same hash iff they would synthesise the +/// same `delegate_` tool set in the orchestrator's +/// function-calling schema. The hash is: +/// +/// - **Order-independent** — callers don't need to sort the input. +/// - **Description-insensitive** — Composio catalogue text edits don't +/// trigger a refresh. The schema's tool-description field still +/// picks up new text on the next *real* (membership-changing) +/// refresh, so descriptions are never permanently stale. +/// - **Process-local** — [`std::collections::hash_map::DefaultHasher`] +/// is randomly seeded per process. Fine because we only compare +/// hashes within one process lifetime. +/// +/// Only `connected == true` entries contribute. Unconnected toolkits are +/// stripped by [`super::super::tools::orchestrator_tools::collect_orchestrator_tools`] +/// anyway, so churn among the unconnected set never changes the agent's +/// surface and shouldn't trigger a refresh. +pub fn connected_set_hash(integrations: &[ConnectedIntegration]) -> u64 { + use std::collections::hash_map::DefaultHasher; + use std::hash::{Hash, Hasher}; + + let mut slugs: Vec<&str> = integrations + .iter() + .filter(|i| i.connected) + .map(|i| i.toolkit.as_str()) + .collect(); + slugs.sort(); + + let mut hasher = DefaultHasher::new(); + slugs.hash(&mut hasher); + hasher.finish() +} + /// Collect the set of toolkit slugs marked `connected` in a snapshot. /// /// Exposed to [`sync_cache_with_connections`] so it can diff the live