From d80b1e3fadf8c3ab951bceb07f125f64583d5b0c Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Thu, 16 Jul 2026 19:03:41 +0530 Subject: [PATCH 1/4] fix(composio): gate per-action tools on their full contract (#4853) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-action Composio tools handed to integrations_agent are built from the thin list_tools schema (often {"type":"object"} with no field descriptions), so the model composes calls before the action's real contract is in context and guesses argument formats — most visibly sending unquoted Gmail queries that return zero results. Add a per-turn contract gate: on the first call to an action, surface its FULL live input schema/description (from the cached live toolkit catalog) as a recoverable tool error and let the retry execute. Degrades to a normal execute whenever the contract can't be resolved, so an unconfigured/offline client never blocks the action. Wired into the per-action surface; the dispatcher, MCP, and Workflow surfaces are follow-up. --- src/openhuman/composio/action_tool.rs | 99 ++++++++++ src/openhuman/composio/contract_gate.rs | 181 ++++++++++++++++++ src/openhuman/composio/contract_gate_tests.rs | 134 +++++++++++++ src/openhuman/composio/mod.rs | 1 + 4 files changed, 415 insertions(+) create mode 100644 src/openhuman/composio/contract_gate.rs create mode 100644 src/openhuman/composio/contract_gate_tests.rs diff --git a/src/openhuman/composio/action_tool.rs b/src/openhuman/composio/action_tool.rs index 867a4ae8b7..23a0e72bff 100644 --- a/src/openhuman/composio/action_tool.rs +++ b/src/openhuman/composio/action_tool.rs @@ -67,6 +67,13 @@ pub struct ComposioActionTool { /// Composio connection. Used when the sub-agent is spawned for a /// particular account (e.g. "send from my work Gmail"). connection_id: Option, + /// Per-turn contract gate (#4853). On the first call to this action the + /// gate surfaces the action's FULL live input schema/description so the + /// model composes well-formed arguments (e.g. correctly-quoted Gmail + /// queries) instead of guessing from the thin spawn-time schema; the retry + /// executes normally. Held per tool instance, which lives for one + /// `integrations_agent` spawn, so "seen" is scoped to that turn. + gate: super::contract_gate::ContractGate, } impl ComposioActionTool { @@ -93,6 +100,7 @@ impl ComposioActionTool { description, parameters, connection_id, + gate: super::contract_gate::ContractGate::new(), } } } @@ -184,6 +192,29 @@ impl Tool for ComposioActionTool { } } + // Contract gate (#4853): the per-action tool is built from the thin + // spawn-time `list_tools` schema (often `{"type":"object"}` with no + // field descriptions), so the model guesses argument formats — most + // visibly sending unquoted Gmail `query` strings that return zero + // results. On the first call this turn, surface the action's FULL live + // contract (input schema + description) as a recoverable tool error and + // let the retry — now with the schema in context — execute. Degrades to + // a normal execute whenever the contract can't be resolved (see + // `contract_gate::consult`), so an unconfigured/offline client never + // blocks the action. + match super::contract_gate::consult(&self.gate, self.config.as_ref(), &self.action_name) + .await + { + super::contract_gate::GateDecision::Surface(contract) => { + tracing::info!( + tool = %self.action_name, + "[composio][contract-gate] returning full contract before first execute" + ); + return Ok(ToolResult::error(contract)); + } + super::contract_gate::GateDecision::Proceed => {} + } + // Inject `timeZone` / `singleEvents` defaults for Google // Calendar list slugs (issue #1714). The per-action surface is // the spawn-time tool an integrations sub-agent picks when it @@ -489,6 +520,74 @@ mod tests { ); } + #[tokio::test] + async fn contract_gate_surfaces_full_contract_then_proceeds_on_retry() { + // Regression for #4853: the FIRST per-action execute this turn must + // return the action's FULL live contract (so the model composes a + // well-formed query) instead of running with the thin spawn-time + // schema; the retry then proceeds to real dispatch. A unique toolkit is + // seeded so this is deterministic and never touches the network. + use crate::openhuman::config::TEST_ENV_LOCK; + use crate::openhuman::tinyflows::caps::{seed_live_catalog_cache, ToolContract}; + let _env_guard = TEST_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + + let toolkit = "cgateexec"; + let slug = "CGATEEXEC_FETCH_ITEMS"; + seed_live_catalog_cache( + toolkit, + vec![ToolContract { + slug: slug.to_string(), + toolkit: toolkit.to_string(), + description: Some("Search items. Quote multi-word phrases.".to_string()), + required_args: vec!["query".to_string()], + input_schema: Some(serde_json::json!({ + "type": "object", + "properties": { "query": { "type": "string" } }, + "required": ["query"] + })), + output_fields: Vec::new(), + output_schema: None, + primary_array_path: None, + is_curated: false, + }], + ); + + let tmp = tempfile::tempdir().expect("tempdir"); + let _workspace_guard = WorkspaceEnvGuard::set(tmp.path()); + let mut config = Config::default(); + config.config_path = tmp.path().join("config.toml"); + config.workspace_dir = tmp.path().join("workspace"); + config.save().await.expect("save fake config to disk"); + + let t = ComposioActionTool::new( + Arc::new(config), + slug.to_string(), + "search items".to_string(), + None, + ); + + // First call: gate surfaces the contract (recoverable tool error). + let first = t.execute(serde_json::json!({})).await.unwrap(); + assert!( + first.is_error, + "first call must surface a recoverable error" + ); + let first_msg = error_text(&first); + assert!( + first_msg.contains("Input JSON schema"), + "first call must carry the full contract, got: {first_msg}" + ); + + // Retry: gate proceeds; dispatch fails downstream (no session token) but + // crucially NOT with the contract text — proving the gate did not block. + let second = t.execute(serde_json::json!({})).await.unwrap(); + let second_msg = error_text(&second); + assert!( + !second_msg.contains("Input JSON schema"), + "retry must proceed past the gate to real dispatch, got: {second_msg}" + ); + } + // ── Factory routing (#1710) ────────────────────────────────────── // // Regression coverage for the bug fix: `ComposioActionTool` now diff --git a/src/openhuman/composio/contract_gate.rs b/src/openhuman/composio/contract_gate.rs new file mode 100644 index 0000000000..99827ef921 --- /dev/null +++ b/src/openhuman/composio/contract_gate.rs @@ -0,0 +1,181 @@ +//! Contract gate for late-bound Composio actions (#4853). +//! +//! Per-action Composio tools handed to `integrations_agent` are built from +//! the lightweight `list_tools` response — a one-line description with a +//! parameter schema that is often thin or absent (see +//! `fetch_toolkit_actions`, consumed in the +//! sub-agent runner). The model therefore composes calls before the action's +//! FULL contract is in context and guesses argument formats — most visibly, it +//! sends Gmail `query` strings without the quoting Gmail search syntax requires, +//! so `GMAIL_FETCH_EMAILS` returns zero results. +//! +//! The gate makes the full contract enter context BEFORE execution: on the +//! first call to an action this turn, if a fuller live contract is available +//! (via the cached `fetch_live_toolkit_catalog`), it is returned as a +//! recoverable tool error instead of executing. The retry — now with the +//! schema/description in context — proceeds normally. This mirrors the +//! discover-then-call discipline the generic `composio_execute` dispatcher +//! already expects (`composio_list_tools` → `composio_execute`), but enforces +//! it on the per-action surface where the model never sees the full schema. +//! +//! Scope: this pass gates the per-action Composio surface +//! ([`super::action_tool::ComposioActionTool`]). Generalising the same gate to +//! the `composio_execute` dispatcher, the MCP bridges, and the Workflow +//! dispatchers — plus resetting the per-turn state on context compaction — is +//! tracked as follow-up (a shared `ToolMiddleware` at the turn-harness seam is +//! the natural home; see the PR description). + +use std::collections::HashSet; +use std::sync::Mutex; + +use crate::openhuman::composio::providers::toolkit_from_slug; +use crate::openhuman::config::Config; +use crate::openhuman::tinyflows::caps::{fetch_live_toolkit_catalog, ToolContract}; + +/// Per-agent-turn record of which action contracts have already been surfaced +/// to the model, so the gate blocks a given action at most once per turn. +/// +/// One [`ContractGate`] is held per [`super::action_tool::ComposioActionTool`] +/// instance; those tools are constructed fresh per `integrations_agent` spawn +/// and live for that spawn's tool loop, so "seen" is scoped to the turn without +/// any task-local plumbing. Interior-mutable so the gate can record state +/// through the tool's `&self` `execute`. +#[derive(Default)] +pub struct ContractGate { + seen: Mutex>, +} + +impl ContractGate { + pub fn new() -> Self { + Self::default() + } + + /// Insert `slug` (normalised to upper-case) into the seen-set. Returns + /// `true` when it was NOT already present — i.e. this is the first time the + /// gate has been consulted for this action this turn. + /// + /// The lock is taken and released entirely within this call, so no guard is + /// held across the caller's later `await`. + fn mark_seen(&self, slug: &str) -> bool { + let mut guard = self + .seen + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + guard.insert(slug.to_ascii_uppercase()) + } +} + +/// Outcome of consulting the gate for one action call. +pub enum GateDecision { + /// Return this text to the model as a recoverable tool error; the model + /// retries with the contract in context. + Surface(String), + /// Execute the action normally. + Proceed, +} + +/// Consult the gate before executing `action_slug`. +/// +/// On the FIRST consult for a slug this turn, if a fuller live contract can be +/// resolved, returns [`GateDecision::Surface`] with the formatted contract and +/// marks the slug seen. Every later consult — and any consult where no live +/// contract is available (unconfigured client, unknown action, network miss) — +/// returns [`GateDecision::Proceed`], so the gate never blocks an action more +/// than once and never blocks when it cannot help. +pub async fn consult(gate: &ContractGate, config: &Config, action_slug: &str) -> GateDecision { + // Mark first (releasing the lock) so the retry — and any concurrent + // sibling call — proceeds even if the contract lookup below is slow. + let first_time = gate.mark_seen(action_slug); + if !first_time { + tracing::debug!( + target: "composio", + slug = %action_slug, + "[composio][contract-gate] contract already surfaced this turn; proceeding" + ); + return GateDecision::Proceed; + } + + match lookup_contract(config, action_slug).await { + Some(contract) => { + tracing::debug!( + target: "composio", + slug = %action_slug, + has_input_schema = contract.input_schema.is_some(), + required_arg_count = contract.required_args.len(), + "[composio][contract-gate] surfacing full contract before first execute" + ); + GateDecision::Surface(format_contract(action_slug, &contract)) + } + None => { + tracing::debug!( + target: "composio", + slug = %action_slug, + "[composio][contract-gate] no live contract available; proceeding without gating" + ); + GateDecision::Proceed + } + } +} + +/// Resolve the full live contract for `action_slug` from the process-cached +/// live toolkit catalog. Returns `None` when the toolkit can't be derived, the +/// catalog can't be fetched (unconfigured / offline — `fetch_live_toolkit_catalog` +/// degrades to `None`), or the action isn't in it. +async fn lookup_contract(config: &Config, action_slug: &str) -> Option { + let toolkit = toolkit_from_slug(action_slug)?; + let contracts = fetch_live_toolkit_catalog(config, &toolkit).await?; + contracts + .into_iter() + .find(|c| c.slug.eq_ignore_ascii_case(action_slug)) +} + +/// Render the contract into a compact instruction for the model. Contains only +/// the provider's own action description + JSON schema — no user data / PII. +fn format_contract(action_slug: &str, contract: &ToolContract) -> String { + let mut out = format!( + "Before running `{action_slug}`, read its full contract below and then re-issue \ + the call with arguments that match it exactly.\n\n" + ); + + if let Some(desc) = contract + .description + .as_deref() + .map(str::trim) + .filter(|d| !d.is_empty()) + { + out.push_str("Description:\n"); + out.push_str(desc); + out.push_str("\n\n"); + } + + match contract.input_schema.as_ref() { + Some(schema) => { + let pretty = + serde_json::to_string_pretty(schema).unwrap_or_else(|_| schema.to_string()); + out.push_str("Input JSON schema:\n"); + out.push_str(&pretty); + out.push('\n'); + } + None => out.push_str("Input JSON schema: not published by the provider for this action.\n"), + } + + if !contract.required_args.is_empty() { + out.push_str(&format!( + "\nRequired arguments: {}\n", + contract.required_args.join(", ") + )); + } + + out.push_str( + "\nCompose every argument to match this schema and any format rules in the \ + description. Text-search fields in particular often require the provider's exact \ + query syntax (for example, Gmail needs multi-word phrases quoted, like \ + subject:\"quarterly report\"). Then call the action again with the corrected \ + arguments.", + ); + out +} + +#[cfg(test)] +#[path = "contract_gate_tests.rs"] +mod tests; diff --git a/src/openhuman/composio/contract_gate_tests.rs b/src/openhuman/composio/contract_gate_tests.rs new file mode 100644 index 0000000000..f391281d3d --- /dev/null +++ b/src/openhuman/composio/contract_gate_tests.rs @@ -0,0 +1,134 @@ +//! Unit tests for the Composio contract gate (#4853). +//! +//! These exercise the gate purely against the process-level live-catalog cache +//! (seeded via [`seed_live_catalog_cache`]), so no Composio client is built and +//! no network call is made. Each test uses a unique toolkit slug so the shared +//! `LIVE_CATALOG_CACHE` can't cross-contaminate between tests. + +use super::{consult, ContractGate, GateDecision}; +use crate::openhuman::config::Config; +use crate::openhuman::tinyflows::caps::{seed_live_catalog_cache, ToolContract}; + +/// Build a full contract for `slug` in `toolkit` with a `query` input field and +/// a description that spells out the quoting rule — the exact detail the model +/// misses when it only sees the thin spawn-time schema. +fn full_contract(slug: &str, toolkit: &str) -> ToolContract { + ToolContract { + slug: slug.to_string(), + toolkit: toolkit.to_string(), + description: Some( + "Search the mailbox. Multi-word phrases in `query` must be quoted, \ + e.g. subject:\"quarterly report\"." + .to_string(), + ), + required_args: vec!["query".to_string()], + input_schema: Some(serde_json::json!({ + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "Gmail search query (quote multi-word phrases)." + } + }, + "required": ["query"] + })), + output_fields: Vec::new(), + output_schema: None, + primary_array_path: None, + is_curated: false, + } +} + +#[tokio::test] +async fn first_call_surfaces_full_contract_then_retry_proceeds() { + // Toolkit derived from the slug prefix: `GMAILGATE_...` -> `gmailgate`. + let toolkit = "gmailgate"; + let slug = "GMAILGATE_FETCH_EMAILS"; + seed_live_catalog_cache(toolkit, vec![full_contract(slug, toolkit)]); + + let config = Config::default(); + let gate = ContractGate::new(); + + // First call: the gate short-circuits execution and hands back the full + // contract (this is the behaviour that is ABSENT before the fix — the thin + // per-action tool would execute immediately with a guessed query). + match consult(&gate, &config, slug).await { + GateDecision::Surface(message) => { + assert!(message.contains(slug), "contract names the action slug"); + assert!( + message.contains("query"), + "contract carries the input schema" + ); + assert!( + message.contains("Required arguments: query"), + "contract lists required args" + ); + assert!( + message.contains("quoted"), + "contract carries the provider description explaining quoting" + ); + } + GateDecision::Proceed => panic!("first call must surface the contract, not execute"), + } + + // The retry — now with the contract in context — proceeds to execution. + assert!( + matches!(consult(&gate, &config, slug).await, GateDecision::Proceed), + "retry must proceed once the contract has been surfaced this turn" + ); +} + +#[tokio::test] +async fn known_toolkit_but_unknown_action_proceeds_without_blocking() { + // Toolkit is cached but does NOT contain the requested action, so no + // fuller contract can be surfaced. The gate must degrade to Proceed rather + // than block the call forever. + let toolkit = "partialkit"; + seed_live_catalog_cache( + toolkit, + vec![full_contract("PARTIALKIT_OTHER_ACTION", toolkit)], + ); + + let config = Config::default(); + let gate = ContractGate::new(); + + assert!( + matches!( + consult(&gate, &config, "PARTIALKIT_FETCH_EMAILS").await, + GateDecision::Proceed + ), + "an action missing from the live catalog must not be gated" + ); +} + +#[tokio::test] +async fn distinct_actions_are_gated_independently() { + let toolkit = "multikit"; + let fetch = "MULTIKIT_FETCH_EMAILS"; + let send = "MULTIKIT_SEND_EMAIL"; + seed_live_catalog_cache( + toolkit, + vec![full_contract(fetch, toolkit), full_contract(send, toolkit)], + ); + + let config = Config::default(); + let gate = ContractGate::new(); + + // Each action surfaces its own contract exactly once, independently. + assert!(matches!( + consult(&gate, &config, fetch).await, + GateDecision::Surface(_) + )); + assert!(matches!( + consult(&gate, &config, send).await, + GateDecision::Surface(_) + )); + assert!(matches!( + consult(&gate, &config, fetch).await, + GateDecision::Proceed + )); + assert!(matches!( + consult(&gate, &config, send).await, + GateDecision::Proceed + )); +} diff --git a/src/openhuman/composio/mod.rs b/src/openhuman/composio/mod.rs index 9120e90851..cb5c2162a8 100644 --- a/src/openhuman/composio/mod.rs +++ b/src/openhuman/composio/mod.rs @@ -40,6 +40,7 @@ pub mod auth_retry; pub mod bus; pub mod client; mod connected_integrations; +pub mod contract_gate; pub(crate) mod direct_auth; pub mod error_mapping; pub mod execute_dispatch; From f002ced9d3cecfdda7cbcd089e6aebe9675b42f7 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Fri, 17 Jul 2026 19:00:24 +0530 Subject: [PATCH 2/4] fix(composio): gate contract-gate's tinyflows dependency on the flows feature The #4912 flows gate (merged after this branch was cut) put `openhuman::tinyflows` behind `#[cfg(feature = "flows")]`, so the gates-off build could not compile contract_gate.rs's unconditional `tinyflows::caps` import (E0433 in the "Rust Feature-Gate Smoke (gates off)" lane). The contract gate's live-catalog lookup is sourced entirely from the flows/tinyflows caps layer. With flows compiled out there is no catalog source, so the gate simply has no fuller contract to surface and always proceeds (the per-action Composio tool still runs; it just loses the pre-execute contract nudge). Guard the import and the lookup/format helpers on `feature = "flows"` in lockstep, restructure `consult` to Proceed when the feature is off, and gate the flows-seeding unit tests (contract_gate_tests.rs + the action_tool retry test) on the same feature. No behavior change in the default (flows-on) build. --- src/openhuman/composio/action_tool.rs | 3 ++ src/openhuman/composio/contract_gate.rs | 58 ++++++++++++++++--------- 2 files changed, 40 insertions(+), 21 deletions(-) diff --git a/src/openhuman/composio/action_tool.rs b/src/openhuman/composio/action_tool.rs index 23a0e72bff..15746fd2d7 100644 --- a/src/openhuman/composio/action_tool.rs +++ b/src/openhuman/composio/action_tool.rs @@ -520,6 +520,9 @@ mod tests { ); } + // Seeds the flows/tinyflows live-catalog cache, so it only builds with the + // `flows` feature on (the gate degrades to a no-op when flows is off). + #[cfg(feature = "flows")] #[tokio::test] async fn contract_gate_surfaces_full_contract_then_proceeds_on_retry() { // Regression for #4853: the FIRST per-action execute this turn must diff --git a/src/openhuman/composio/contract_gate.rs b/src/openhuman/composio/contract_gate.rs index 99827ef921..e3924b7547 100644 --- a/src/openhuman/composio/contract_gate.rs +++ b/src/openhuman/composio/contract_gate.rs @@ -28,8 +28,14 @@ use std::collections::HashSet; use std::sync::Mutex; -use crate::openhuman::composio::providers::toolkit_from_slug; use crate::openhuman::config::Config; +// The live-contract lookup is sourced from the flows/tinyflows caps catalog, +// which is compiled out when the `flows` feature is off (#4912). The gate then +// simply has no fuller contract to surface and always proceeds, so the import +// and the lookup/format helpers below are gated in lockstep. +#[cfg(feature = "flows")] +use crate::openhuman::composio::providers::toolkit_from_slug; +#[cfg(feature = "flows")] use crate::openhuman::tinyflows::caps::{fetch_live_toolkit_catalog, ToolContract}; /// Per-agent-turn record of which action contracts have already been surfaced @@ -95,32 +101,39 @@ pub async fn consult(gate: &ContractGate, config: &Config, action_slug: &str) -> return GateDecision::Proceed; } - match lookup_contract(config, action_slug).await { - Some(contract) => { - tracing::debug!( - target: "composio", - slug = %action_slug, - has_input_schema = contract.input_schema.is_some(), - required_arg_count = contract.required_args.len(), - "[composio][contract-gate] surfacing full contract before first execute" - ); - GateDecision::Surface(format_contract(action_slug, &contract)) - } - None => { - tracing::debug!( - target: "composio", - slug = %action_slug, - "[composio][contract-gate] no live contract available; proceeding without gating" - ); - GateDecision::Proceed - } + // The live catalog lives in the flows/tinyflows caps layer. With `flows` + // compiled out there is no catalog source, so the gate can never surface a + // fuller contract and always proceeds (the per-action tool still runs; it + // just does not get the pre-execute contract nudge). + #[cfg(feature = "flows")] + if let Some(contract) = lookup_contract(config, action_slug).await { + tracing::debug!( + target: "composio", + slug = %action_slug, + has_input_schema = contract.input_schema.is_some(), + required_arg_count = contract.required_args.len(), + "[composio][contract-gate] surfacing full contract before first execute" + ); + return GateDecision::Surface(format_contract(action_slug, &contract)); } + + // `config` is only consulted through the flows-gated lookup above. + #[cfg(not(feature = "flows"))] + let _ = config; + + tracing::debug!( + target: "composio", + slug = %action_slug, + "[composio][contract-gate] no live contract available; proceeding without gating" + ); + GateDecision::Proceed } /// Resolve the full live contract for `action_slug` from the process-cached /// live toolkit catalog. Returns `None` when the toolkit can't be derived, the /// catalog can't be fetched (unconfigured / offline — `fetch_live_toolkit_catalog` /// degrades to `None`), or the action isn't in it. +#[cfg(feature = "flows")] async fn lookup_contract(config: &Config, action_slug: &str) -> Option { let toolkit = toolkit_from_slug(action_slug)?; let contracts = fetch_live_toolkit_catalog(config, &toolkit).await?; @@ -131,6 +144,7 @@ async fn lookup_contract(config: &Config, action_slug: &str) -> Option String { let mut out = format!( "Before running `{action_slug}`, read its full contract below and then re-issue \ @@ -176,6 +190,8 @@ fn format_contract(action_slug: &str, contract: &ToolContract) -> String { out } -#[cfg(test)] +// The gate's unit tests seed the flows/tinyflows live-catalog cache, so they +// only compile and run with the `flows` feature on. +#[cfg(all(test, feature = "flows"))] #[path = "contract_gate_tests.rs"] mod tests; From 3561130cb26e9c8eb64592dbcfa388dc4519ecd5 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Fri, 17 Jul 2026 20:03:39 +0530 Subject: [PATCH 3/4] fix(composio): reload live config once before the contract gate (CodeRabbit) The per-action execute path consulted the contract gate with the captured spawn-time `self.config`, then separately reloaded a fresh snapshot for client creation + dispatch. After a mid-session `composio.mode` / credential / workspace change the gate's live-catalog fetch could resolve (or skip) a contract against stale routing while dispatch used fresh routing. Reload the snapshot once, up front (mirroring the existing #1710 Wave-4 principle), and reuse it for the gate consult, client creation, and dispatch so all three share identical live config. --- src/openhuman/composio/action_tool.rs | 61 ++++++++++++++------------- 1 file changed, 32 insertions(+), 29 deletions(-) diff --git a/src/openhuman/composio/action_tool.rs b/src/openhuman/composio/action_tool.rs index 15746fd2d7..bf2139eb1f 100644 --- a/src/openhuman/composio/action_tool.rs +++ b/src/openhuman/composio/action_tool.rs @@ -192,6 +192,31 @@ impl Tool for ComposioActionTool { } } + // [#1710 Wave 4 / #4853] Reload the live config snapshot ONCE, up front, + // and use it for BOTH the contract-gate lookup and dispatch. A mid-session + // `composio.mode` / credential / workspace change must route the gate's + // live-catalog fetch and the actual execution through the SAME config; + // consulting the captured spawn-time `self.config` here would let the gate + // resolve (or skip) a contract against stale routing while dispatch used + // fresh routing. Anchored to this tool's original config path rather than + // re-resolving process-global `OPENHUMAN_WORKSPACE` (the tool is scoped to + // the user/workspace it was created for). + let live_config = + match config_rpc::reload_config_snapshot_with_timeout(self.config.as_ref()).await { + Ok(c) => c, + Err(e) => { + tracing::warn!( + tool = %self.action_name, + error = %e, + "[composio] per-action execute: load_config failed" + ); + return Ok(ToolResult::error(format!( + "{}: failed to load live config: {e}", + self.action_name + ))); + } + }; + // Contract gate (#4853): the per-action tool is built from the thin // spawn-time `list_tools` schema (often `{"type":"object"}` with no // field descriptions), so the model guesses argument formats — most @@ -201,10 +226,8 @@ impl Tool for ComposioActionTool { // let the retry — now with the schema in context — execute. Degrades to // a normal execute whenever the contract can't be resolved (see // `contract_gate::consult`), so an unconfigured/offline client never - // blocks the action. - match super::contract_gate::consult(&self.gate, self.config.as_ref(), &self.action_name) - .await - { + // blocks the action. Uses `live_config` so gate routing matches dispatch. + match super::contract_gate::consult(&self.gate, &live_config, &self.action_name).await { super::contract_gate::GateDecision::Surface(contract) => { tracing::info!( tool = %self.action_name, @@ -233,31 +256,11 @@ impl Tool for ComposioActionTool { &iana, ); - // Resolve the client through the mode-aware factory on every - // call so a direct-mode toggle takes effect immediately - // (#1710). The pre-baked-client variant of this code routed all - // executions through the backend tinyhumans tenant regardless - // of mode — silently breaking direct mode for tool execution. - // [#1710 Wave 4] Reload config fresh per execute so a mid-session - // `composio.mode` toggle takes effect at the very next tool call. - // Anchor the reload to this tool's original config path rather - // than re-resolving process-global `OPENHUMAN_WORKSPACE`; the - // tool is scoped to the user/workspace it was created for. - let live_config = - match config_rpc::reload_config_snapshot_with_timeout(self.config.as_ref()).await { - Ok(c) => c, - Err(e) => { - tracing::warn!( - tool = %self.action_name, - error = %e, - "[composio] per-action execute: load_config failed" - ); - return Ok(ToolResult::error(format!( - "{}: failed to load live config: {e}", - self.action_name - ))); - } - }; + // Resolve the client through the mode-aware factory on every call so a + // direct-mode toggle takes effect immediately (#1710), reusing the + // `live_config` snapshot reloaded above so the gate lookup and this + // dispatch share identical routing. The pre-baked-client variant routed + // all executions through the backend tinyhumans tenant regardless of mode. let kind = match create_composio_client(&live_config) { Ok(kind) => kind, Err(e) => { From 8714bb42eadebe3b3f1f163e486d220f9364cfa7 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Sat, 18 Jul 2026 01:04:20 +0530 Subject: [PATCH 4/4] docs(composio): correct gate scope docs + note per-call gate in resolver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses maintainer review (graycyrus / oxoxDev), docs-only: - contract_gate.rs: the ContractGate's seen-set doc overclaimed "per-turn". It is actually scoped to the ContractGate instance (one per ComposioActionTool spawn) — a single turn in the common case, but a long-lived integrations_agent spawn can span turns, and the gate does not reset on context compaction (deferred). Reworded to say so accurately. - subagent_runner/ops/provider.rs: document that LazyToolkitResolver::resolve builds a fresh ComposioActionTool (and gate) per call, so once wired to dispatch the resolved tool must be cached per turn or a surfaced action can never proceed (oxoxDev forward note). No behavior change. --- .../agent/harness/subagent_runner/ops/provider.rs | 8 ++++++++ src/openhuman/composio/contract_gate.rs | 15 +++++++++------ 2 files changed, 17 insertions(+), 6 deletions(-) diff --git a/src/openhuman/agent/harness/subagent_runner/ops/provider.rs b/src/openhuman/agent/harness/subagent_runner/ops/provider.rs index acdf56f1e4..e8acc67533 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops/provider.rs +++ b/src/openhuman/agent/harness/subagent_runner/ops/provider.rs @@ -247,6 +247,14 @@ pub(crate) struct LazyToolkitResolver { const TIER4_MIN_SLUG_LEN: usize = 8; impl LazyToolkitResolver { + /// NOTE (contract gate, #4853): this builds a *fresh* `ComposioActionTool` + /// — and therefore a fresh, empty `ContractGate` — on every call. The gate's + /// surface-once state lives in the tool instance, so if a future wiring + /// resolves a new tool per invocation the gate would surface the full + /// contract on every call and the retry would never see `first_time == false` + /// to proceed. When this path is wired to actually dispatch, cache the + /// resolved tool (and its gate) per turn so a given action can proceed after + /// its contract has been surfaced once. pub(super) fn resolve(&self, name: &str) -> Option> { let action = self.find_action(name)?; Some(Box::new( diff --git a/src/openhuman/composio/contract_gate.rs b/src/openhuman/composio/contract_gate.rs index e3924b7547..5eb7479fc7 100644 --- a/src/openhuman/composio/contract_gate.rs +++ b/src/openhuman/composio/contract_gate.rs @@ -38,14 +38,17 @@ use crate::openhuman::composio::providers::toolkit_from_slug; #[cfg(feature = "flows")] use crate::openhuman::tinyflows::caps::{fetch_live_toolkit_catalog, ToolContract}; -/// Per-agent-turn record of which action contracts have already been surfaced -/// to the model, so the gate blocks a given action at most once per turn. +/// Record of which action contracts have already been surfaced to the model, +/// so the gate blocks a given action at most once per gate instance. /// /// One [`ContractGate`] is held per [`super::action_tool::ComposioActionTool`] /// instance; those tools are constructed fresh per `integrations_agent` spawn -/// and live for that spawn's tool loop, so "seen" is scoped to the turn without -/// any task-local plumbing. Interior-mutable so the gate can record state -/// through the tool's `&self` `execute`. +/// and live for that spawn's tool loop. That loop is a single agent turn in the +/// common case, so "seen" behaves as per-turn state without any task-local +/// plumbing — but a long-lived spawn can span multiple turns, and this gate +/// does NOT reset when the surfaced schema drops out of context via compaction +/// (tracked as follow-up; see the module-level note). Interior-mutable so the +/// gate can record state through the tool's `&self` `execute`. #[derive(Default)] pub struct ContractGate { seen: Mutex>, @@ -58,7 +61,7 @@ impl ContractGate { /// Insert `slug` (normalised to upper-case) into the seen-set. Returns /// `true` when it was NOT already present — i.e. this is the first time the - /// gate has been consulted for this action this turn. + /// gate has been consulted for this action for this gate's lifetime. /// /// The lock is taken and released entirely within this call, so no guard is /// held across the caller's later `await`.