feat(composio,agent): async sync + gated-tools surface + UI-only scope elevation + force-delegate capability questions (#2348)

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

src/openhuman/agent/agents/integrations_agent/prompt.rs

@@ -137,6 +137,63 @@ fn render_connected_integrations(integrations: &[ConnectedIntegration]) -> Strin
     for ci in connected {
         let _ = writeln!(out, "- **{}** — {}", ci.toolkit, ci.description);
     }
+
+    // Surface pref-gated tools so the agent can honestly say "I have this
+    // capability but it needs the {scope} toggle in Connections → {toolkit}".
+    // The agent CANNOT call these directly (no parameters schema is exposed)
+    // and CANNOT flip the gating scope itself — there is no agent-callable
+    // scope-elevate tool. The user must toggle the scope in the Connections
+    // UI; after the next prompt rebuild the action graduates into the
+    // callable list above. The per-row `unlock paths` rendered below carry
+    // the exact UI hint the agent should show.
+    let mut has_gated = false;
+    let mut connected_with_gated = 0usize;
+    for ci in integrations.iter().filter(|ci| ci.connected) {
+        if !ci.gated_tools.is_empty() {
+            has_gated = true;
+            connected_with_gated += 1;
+        }
+    }
+    tracing::debug!(
+        total_integrations = integrations.len(),
+        has_gated,
+        connected_with_gated,
+        "[integrations-prompt] gated-tools scan complete"
+    );
+    if has_gated {
+        out.push_str(
+            "\n### Additional capabilities behind a permission toggle\n\n\
+             These actions exist in the toolkit but are NOT currently in your callable \
+             tool list — the user has not granted the required scope. Do NOT pretend \
+             they're unavailable. When the user asks for one (or you'd otherwise need \
+             it), tell them what the action does and present ALL of its `unlock paths` \
+             listed below so the user can choose how to enable it. Never drop a path or \
+             rewrite it into your own framing.\n\n",
+        );
+        for ci in integrations
+            .iter()
+            .filter(|ci| ci.connected && !ci.gated_tools.is_empty())
+        {
+            let _ = writeln!(out, "- **{}**:", ci.toolkit);
+            for gt in &ci.gated_tools {
+                let desc = if gt.description.is_empty() {
+                    "(no description)".to_string()
+                } else {
+                    gt.description.clone()
+                };
+                let _ = writeln!(
+                    out,
+                    "  - `{}` — {} (requires `{}` scope)",
+                    gt.name, desc, gt.required_scope
+                );
+                for path in &gt.unlock_paths {
+                    let _ = writeln!(out, "    - unlock path: {path}");
+                }
+            }
+        }
+        out.push('\n');
+    }
+
     out
 }
 
@@ -187,6 +244,7 @@ mod tests {
             toolkit: "gmail".into(),
             description: "Email access.".into(),
             tools: Vec::new(),
+            gated_tools: Vec::new(),
             connected: true,
         }];
         let body = build(&ctx_with(&integrations, &[])).unwrap();
@@ -205,6 +263,7 @@ mod tests {
             toolkit: "notion".into(),
             description: "Pages.".into(),
             tools: Vec::new(),
+            gated_tools: Vec::new(),
             connected: false,
         }];
         let body = build(&ctx_with(&integrations, &[])).unwrap();

src/openhuman/agent/agents/orchestrator/prompt.rs

@@ -91,7 +91,10 @@ fn render_delegation_guide(integrations: &[ConnectedIntegration]) -> String {
     }
     let mut out = String::from(
         "## Connected Integrations\n\n\
-         Delegate tasks for these services with `delegate_to_integrations_agent`, passing the toolkit slug as `toolkit`:\n\n",
+         The following services have an active connection. Their tool implementations \
+         live inside the `integrations_agent` sub-agent — NOT in your own tool list. \
+         Delegate with `delegate_to_integrations_agent`, passing the toolkit slug as \
+         `toolkit`:\n\n",
     );
     for ci in connected {
         // Use the same slug canonicalisation as `collect_orchestrator_tools`
@@ -104,6 +107,53 @@ fn render_delegation_guide(integrations: &[ConnectedIntegration]) -> String {
             ci.toolkit, slug, ci.description
         );
     }
