fix(agent): refresh delegation surface on mid-session Composio connect/revoke (#1687)

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/openhuman/agent/debug/mod.rs

@@ -218,7 +218,10 @@ async fn render_via_session(config: &Config, agent_id: &str) -> Result<DumpedPro
     agent.fetch_connected_integrations().await;
     // Mirror turn-1: synthesise `delegate_*` tools for connected
     // Composio toolkits now that we know what's actually authorised.
-    agent.refresh_delegation_tools();
+    // The shared-Arc failure path is unreachable here (this is the
+    // debug dumper running against a freshly-built agent — no
+    // sub-agent has cloned the tool list), so ignore the bool return.
+    let _ = agent.refresh_delegation_tools();
 
     let text = agent
         .build_system_prompt(LearnedContextData::default())

src/openhuman/agent/harness/session/builder.rs

@@ -420,6 +420,8 @@ impl AgentBuilder {
             omit_profile: self.omit_profile.unwrap_or(true),
             omit_memory_md: self.omit_memory_md.unwrap_or(true),
             payload_summarizer: self.payload_summarizer,
+            last_seen_integrations_hash: 0,
+            synthesized_tool_names: std::collections::HashSet::new(),
         })
     }
 }

src/openhuman/agent/harness/session/turn.rs

@@ -88,8 +88,12 @@ impl Agent {
             // Now that integrations are live, inject the matching
             // delegation tools so the orchestrator's prompt + tool-spec
             // list actually expose `delegate_gmail`, `delegate_notion`,
-            // etc.
-            self.refresh_delegation_tools();
+            // etc. The shared-Arc failure path returns `false`, but on
+            // turn 1 the Arc is uniquely owned (no sub-agent has run
+            // yet); a `false` return here would indicate a programmer
+            // error and the warn-level log inside the helper already
+            // surfaces it, so we ignore the return value here.
+            let _ = self.refresh_delegation_tools();
             let learned = self.fetch_learned_context().await;
             let rendered_prompt = self.build_system_prompt(learned)?;
             log::info!("[agent] system prompt built — initialising conversation history");
@@ -119,6 +123,11 @@ impl Agent {
                 .push(ConversationMessage::Chat(ChatMessage::system(
                     rendered_prompt,
                 )));
+            // Seed the per-turn mid-session refresh baseline with the
+            // hash of whatever Composio actually returned just now.
+            // Subsequent turns short-circuit unless this hash changes.
+            self.last_seen_integrations_hash =
+                crate::openhuman::composio::connected_set_hash(&self.connected_integrations);
         } else {
             // Deliberately do NOT rebuild the system prompt on subsequent
             // turns. The rendered prompt is the KV-cache prefix the inference
@@ -129,6 +138,75 @@ impl Agent {
             // rides on the user message via `memory_loader.load_context()`
             // — that's where the caller should inject anything that varies
             // between turns.
+            //
+            // *** Mid-session schema-only refresh ***
+            //
+            // The system prompt stays frozen, but the function-calling
+            // schema (the `tools` field in the provider request) is sent
+            // fresh on every API call — it's not part of the KV-cache
+            // prefix. So we *can* react to Composio connect/disconnect
+            // events mid-session by re-synthesising the `delegate_<toolkit>`
+            // 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<String> =
+            synthed.iter().map(|t| t.name().to_string()).collect();
+        let synthed_specs: Vec<crate::openhuman::tools::ToolSpec> =
+            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<String> =
-            self.tools.iter().map(|t| t.name().to_string()).collect();
-        let new_tools: Vec<Box<dyn Tool>> = 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<String> = new_tools.iter().map(|t| t.name().to_string()).collect();
-        let new_specs: Vec<crate::openhuman::tools::ToolSpec> =
-            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<crate::openhuman::tools::ToolSpec> =
             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<String>`
+        // so we can move `synthed_names` into `self.synthesized_tool_names`
+        // below without the log-statement reborrow blocking the move.
+        let added: Vec<String> = synthed_names
+            .iter()
+            .filter(|n| !old_synth.contains(n.as_str()))
+            .cloned()
+            .collect();
+        let removed: Vec<String> = 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

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<Arc<dyn crate::openhuman::agent::harness::payload_summarizer::PayloadSummarizer>>,
+    /// 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_<toolkit>` 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_<toolkit>` 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_<toolkit>` 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<String>,
 }
 
 /// A builder for creating `Agent` instances with custom configuration.

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,

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};

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<Vec<ConnectedIntegration>> {
+    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_<toolkit>` 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