From 1e3ecc55eeab6bf2d2a8d6ae4f622a77e5b68a13 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Thu, 21 May 2026 00:13:56 +0530 Subject: [PATCH] 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) --- .../agent/agents/integrations_agent/prompt.rs | 59 ++++++ .../agent/agents/orchestrator/prompt.rs | 57 +++++- src/openhuman/agent/agents/welcome/prompt.rs | 2 + src/openhuman/agent/harness/memory_context.rs | 11 +- .../agent/harness/subagent_runner/ops.rs | 6 + .../agent/harness/test_support_test.rs | 1 + src/openhuman/agent/memory_loader.rs | 34 +++- src/openhuman/agent/prompts/types.rs | 49 +++++ src/openhuman/composio/ops.rs | 171 ++++++++++++++---- src/openhuman/composio/ops_test.rs | 64 +++++-- src/openhuman/composio/providers/mod.rs | 5 + .../composio/providers/scope_lookup.rs | 78 ++++++++ src/openhuman/composio/tools.rs | 91 +++++++++- src/openhuman/composio/tools_tests.rs | 9 +- src/openhuman/tools/orchestrator_tools.rs | 3 + 15 files changed, 568 insertions(+), 72 deletions(-) create mode 100644 src/openhuman/composio/providers/scope_lookup.rs diff --git a/src/openhuman/agent/agents/integrations_agent/prompt.rs b/src/openhuman/agent/agents/integrations_agent/prompt.rs index 4ff57a7e3..2913d220f 100644 --- a/src/openhuman/agent/agents/integrations_agent/prompt.rs +++ b/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 >.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(); diff --git a/src/openhuman/agent/agents/orchestrator/prompt.rs b/src/openhuman/agent/agents/orchestrator/prompt.rs index d8688cd9d..874ee2718 100644 --- a/src/openhuman/agent/agents/orchestrator/prompt.rs +++ b/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(); diff --git a/src/openhuman/agent/agents/welcome/prompt.rs b/src/openhuman/agent/agents/welcome/prompt.rs index 28e6da164..7c3366115 100644 --- a/src/openhuman/agent/agents/welcome/prompt.rs +++ b/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, }, ]; diff --git a/src/openhuman/agent/harness/memory_context.rs b/src/openhuman/agent/harness/memory_context.rs index 43b5a64a1..620734209 100644 --- a/src/openhuman/agent/harness/memory_context.rs +++ b/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}" ); } diff --git a/src/openhuman/agent/harness/subagent_runner/ops.rs b/src/openhuman/agent/harness/subagent_runner/ops.rs index eee38e4e5..7be0d0d72 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops.rs +++ b/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; diff --git a/src/openhuman/agent/harness/test_support_test.rs b/src/openhuman/agent/harness/test_support_test.rs index 74732644b..424c59566 100644 --- a/src/openhuman/agent/harness/test_support_test.rs +++ b/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 = { diff --git a/src/openhuman/agent/memory_loader.rs b/src/openhuman/agent/memory_loader.rs index 7af93bc20..fd656b9c2 100644 --- a/src/openhuman/agent/memory_loader.rs +++ b/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!( diff --git a/src/openhuman/agent/prompts/types.rs b/src/openhuman/agent/prompts/types.rs index 9b0168de1..d920c2d48 100644 --- a/src/openhuman/agent/prompts/types.rs +++ b/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, + /// 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, /// 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, +} + /// A single action available on a connected integration. #[derive(Debug, Clone)] pub struct ConnectedIntegrationTool { diff --git a/src/openhuman/composio/ops.rs b/src/openhuman/composio/ops.rs index e1c939b47..e2f758633 100644 --- a/src/openhuman/composio/ops.rs +++ b/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 { // ── 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 = 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, Vec) = + 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 = Vec::new(); + let mut gated: Vec = 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, }); } diff --git a/src/openhuman/composio/ops_test.rs b/src/openhuman/composio/ops_test.rs index 8cdc0266b..bc365380d 100644 --- a/src/openhuman/composio/ops_test.rs +++ b/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, } } diff --git a/src/openhuman/composio/providers/mod.rs b/src/openhuman/composio/providers/mod.rs index 620137743..7b854a4c7 100644 --- a/src/openhuman/composio/providers/mod.rs +++ b/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(); diff --git a/src/openhuman/composio/providers/scope_lookup.rs b/src/openhuman/composio/providers/scope_lookup.rs new file mode 100644 index 000000000..e3bae0280 --- /dev/null +++ b/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 { + 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)); + } +} diff --git a/src/openhuman/composio/tools.rs b/src/openhuman/composio/tools.rs index 5c98b4a49..f227f1c11 100644 --- a/src/openhuman/composio/tools.rs +++ b/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 diff --git a/src/openhuman/composio/tools_tests.rs b/src/openhuman/composio/tools_tests.rs index 219709eb5..4174395dc 100644 --- a/src/openhuman/composio/tools_tests.rs +++ b/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"); diff --git a/src/openhuman/tools/orchestrator_tools.rs b/src/openhuman/tools/orchestrator_tools.rs index 6cbf0bb46..1ed23f4ac 100644 --- a/src/openhuman/tools/orchestrator_tools.rs +++ b/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."),