+    // CRITICAL behavioural rule. Without this, the orchestrator answers
+    // "can you do X with {toolkit}?" from its training-data priors about
+    // "what gmail/notion/slack usually does", which is consistently a
+    // SUBSET of the real per-toolkit catalogue (no bulk-delete, no
+    // batch-modify, no admin/destructive actions, etc.). The result is a
+    // confident wrong refusal ("nope, I can't delete emails") even when
+    // the action is in the actual tool list. The `integrations_agent`
+    // has the ground-truth tool catalogue (`tools` + `gated_tools`); only
+    // it can answer "can I do X?" honestly. Force-delegate capability
+    // questions, not just task requests.
+    // The cross-chat bullet names the canonical header literal verbatim
+    // so the model knows exactly which block to mistrust. Sourced from
+    // CROSS_CHAT_HEADER (single source of truth) — drift would silently
+    // detune the rule.
+    let cross_chat_header_for_prompt =
+        crate::openhuman::agent::memory_loader::CROSS_CHAT_HEADER.trim_end();
+    let _ = write!(
+        out,
+        "\n### Capability questions about connected toolkits\n\n\
+         Your prior knowledge of \"what a toolkit can do\" is UNRELIABLE — the \
+         real per-toolkit catalogue is wider than the common-knowledge summary \
+         (e.g. Gmail exposes bulk delete, batch modify, thread trash, etc.) and \
+         the user may have enabled scopes that expose further destructive actions. \
+         Therefore:\n\n\
+         - If the user asks **\"can you do X with {{toolkit}}?\"** or \"does \
+         {{toolkit}} support Y?\" for a connected toolkit above, **DO NOT** answer \
+         from priors. **DELEGATE** to `integrations_agent` first and let it \
+         inspect its live tool list (including `gated_tools` behind permission \
+         toggles) before answering.\n\
+         - If the user requests an **action** on a connected toolkit (delete, \
+         move, send, modify, label, etc.), **DELEGATE immediately**. Do not \
+         pre-emptively refuse with \"I can't do that\" — that's a confabulation \
+         unless `integrations_agent` itself has already reported the action as \
+         unavailable.\n\
+         - The only honest \"no\" comes back from a delegation that found the \
+         action neither in the visible `tools` list nor in the `gated_tools` \
+         (permission-toggle) list of the sub-agent.\n\
+         - **Cross-chat context is historical, not authoritative.** If the \
+         `{cross_chat_header_for_prompt}` block contains a past \"I can / can't \
+         do X with {{toolkit}}\" statement, treat it as a snapshot from an \
+         earlier moment. The tool list, connected integrations, and per-toolkit \
+         scope toggles (read / write / admin) can all change between chats — a \
+         past refusal may be stale. Verify against the **current** `## Connected \
+         Integrations` block above and (when in doubt) **DELEGATE** before \
+         quoting any past capability claim. Never echo a stale \"I can't\" \
+         without re-checking.\n\n",
+    );
     tracing::debug!(
         section_len = out.len(),
         "[delegation-guide] section emitted ({} bytes)",
@@ -172,6 +222,7 @@ mod tests {
             toolkit: "gmail".into(),
             description: "Email access.".into(),
             tools: Vec::new(),
+            gated_tools: Vec::new(),
             connected: true,
         }];
         let body = build(&ctx_with(&integrations)).unwrap();
@@ -192,6 +243,7 @@ mod tests {
             toolkit: "gmail".into(),
             description: "Email access.".into(),
             tools: Vec::new(),
+            gated_tools: Vec::new(),
             connected: true,
         }];
         let body = build(&ctx_with(&integrations)).unwrap();
@@ -213,12 +265,14 @@ mod tests {
                 toolkit: "gmail".into(),
                 description: "Email.".into(),
                 tools: Vec::new(),
+                gated_tools: Vec::new(),
                 connected: true,
             },
             ConnectedIntegration {
                 toolkit: "linear".into(),
                 description: "Tracker.".into(),
                 tools: Vec::new(),
+                gated_tools: Vec::new(),
                 connected: false,
             },
         ];
@@ -233,6 +287,7 @@ mod tests {
             toolkit: "linear".into(),
             description: "Tracker.".into(),
             tools: Vec::new(),
+            gated_tools: Vec::new(),
             connected: false,
         }];
         let body = build(&ctx_with(&integrations)).unwrap();

src/openhuman/agent/agents/welcome/prompt.rs

@@ -179,12 +179,14 @@ mod tests {
                 toolkit: "gmail".into(),
                 description: "Email access.".into(),
                 tools: Vec::new(),
+                gated_tools: Vec::new(),
                 connected: true,
             },
             ConnectedIntegration {
                 toolkit: "notion".into(),
                 description: "Pitch during onboarding.".into(),
                 tools: Vec::new(),
+                gated_tools: Vec::new(),
                 connected: false,
             },
         ];

src/openhuman/agent/harness/memory_context.rs

@@ -135,7 +135,12 @@ pub(crate) async fn build_context(
         );
 
         if !cross.is_empty() {
-            context.push_str("[Cross-chat context]\n");
+            // Use the canonical CROSS_CHAT_HEADER from `memory_loader` so
+            // this fallback recall path emits the same literal as the
+            // primary JSONL path, and the orchestrator prompt's
+            // "Capability questions" section that names this header stays
+            // in sync. See CROSS_CHAT_HEADER's doc for the rationale.
+            context.push_str(crate::openhuman::agent::memory_loader::CROSS_CHAT_HEADER);
             for entry in &cross {
                 let prov = entry
                     .session_id
@@ -351,7 +356,7 @@ mod tests {
 
         let context = build_context(&mem, "what database should I use?", 0.4).await;
         assert!(
-            context.contains("[Cross-chat context]"),
+            context.contains(crate::openhuman::agent::memory_loader::CROSS_CHAT_HEADER.trim_end()),
             "expected cross-chat header, got:\n{context}"
         );
         assert!(
@@ -417,7 +422,7 @@ mod tests {
         let mem = MockMemory::new(Vec::new(), Vec::new(), false);
         let context = build_context(&mem, "Postgres", 0.4).await;
         assert!(
-            !context.contains("[Cross-chat context]"),
+            !context.contains(crate::openhuman::agent::memory_loader::CROSS_CHAT_HEADER.trim_end()),
             "no cross-chat hits must produce no header, got:\n{context}"
         );
     }

src/openhuman/agent/harness/subagent_runner/ops.rs

@@ -682,6 +682,12 @@ async fn run_typed_mode(
                     toolkit: cached_integration.toolkit.clone(),
                     description: cached_integration.description.clone(),
                     tools: fresh_actions,
+                    // Inherit the cached gated set: this spawn path only
+                    // refreshes the *visible* (callable) actions from the
+                    // backend; the gated/unlock-hint surface is computed
+                    // by `fetch_connected_integrations_uncached` against
+                    // the user pref and doesn't change per-spawn.
+                    gated_tools: cached_integration.gated_tools.clone(),
                     connected: cached_integration.connected,
                 };
                 let integration = &integration;

src/openhuman/agent/harness/test_support_test.rs

@@ -1562,6 +1562,7 @@ async fn orchestrator_prompt_drives_composio_call_via_delegation_chain() {
         toolkit: "gmail".into(),
         description: "Email send/fetch via Gmail.".into(),
         tools: Vec::new(),
+        gated_tools: Vec::new(),
         connected: true,
     }];
     let ctx = {

src/openhuman/agent/memory_loader.rs

@@ -21,6 +21,25 @@ const PRIOR_CONVERSATION_LIMIT: usize = 3;
 /// do not auto-pollute every fresh chat.
 const PRIOR_CONVERSATION_KEY_PREFIX: &str = "high.";
 
+/// Canonical header for the `[Cross-chat context]` block injected on
+/// every turn that has FTS-surfaced hits from other threads.
+///
+/// The "historical" / "capabilities may have changed since" suffix is
+/// deliberate: it tells the model these snippets are snapshots from
+/// earlier moments and that capability claims (e.g. "I can't delete
+/// emails") may be stale because the tool surface or per-toolkit scope
+/// toggles can change between chats.
+///
+/// Single source of truth — all three call sites bind to this constant
+/// so a wording tweak doesn't drift between (a) `memory_loader.rs`'s
+/// primary JSONL path, (b) `harness/memory_context.rs`'s fallback
+/// recall path, and (c) the orchestrator's "Capability questions"
+/// prompt section that names the header verbatim. Tests assert on this
+/// constant too — see `memory_loader::tests` and
+/// `harness::memory_context::tests`.
+pub const CROSS_CHAT_HEADER: &str =
+    "[Cross-chat context — historical; capabilities may have changed since]\n";
+
 #[async_trait]
 pub trait MemoryLoader: Send + Sync {
     async fn load_context(&self, memory: &dyn Memory, user_message: &str)
@@ -371,11 +390,14 @@ impl MemoryLoader for DefaultMemoryLoader {
             };
             let prov = provenance_tag(&sid);
             if !appended_cross_header {
-                let section = "[Cross-chat context]\n";
-                if context.len() + section.len() > budget {
+                // The header explicitly labels these snippets as historical so
+                // the model down-weights them when answering capability
+                // questions — see CROSS_CHAT_HEADER doc for the rationale and
+                // the cross-module wording contract.
+                if context.len() + CROSS_CHAT_HEADER.len() > budget {
                     break;
                 }
-                context.push_str(section);
+                context.push_str(CROSS_CHAT_HEADER);
                 appended_cross_header = true;
             }
             let line = format!("- [{prov}] {snippet}\n");
@@ -594,7 +616,7 @@ mod tests {
             .await
             .expect("loader must succeed");
         assert!(
-            out.contains("[Cross-chat context]"),
+            out.contains(CROSS_CHAT_HEADER.trim_end()),
             "expected cross-chat header, got:\n{out}"
         );
         assert!(
@@ -669,7 +691,7 @@ mod tests {
             .await
             .expect("loader must succeed");
         assert!(
-            !out.contains("[Cross-chat context]"),
+            !out.contains(CROSS_CHAT_HEADER.trim_end()),
             "no cross-chat hits must produce no header, got:\n{out}"
         );
     }
@@ -741,7 +763,7 @@ mod tests {
             .expect("loader must succeed");
 
         assert!(
-            out.contains("[Cross-chat context]"),
+            out.contains(CROSS_CHAT_HEADER.trim_end()),
             "JSONL primary path must emit the cross-chat header, got:\n{out}"
         );
         assert!(

src/openhuman/agent/prompts/types.rs

@@ -89,6 +89,23 @@ pub struct ConnectedIntegration {
     pub description: String,
     /// Per-action catalogue (only populated when `connected == true`).
     pub tools: Vec<ConnectedIntegrationTool>,
+    /// Per-action catalogue for actions that the toolkit **does** support but
+    /// the user has **not** unlocked via their per-toolkit scope preferences.
+    /// The prompt renderer surfaces these descriptively (name + one-line +
+    /// which scope is missing) so the agent can honestly answer "do you have
+    /// X?" with "yes, but you need to flip the {scope} toggle in
+    /// Connections → {toolkit}" — instead of silently claiming the
+    /// capability doesn't exist (which is what happens when the agent has
+    /// zero awareness of pref-gated actions).
+    ///
+    /// The agent CANNOT directly invoke these (no `parameters` schema is
+    /// exposed; the LLM lacks the function definition) and it cannot flip
+    /// the gating scope itself — there is no agent-callable scope-elevate
+    /// tool. Intended flow: agent sees a gated tool → tells the user what
+    /// it does + names the `unlock_paths` from the data → the user toggles
+    /// the scope in the Connections UI → on the next turn the action
+    /// graduates from `gated_tools` to `tools` and becomes callable.
+    pub gated_tools: Vec<GatedIntegrationTool>,
     /// Whether the user has an active OAuth connection for this
     /// toolkit. When `false`, the toolkit is in the backend allowlist
     /// but no authorization has been completed yet — `tools` is empty
@@ -97,6 +114,38 @@ pub struct ConnectedIntegration {
     pub connected: bool,
 }
 
+/// A toolkit action that exists in the catalog but is currently hidden from
+/// the agent's callable function list because the user's scope preference
+/// for this toolkit does not allow the action's required scope.
+///
+/// Deliberately no `parameters` field: the LLM should NOT be able to construct
+/// a call envelope for a gated tool — it can only describe its existence and
+/// point the user at the unlock path. The agent has no scope-elevate tool;
+/// once the user toggles the gating scope in the Connections UI, the action
+/// moves from `ConnectedIntegration.gated_tools` to `ConnectedIntegration.tools`
+/// on the next prompt rebuild and becomes a real callable function.
+#[derive(Debug, Clone)]
+pub struct GatedIntegrationTool {
+    /// Action slug, e.g. `"GMAIL_BATCH_DELETE_MESSAGES"`.
+    pub name: String,
+    /// One-line description of the action.
+    pub description: String,
+    /// Which scope the user must enable for this action to become callable.
+    /// Lowercase: `"read"`, `"write"`, `"admin"`. The vast majority of gated
+    /// rows are `"admin"` (destructive actions); `"write"` only appears for
+    /// users who have explicitly turned write off, which is unusual.
+    pub required_scope: String,
+    /// Literal lines the agent should show the user, verbatim, when offering
+    /// to unlock this action — one entry per available path (typically: the
+    /// agent-side meta-tool, and the manual UI toggle). Populated at
+    /// partition time in `composio::ops`. The prompt-side rule is "show
+    /// these to the user, don't substitute your own framing" — keeping the
+    /// text in the data (not in the system prompt) lets us tweak wording
+    /// without invalidating the KV-cache prefix and avoids biasing the
+    /// model toward a memorized template that drops options.
+    pub unlock_paths: Vec<String>,
+}
+
 /// A single action available on a connected integration.
 #[derive(Debug, Clone)]
 pub struct ConnectedIntegrationTool {

src/openhuman/composio/ops.rs

@@ -1018,28 +1018,79 @@ pub async fn composio_sync(
     tracing::debug!(
         connection_id = %connection_id,
         reason = reason.as_str(),
-        "[composio] rpc sync"
+        "[composio] rpc sync (spawned)"
     );
+    // Validate synchronously — a bad request (unknown connection / no native
+    // provider for toolkit) must surface to the caller via the RPC error
+    // envelope, not silently inside a spawned task.
     let client = resolve_client(config)?;
     let toolkit = resolve_toolkit_for_connection(&client, connection_id).await?;
-
     let provider = get_provider(&toolkit).ok_or_else(|| {
         format!("[composio] no native provider registered for toolkit '{toolkit}'")
     })?;
-
     let _ = client; // see analogous comment above — drop the pre-baked client (#1710).
+
+    // `provider.sync` walks every page of the upstream API and ingests every
+    // message in-band — on a real prod inbox a healthy run can legitimately
+    // exceed the frontend's 30s `composio_sync` RPC `.await` cap (one
+    // healthy periodic tick is already ~100s for 20 pages / 500 messages).
+    // There is no reason for the UI to block on it: per-source progress is
+    // already exposed via the polled `openhuman.memory_sync_status_list` RPC,
+    // which reads `mem_tree_chunks` directly and therefore reflects the
+    // spawned task's per-message ingest in real time. So we spawn the sync
+    // as a background task and return immediately with a "started" envelope.
+    // The periodic scheduler (`composio::periodic`) already runs
+    // `provider.sync` from inside its own `tokio::spawn` loop — same pattern.
     let ctx = ProviderContext {
         config: Arc::new(config.clone()),
         toolkit: toolkit.clone(),
         connection_id: Some(connection_id.to_string()),
     };
+    let started_at_ms = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0);
+    let toolkit_for_outcome = toolkit.clone();
+    let connection_id_for_log = connection_id.to_string();
 
-    let outcome = provider.sync(&ctx, reason).await.map_err(|e| {
-        report_composio_op_error("sync", &e);
-        format!("[composio] sync({toolkit}) failed: {e}")
-    })?;
+    tokio::spawn(async move {
+        let toolkit_in_task = ctx.toolkit.clone();
+        match provider.sync(&ctx, reason).await {
+            Ok(out) => {
+                tracing::info!(
+                    toolkit = %toolkit_in_task,
+                    connection_id = %connection_id_for_log,
+                    items_ingested = out.items_ingested,
+                    elapsed_ms = out.elapsed_ms(),
+                    "[composio] background sync ok"
+                );
+            }
+            Err(e) => {
+                report_composio_op_error("sync", &e);
+                tracing::warn!(
+                    toolkit = %toolkit_in_task,
+                    connection_id = %connection_id_for_log,
+                    error = %e,
+                    "[composio] background sync failed"
+                );
+            }
+        }
+    });
 
-    let summary = outcome.summary.clone();
+    let summary = format!("composio: {toolkit_for_outcome} sync started (background)");
+    let outcome = SyncOutcome {
+        toolkit: toolkit_for_outcome,
+        connection_id: Some(connection_id.to_string()),
+        reason: reason.as_str().to_string(),
+        items_ingested: 0,
+        started_at_ms,
+        // Sentinel: still running. Frontend should rely on
+        // `memory_sync_status_list` for progress; `finished_at_ms == 0`
+        // means "spawned, not yet complete".
+        finished_at_ms: 0,
+        summary: summary.clone(),
+        details: serde_json::json!({ "status": "started" }),
+    };
     Ok(RpcOutcome::new(outcome, vec![summary]))
 }
 
@@ -1063,7 +1114,9 @@ fn parse_sync_reason(raw: Option<&str>) -> OpResult<SyncReason> {
 
 // ── Prompt integration discovery ────────────────────────────────────
 
-use crate::openhuman::context::prompt::{ConnectedIntegration, ConnectedIntegrationTool};
+use crate::openhuman::context::prompt::{
+    ConnectedIntegration, ConnectedIntegrationTool, GatedIntegrationTool,
+};
 use std::collections::{HashMap, HashSet};
 use std::sync::{LazyLock, RwLock};
 use std::time::{Duration, Instant};
@@ -1632,38 +1685,84 @@ async fn fetch_connected_integrations_uncached(
         // each other's actions. `GMAIL_SEND_EMAIL` matches `gmail_`,
         // not just `gmail`, so siblings stay in their own buckets.
         let action_prefix = format!("{}_", slug.to_uppercase());
-        let tools: Vec<ConnectedIntegrationTool> = if connected {
-            // Apply the same curated-whitelist + user-scope filter the
-            // meta-tool layer uses, so the integrations_agent prompt
-            // only advertises actions the agent is actually allowed to
-            // call. One pref load per toolkit (not per action).
-            let pref = super::providers::load_user_scope_or_default(slug).await;
-            let filtered: Vec<&super::types::ComposioToolSchema> = tools_by_toolkit
-                .iter()
-                .filter(|t| t.function.name.starts_with(&action_prefix))
-                .filter(|t| super::providers::is_action_visible_with_pref(&t.function.name, &pref))
-                .collect();
-            tracing::debug!(
-                toolkit = %slug,
-                kept = filtered.len(),
-                "[composio][scopes] integrations prompt action set"
-            );
-            filtered
-                .into_iter()
-                .map(|t| ConnectedIntegrationTool {
-                    name: t.function.name.clone(),
-                    description: t.function.description.clone().unwrap_or_default(),
-                    parameters: t.function.parameters.clone(),
-                })
-                .collect()
-        } else {
-            Vec::new()
-        };
+        let (tools, gated_tools): (Vec<ConnectedIntegrationTool>, Vec<GatedIntegrationTool>) =
+            if connected {
+                // Apply the same curated-whitelist + user-scope filter the
+                // meta-tool layer uses, so the integrations_agent prompt
+                // only advertises actions the agent is actually allowed to
+                // call. One pref load per toolkit (not per action).
+                //
+                // Actions that the catalog *does* know about but the user's
+                // current scope pref denies are routed into `gated_tools` so
+                // the agent can honestly answer "I have this capability but
+                // it needs the {scope} toggle in Connections → {toolkit}".
+                // The agent cannot flip the scope itself — that's a UI-only
+                // action. Without this gated surface the LLM has no way to
+                // know the gated action exists at all and will tell the user
+                // "I don't support that" — technically correct about its
+                // callable surface, but misleading about the toolkit.
+                let pref = super::providers::load_user_scope_or_default(slug).await;
+                let mut visible: Vec<ConnectedIntegrationTool> = Vec::new();
+                let mut gated: Vec<GatedIntegrationTool> = Vec::new();
+                for t in tools_by_toolkit
+                    .iter()
+                    .filter(|t| t.function.name.starts_with(&action_prefix))
+                {
+                    if super::providers::is_action_visible_with_pref(&t.function.name, &pref) {
+                        visible.push(ConnectedIntegrationTool {
+                            name: t.function.name.clone(),
+                            description: t.function.description.clone().unwrap_or_default(),
+                            parameters: t.function.parameters.clone(),
+                        });
+                    } else if let Some(required_scope) =
+                        super::providers::curated_scope_for(&t.function.name)
+                    {
+                        // Only surface CURATED actions as `gated` — uncurated
+                        // tools (which fall through to `classify_unknown` and
+                        // happen to land outside the user's pref) are not
+                        // first-class user-facing capabilities, and listing
+                        // them would clutter the prompt with internal slugs.
+                        // Deliberately NO `parameters` field: the LLM should
+                        // not be able to construct a call envelope; it can
+                        // only describe + point at the unlock path.
+                        // Ship the unlock path as data — single path today
+                        // (the Connections UI toggle). The agent does NOT
+                        // have a tool to flip scopes; that capability was
+                        // removed because LLM-mediated scope elevation made
+                        // the safety contract depend on model behavior and
+                        // was a soft gate the model could route around. If
+                        // more unlock paths exist in future (per-action
+                        // approval modal, time-boxed elevation, etc.) they
+                        // land here and the prompt renderer picks them up.
+                        let scope_str = required_scope.as_str();
+                        let unlock_paths = vec![format!(
+                            "the user enables it themselves in \
+                             Connections → {slug} → {scope_str}"
+                        )];
+                        gated.push(GatedIntegrationTool {
+                            name: t.function.name.clone(),
+                            description: t.function.description.clone().unwrap_or_default(),
+                            required_scope: scope_str.to_string(),
+                            unlock_paths,
+                        });
+                    }
+                }
+                tracing::debug!(
+                    toolkit = %slug,
+                    visible = visible.len(),
+                    gated = gated.len(),
+                    "[composio][scopes] integrations prompt action set"
+                );
+                (visible, gated)
+            } else {
+                (Vec::new(), Vec::new())
+            };
 
         integrations.push(ConnectedIntegration {
             toolkit: slug.clone(),
             description: toolkit_description(slug).to_string(),
             tools,
+            gated_tools,
             connected,
         });
     }

src/openhuman/composio/ops_test.rs

@@ -593,23 +593,54 @@ async fn composio_sync_gmail_via_mock_archives_raw_email_and_updates_outcome() {
 
     assert_eq!(outcome.value.toolkit, "gmail");
     assert_eq!(outcome.value.connection_id.as_deref(), Some("c1"));
-    assert_eq!(outcome.value.items_ingested, 1);
-    assert!(outcome.value.summary.contains("persisted 1 new"));
+    // composio_sync is now spawn-and-return: the immediate envelope is a
+    // "started" sentinel, and the actual ingestion runs on a detached
+    // tokio task. items_ingested == 0 / finished_at_ms == 0 / summary
+    // contains "started" are the contract of that sentinel.
+    assert_eq!(
+        outcome.value.items_ingested, 0,
+        "spawn-and-return: items_ingested on the immediate envelope is a 'started' sentinel, not a final count"
+    );
+    assert_eq!(
+        outcome.value.finished_at_ms, 0,
+        "spawn-and-return: finished_at_ms == 0 means 'task spawned, not yet complete'"
+    );
+    assert!(
+        outcome.value.summary.contains("started"),
+        "expected spawn-and-return summary to mention 'started', got: {}",
+        outcome.value.summary
+    );
 
-    let chunks = list_chunks_rpc(
-        &config,
-        ListChunksRequest {
-            source_kind: Some("email".to_string()),
-            source_id: Some("gmail:pilot-at-example-dot-com".to_string()),
-            limit: Some(10),
-            ..Default::default()
-        },
-    )
-    .await
-    .unwrap()
-    .value
-    .chunks;
-    assert_eq!(chunks.len(), 1, "expected one ingested Gmail chunk");
+    // Poll for the spawned ingest task to drain. The mock backend is
+    // local + in-memory, so this normally lands in well under a second.
+    let chunks = {
+        let mut chunks = Vec::new();
+        for _ in 0..50 {
+            chunks = list_chunks_rpc(
+                &config,
+                ListChunksRequest {
+                    source_kind: Some("email".to_string()),
+                    source_id: Some("gmail:pilot-at-example-dot-com".to_string()),
+                    limit: Some(10),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap()
+            .value
+            .chunks;
+            if !chunks.is_empty() {
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        }
+        chunks
+    };
+    assert_eq!(
+        chunks.len(),
+        1,
+        "expected one ingested Gmail chunk after spawned task drains"
+    );
     assert!(
         chunks[0].content.contains("Phoenix launch canary"),
         "chunk content missing mock email subject: {}",
@@ -839,6 +870,7 @@ fn integration(toolkit: &str, connected: bool) -> ConnectedIntegration {
         toolkit: toolkit.to_string(),
         description: String::new(),
         tools: Vec::new(),
+        gated_tools: Vec::new(),
         connected,
     }
 }

src/openhuman/composio/providers/mod.rs

@@ -34,6 +34,7 @@
 
 mod descriptions;
 pub(crate) mod helpers;
+mod scope_lookup;
 pub mod tool_scope;
 mod traits;
 mod types;
@@ -212,6 +213,7 @@ pub(crate) use helpers::pick_str;
 pub use registry::{
     all_providers, get_provider, init_default_providers, register_provider, ProviderArc,
 };
+pub use scope_lookup::{curated_scope_for, toolkit_has_scope};
 pub use tool_scope::{classify_unknown, find_curated, toolkit_from_slug, CuratedTool, ToolScope};
 pub use traits::ComposioProvider;
 pub use types::{ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason};
@@ -283,6 +285,9 @@ mod tests {
         assert_eq!(back, SyncReason::ConnectionCreated);
     }
 
+    // Note: `toolkit_has_scope` tests now live in `scope_lookup.rs`
+    // alongside the implementation.
+
     #[test]
     fn capability_matrix_distinguishes_native_from_catalog_only_toolkits() {
         let matrix = capability_matrix();

src/openhuman/composio/providers/scope_lookup.rs

@@ -0,0 +1,78 @@
+//! Scope-lookup operational helpers for the curated tool catalogs.
+//!
+//! Lives in a sibling module (extracted from the formerly-thick
+//! `providers/mod.rs`) to keep the module entrypoint export-focused —
+//! matches the project rule "keep mod.rs light; operational logic in
+//! ops.rs / store.rs / types.rs" from CLAUDE.md.
+//!
+//! - [`curated_scope_for`] answers "what scope does this action slug
+//!   require?" — used by `composio::ops` to render `gated_tools`
+//!   unlock hints.
+//! - [`toolkit_has_scope`] answers "does this toolkit have any
+//!   actions at the given scope?" — currently used by tests; intended
+//!   for future UI hints (grey-out a toggle that unlocks nothing).
+//!
+//! Both walk the native provider catalog first, then fall back to the
+//! static `catalog_for_toolkit` map — so the answers match what
+//! [`super::is_action_visible_with_pref`] would gate against.
+
+use super::tool_scope::{find_curated, toolkit_from_slug, ToolScope};
+use super::{catalog_for_toolkit, get_provider};
+
+/// Look up the curated scope for `slug` if it appears in any registered
+/// catalog (native provider's `curated_tools()` first, then the fallback
+/// catalog from [`super::catalog_for_toolkit`]). Returns `None` for
+/// genuinely uncurated slugs — callers that want a defensible heuristic
+/// for those should fall back to [`super::classify_unknown`] explicitly.
+///
+/// Sibling of [`super::is_action_visible_with_pref`]: that one decides
+/// "visible?", this one returns "what scope is required?" so callers
+/// (e.g. the `gated_tools` partition in
+/// `composio::ops::fetch_connected_integrations`) can render a useful
+/// unlock hint to the agent without re-doing the catalog walk.
+pub fn curated_scope_for(slug: &str) -> Option<ToolScope> {
+    let toolkit = toolkit_from_slug(slug)?;
+    let catalog = get_provider(&toolkit)
+        .and_then(|p| p.curated_tools())
+        .or_else(|| catalog_for_toolkit(&toolkit))?;
+    find_curated(catalog, slug).map(|c| c.scope)
+}
+
+/// Does any curated action for `toolkit` require `scope`?
+///
+/// Currently used by this module's tests only (added when the
+/// now-removed `composio_enable_scope` meta-tool needed a no-op
+/// short-circuit). Kept because the same probe is useful any time we
+/// ask "would flipping the {scope} bit unlock anything in this
+/// toolkit?" — e.g. a UI hint that greys out a toggle with no effect.
+///
+/// Walks both the native provider catalog and the fallback
+/// [`super::catalog_for_toolkit`] so the answer matches what
+/// [`super::is_action_visible_with_pref`] would gate against.
+pub fn toolkit_has_scope(toolkit: &str, scope: ToolScope) -> bool {
+    let catalog = get_provider(toolkit)
+        .and_then(|p| p.curated_tools())
+        .or_else(|| catalog_for_toolkit(toolkit));
+    match catalog {
+        Some(cat) => cat.iter().any(|t| t.scope == scope),
+        None => false,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn toolkit_has_scope_distinguishes_gated_from_ungated_scopes() {
+        // gmail catalog includes destructive verbs (delete / trash /
+        // batch_delete), so admin-gating actually unlocks something.
+        assert!(toolkit_has_scope("gmail", ToolScope::Admin));
+        assert!(toolkit_has_scope("gmail", ToolScope::Read));
+        assert!(toolkit_has_scope("gmail", ToolScope::Write));
+        // Case-insensitive toolkit slug → still routes to the catalog.
+        assert!(toolkit_has_scope("GMAIL", ToolScope::Admin));
+        // Unknown toolkit → no catalog → no scope is "gating" anything.
+        assert!(!toolkit_has_scope("nonexistent-toolkit", ToolScope::Admin));
+    }
+}

src/openhuman/composio/tools.rs

@@ -15,6 +15,9 @@
 //! | `composio_list_tools`         | Discover available action slugs + their JSON schemas        |
 //! | `composio_execute`            | Run a Composio action with `{tool, arguments}`              |
 //!
+//! Scope elevation (read/write/admin) is deliberately NOT an agent tool;
+//! the user must toggle it themselves in the Connections UI.
+//!
 //! The agent loop is expected to chain `composio_list_tools` →
 //! `composio_execute` when it needs to use a new action. The full schema
 //! is returned in `composio_list_tools`'s output so the model can pick
@@ -264,15 +267,24 @@ fn render_tools_markdown(resp: &super::types::ComposioToolsResponse) -> String {
 // translation contract.
 
 /// Format a user-facing error message for a scope-blocked execution.
+///
+/// Embeds the unlock path in the error itself so the agent reads the
+/// instruction straight off the tool response — same policy-in-data
+/// approach as the `gated_tools` surface. Only ONE path: the user
+/// toggles the scope in the Connections UI. The agent has no tool to
+/// flip scopes (see the note above the removed `ComposioEnableScopeTool`
+/// for why) — it can only describe the gate and point at the UI.
 fn scope_error_message(slug: &str, scope: ToolScope, pref: UserScopePref) -> String {
+    let toolkit = toolkit_from_slug(slug).unwrap_or_default();
+    let scope_str = scope.as_str();
     format!(
-        "composio_execute: action `{slug}` is classified `{}` and is disabled in your \
-         current scope preferences (read={}, write={}, admin={}). Update the toolkit's \
-         scope preference (composio_set_user_scopes) to enable this category.",
-        scope.as_str(),
-        pref.read,
-        pref.write,
-        pref.admin,
+        "composio_execute: action `{slug}` is classified `{scope_str}` and is \
+         disabled in the user's current scope preferences for `{toolkit}` \
+         (read={}, write={}, admin={}). Tell the user this action requires the \
+         `{scope_str}` scope and they can enable it themselves in \
+         **Connections → {toolkit} → {scope_str}**. Do not claim you can flip \
+         it — you cannot.",
+        pref.read, pref.write, pref.admin,
     )
 }
 
@@ -849,7 +861,33 @@ impl Tool for ComposioExecuteTool {
             ));
         }
         let arguments = args.get("arguments").cloned();
-        tracing::debug!(tool = %tool, "[composio] tool execute.execute");
+        let connection_id = args
+            .get("connection_id")
+            .and_then(|v| v.as_str())
+            .unwrap_or("");
+        // INFO-level entry log — visible without bumping log level. Logs
+        // ARG KEYS only by default so traces don't leak email bodies / PII;
+        // the full arg JSON goes through a paired DEBUG log below for deep
+        // debugging. Together these let `grep "[composio][execute]"` show
+        // the full agent→backend audit trail at default verbosity.
+        let arg_keys: Vec<&str> = arguments
+            .as_ref()
+            .and_then(|v| v.as_object())
+            .map(|obj| obj.keys().map(|k| k.as_str()).collect())
+            .unwrap_or_default();
+        tracing::info!(
+            tool = %tool,
+            connection_id = %connection_id,
+            arg_keys = ?arg_keys,
+            "[composio][execute] >> dispatch"
+        );
+        if let Some(ref args_json) = arguments {
+            tracing::debug!(
+                tool = %tool,
+                args = %args_json,
+                "[composio][execute] >> full args (DEBUG)"
+            );
+        }
 
         // Agent-level sandbox gate (issue #685) — applies on top of the
         // user's scope preference below. When the currently-executing
@@ -962,6 +1000,19 @@ impl Tool for ComposioExecuteTool {
         let elapsed_ms = started.elapsed().as_millis() as u64;
         match res {
             Ok(resp) => {
+                tracing::info!(
+                    tool = %tool,
+                    successful = resp.successful,
+                    error = ?resp.error,
+                    elapsed_ms,
+                    cost_usd = resp.cost_usd,
+                    "[composio][execute] << result"
+                );
+                tracing::debug!(
+                    tool = %tool,
+                    response = ?resp,
+                    "[composio][execute] << full response (DEBUG)"
+                );
                 crate::core::event_bus::publish_global(
                     crate::core::event_bus::DomainEvent::ComposioActionExecuted {
                         tool: tool.clone(),
@@ -992,6 +1043,12 @@ impl Tool for ComposioExecuteTool {
                 Ok(ToolResult::success(body))
             }
             Err(e) => {
+                tracing::warn!(
+                    tool = %tool,
+                    error = %e,
+                    elapsed_ms,
+                    "[composio][execute] << dispatch error"
+                );
                 crate::core::event_bus::publish_global(
                     crate::core::event_bus::DomainEvent::ComposioActionExecuted {
                         tool: tool.clone(),
@@ -1007,6 +1064,21 @@ impl Tool for ComposioExecuteTool {
     }
 }
 
+// NOTE: A `composio_enable_scope` agent-callable meta-tool used to live
+// here. It was removed deliberately: scope elevation is a
+// security-sensitive, cross-session state change that unlocks
+// destructive actions, and putting that flip behind LLM-mediated
+// "user consent" both (a) made the safety contract depend on model
+// behavior — the weakest place for it — and (b) was a soft gate the
+// model could route around (e.g. trash-via-label). The user must
+// toggle scopes themselves in **Connections → {toolkit} → {scope}
+// row**; the agent only describes the gated capability and points at
+// that UI path.
+//
+// Two surfaces name this policy; keep them in sync:
+//   - `GatedIntegrationTool.unlock_paths` populated in `composio::ops`
+//   - `scope_error_message` returned from `composio_execute` blocks
+
 // ── Bulk registration helper ────────────────────────────────────────
 
 /// Build the full set of composio agent tools when the integrations
@@ -1040,6 +1112,9 @@ pub fn all_composio_agent_tools(config: &crate::openhuman::config::Config) -> Ve
         Box::new(ComposioAuthorizeTool::new(config_arc.clone())),
         Box::new(ComposioListToolsTool::new(config_arc.clone())),
         Box::new(ComposioExecuteTool::new(config_arc)),
+        // Pref-elevation is intentionally NOT an agent-callable tool;
+        // the user must flip it themselves in the Connections UI.
+        // See the long comment above the (removed) ComposioEnableScopeTool.
     ];
     tracing::debug!(count = tools.len(), "[composio] agent tools registered");
     tools

src/openhuman/composio/tools_tests.rs

@@ -239,7 +239,10 @@ fn agent_tools_register_when_backend_signed_in() {
     assert_eq!(
         tools.len(),
         5,
-        "backend session present → all 5 generic composio agent tools should register"
+        "backend session present → all 5 generic composio agent tools should register \
+         (list_toolkits, list_connections, authorize, list_tools, execute). Scope \
+         elevation is intentionally NOT exposed as an agent tool — the user must \
+         flip scopes themselves in the Connections UI."
     );
 }
 
@@ -249,7 +252,9 @@ fn agent_tools_register_when_direct_mode_with_stored_key_and_no_backend_session(
     // user with a working personal Composio API key was getting `0`
     // tools registered because the gate hard-bound to
     // `build_composio_client` (backend-only). With the mode-aware probe
-    // in place this now correctly returns the full 5 generic tools.
+    // in place this now correctly returns the full generic tool set
+    // (5 tools: list_toolkits, list_connections, authorize, list_tools,
+    // execute). Scope elevation is not an agent tool — UI-only.
     let tmp = tempfile::tempdir().expect("tempdir");
     let mut config = crate::openhuman::config::Config::default();
     config.config_path = tmp.path().join("config.toml");

src/openhuman/tools/orchestrator_tools.rs

@@ -320,6 +320,7 @@ mod tests {
             toolkit: toolkit.into(),
             description: description.into(),
             tools: vec![],
+            gated_tools: vec![],
             connected: true,
         }
     }
@@ -462,6 +463,7 @@ mod tests {
                 toolkit: "github".into(),
                 description: "GitHub access.".into(),
                 tools: vec![],
+                gated_tools: vec![],
                 connected: false, // not connected — must not appear in the enum
             },
             integration("notion", "Read and write pages."),
@@ -535,6 +537,7 @@ mod tests {
                 toolkit: "Brand.New".into(),
                 description: "   ".into(),
                 tools: vec![],
+                gated_tools: vec![],
                 connected: true,
             },
             integration("gmail", "Email."),