diff --git a/src/openhuman/composio/client.rs b/src/openhuman/composio/client.rs index e4b72faaa..bcd4cf074 100644 --- a/src/openhuman/composio/client.rs +++ b/src/openhuman/composio/client.rs @@ -1066,7 +1066,15 @@ pub async fn direct_list_connections( /// the same model-callable shape backend-mode does. Downstream curated- /// whitelist filtering (`evaluate_tool_visibility` / `find_curated`) /// still applies at the `ops::composio_list_tools` layer. -pub(super) async fn direct_list_tools( +/// +/// `pub(crate)` (widened from `pub(super)`) so +/// `tinyflows::caps::fetch_raw_toolkit_tools` can call this directly for +/// the LIVE (uncurated) tool-contract catalog the Workflow builder grounds +/// against — that caller deliberately bypasses `composio_list_tools`'s +/// curated-whitelist filter (`filter_list_tools_response_for_direct`), +/// which this function never applies itself; the filter is layered on by +/// its `composio_list_tools` caller, not baked in here. +pub(crate) async fn direct_list_tools( direct: &Arc, toolkits: &[String], tags: Option<&[String]>, diff --git a/src/openhuman/flows/agents/workflow_builder/agent.toml b/src/openhuman/flows/agents/workflow_builder/agent.toml index e8da3b39b..93f46a1c9 100644 --- a/src/openhuman/flows/agents/workflow_builder/agent.toml +++ b/src/openhuman/flows/agents/workflow_builder/agent.toml @@ -47,6 +47,7 @@ named = [ "get_flow_run", "list_flow_connections", "search_tool_catalog", + "get_tool_contract", "list_agent_profiles", "dry_run_workflow", "run_flow", diff --git a/src/openhuman/flows/agents/workflow_builder/prompt.md b/src/openhuman/flows/agents/workflow_builder/prompt.md index 4266a313c..acab92c3e 100644 --- a/src/openhuman/flows/agents/workflow_builder/prompt.md +++ b/src/openhuman/flows/agents/workflow_builder/prompt.md @@ -70,12 +70,20 @@ it as real). Rules: - `list_flow_connections` → the exact `connection_ref` values available (Composio accounts + named HTTP creds). Put these verbatim on nodes that act on a connected account. Never invent a connection. - - `search_tool_catalog` → real Composio action **slugs** for `tool_call` - nodes. **Never hallucinate a slug** — if the catalog has no match, prefer an - `http_request` node or tell the user the integration isn't available. - Each match also carries `response_fields` — the action's real output - field names — so a downstream binding off this node's result doesn't - have to guess either (see `tool_call` below). + - `search_tool_catalog { query, toolkit? }` → real Composio action + **slugs** from the FULL LIVE catalog for ANY named app — connected or + not, curated or not (curated matches come back `featured: true` and are + ranked first). **Never hallucinate a slug** — if the catalog has no + match for the app, prefer an `http_request` node or tell the user the + integration isn't available. Each match also carries `required_args` / + `output_fields` / `primary_array_path` — but call `get_tool_contract + { slug }` before you actually WIRE a match: it hands back the exact + required args, the full input/output schema, and the array path a + `split_out` should use (see `tool_call` below). `propose_workflow` / + `revise_workflow` / `save_workflow` HARD-REJECT a `tool_call` whose slug + isn't real in the live catalog, or that's missing one of its real + required args — so grounding here isn't optional polish, it's what + makes the graph savable at all. - `list_flows` / `get_flow` → reuse or clone an existing flow instead of duplicating one. - **Missing the integration the workflow needs?** See "Connecting @@ -186,21 +194,42 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. 3. **`tool_call`** — an action. Two flavours by `config.slug`: - **Composio app action** — `config.slug` = a real action slug (from `search_tool_catalog`, e.g. `GMAIL_SEND_EMAIL`) + `config.connection_ref` - for the account. **Wire every REQUIRED arg in `config.args` from a named - upstream node** — e.g. an email send needs `to`/`recipient_email`, usually - `"to": "=nodes..item.json.email"` (drop `.json` only if - `` is a `code`/`transform`/`split_out`/`merge`/`trigger` node - — see "the envelope" below). A required arg left unwired (or whose - expression misses) now fails BEFORE the provider call — both in - `dry_run_workflow` and in real runs — with an error naming the field. + for the account. **Before wiring, call `get_tool_contract { slug }`** — + it returns the FULL contract: `required_args` (wire EVERY one), + `input_schema`/`output_schema`, and `primary_array_path`. Wire every + required arg in `config.args` from a named upstream node — e.g. an + email send needs `to`/`recipient_email`, usually `"to": + "=nodes..item.json.email"` (drop `.json` only if + `` is a `code`/`transform`/`split_out`/`merge`/`trigger` + node — see "the envelope" below). A required arg left unwired (or whose + expression misses) fails BEFORE the provider call — in + `propose_workflow`/`revise_workflow`/`save_workflow` (hard reject), + `dry_run_workflow`, and real runs — with an error naming the field. + - **The slug itself is enforced too.** `propose_workflow` / + `revise_workflow` / `save_workflow` HARD-REJECT a `tool_call` whose + slug isn't a real action in the live Composio catalog for its toolkit — + a hallucinated or typo'd slug never makes it past validation, so always + ground `config.slug` in a `search_tool_catalog` result first. - **Wiring a DOWNSTREAM node off THIS tool's output?** Don't guess the field name (e.g. assuming `GMAIL_FETCH_EMAILS` returns `.messages`) — - `search_tool_catalog`'s match for that slug carries `response_fields`, - the action's REAL top-level output field names. Bind - `=nodes..item.json.` to one of those. If - `response_fields` is empty (a `response_fields_note` will say the shape - is unknown), `dry_run_workflow` the binding before you propose/save it — + `get_tool_contract`'s `output_fields` names the action's REAL top-level + output field names. Bind `=nodes..item.json.` to + one of those. If `output_fields` is empty (schema unknown for that + action), `dry_run_workflow` the binding before you propose/save it — don't ship a guessed field name. + - **Fanning out over THIS tool's result list (`split_out`)?** Use + `get_tool_contract`'s `primary_array_path`, prefixed `json.` — e.g. + `"path": "json.data.messages"` — as the downstream `split_out.path`, + rather than guessing where the array lives in the response. + - **App not connected yet?** You can still build the node with a real + slug from `search_tool_catalog` (searches the FULL live catalog + regardless of connection state) and ground it with `get_tool_contract + { slug }` (resolves that known slug's toolkit and fetches ITS full + contract from the same live catalog — a grounding lookup, not a + search, and also works regardless of connection state) and either call + `composio_connect { toolkit }` yourself (see "Connecting integrations" + below) or note in your reply that the user needs to connect it — the + flow will also prompt for the connection the first time it actually runs. - **Native OpenHuman tool** — `config.slug` = `oh:` (e.g. `oh:web_search`) to call one of the assistant's own built-in tools (search, media generation, files, …). No `connection_ref`. Args go in `config.args`. diff --git a/src/openhuman/flows/builder_tools.rs b/src/openhuman/flows/builder_tools.rs index 6b4e20e75..c6fa619d2 100644 --- a/src/openhuman/flows/builder_tools.rs +++ b/src/openhuman/flows/builder_tools.rs @@ -11,7 +11,8 @@ //! | [`GetFlowTool`] | `None` | read: fetch a saved flow's graph | //! | [`GetFlowRunTool`] | `None` | read: fetch a run's steps | //! | [`ListFlowConnectionsTool`] | `None` | read: connection refs (ids/names only) | -//! | [`SearchToolCatalogTool`] | `None` | read: real Composio tool slugs | +//! | [`SearchToolCatalogTool`] | `None` | read: real Composio tool slugs (live catalog) | +//! | [`GetToolContractTool`] | `None` | read: one action's FULL live contract | //! | [`ListAgentProfilesTool`] | `None` | read: selectable agent kinds (`agent_ref`)| //! | [`DryRunWorkflowTool`] | `Execute` (tier-gated) | run a *draft* against MOCK capabilities | //! | [`SaveWorkflowTool`] | `Write` | persist a graph onto an EXISTING flow | @@ -186,6 +187,23 @@ impl Tool for ReviseWorkflowTool { ))); } + // Tool-contract enforcement gate (systemic tool-contract fix, Part 2): + // reject a `tool_call` node whose slug isn't a REAL action in the + // live Composio catalog, or whose real required args aren't all wired. + let contract_errors = ops::validate_tool_contracts(&self.config, &graph).await; + if !contract_errors.is_empty() { + tracing::debug!( + target: "flows", + %name, + error_count = contract_errors.len(), + "[flows] revise_workflow: tool-contract check rejected the revised graph" + ); + return Ok(ToolResult::error(format!( + "{}\n\nFix these tool_call nodes and call revise_workflow again.", + contract_errors.join("\n\n") + ))); + } + let summary = super::tools::build_summary(&graph); let mut warnings = ops::graph_trigger_warnings(&graph); // Author-time wiring check: unwired REQUIRED Composio args come back @@ -495,18 +513,23 @@ impl Tool for ListFlowConnectionsTool { } // ───────────────────────────────────────────────────────────────────────────── -// search_tool_catalog — read-only: real Composio tool slugs +// search_tool_catalog — read-only: real Composio tool slugs from the FULL +// LIVE catalog (systemic tool-contract fix, Part 1) // ───────────────────────────────────────────────────────────────────────────── -/// `search_tool_catalog`: search OpenHuman's curated Composio catalog for REAL -/// action slugs so `tool_call` nodes are grounded in slugs that actually exist -/// (rather than a hallucinated slug that fails the save-time curation gate). +/// `search_tool_catalog`: search the FULL LIVE Composio catalog — every real +/// action for a named app, connected or not, curated or not — so `tool_call` +/// nodes are grounded in slugs that actually exist (rather than a hallucinated +/// slug that fails the save-time [`crate::openhuman::flows::ops::validate_tool_contracts`] +/// gate). /// -/// Also grounds the OUTPUT side: each result carries a best-effort -/// `response_fields` list — the action's real top-level response field names -/// (see [`crate::openhuman::tinyflows::caps::composio_response_fields`]) — so -/// a downstream binding (`=nodes..item.json.`) can be wired to a -/// field that actually exists instead of a guessed one. +/// Also grounds the OUTPUT side: each result carries the action's real +/// `output_fields` (top-level response field names) and — when known — a +/// `primary_array_path`, so a downstream binding +/// (`=nodes..item.json.`) or a `split_out.path` can be wired to a +/// real field/path instead of a guessed one. Call +/// [`GetToolContractTool`]/`get_tool_contract` for the FULL contract (schemas +/// included) before wiring a match's args. pub struct SearchToolCatalogTool { config: Arc, } @@ -520,17 +543,30 @@ impl SearchToolCatalogTool { /// Cap on returned matches so a broad query can't flood the agent's context. const MAX_CATALOG_RESULTS: usize = 40; -/// Search the curated catalog for action slugs whose slug (or toolkit) matches -/// every whitespace-separated term in `query` (case-insensitive AND). When -/// `toolkit` is set, only that toolkit's catalog is scanned. Pure — no I/O. -pub(crate) fn search_curated_catalog( +/// Search the FULL LIVE Composio catalog (via +/// [`crate::openhuman::tinyflows::caps::fetch_live_toolkit_catalog`]) for +/// actions whose slug or description matches every whitespace-separated term +/// in `query` (case-insensitive AND). When `toolkit` is set, only that +/// toolkit is scanned — this is how the builder can search ANY named app +/// (connected or not) rather than only the toolkits already +/// [`agent_ready_toolkits`](crate::openhuman::memory_sync::composio::providers::agent_ready_toolkits); +/// with no `toolkit` filter, the search is scoped to that agent-ready set (a +/// bare keyword query with no app named would otherwise have to fan out to +/// every toolkit Composio knows about). +/// +/// Curated matches (`is_curated`) are ranked first (a stable sort, so ties +/// preserve fetch order) — never filtered out; a real, uncurated action is +/// just as valid a result, only ranked after the curated ones. A toolkit +/// whose live-catalog fetch fails (no backend session, network error) +/// contributes zero results rather than erroring the whole search. +pub(crate) async fn search_live_catalog( + config: &Config, query: &str, toolkit_filter: Option<&str>, limit: usize, ) -> Vec { - use crate::openhuman::memory_sync::composio::providers::{ - agent_ready_toolkits, catalog_for_toolkit, - }; + use crate::openhuman::memory_sync::composio::providers::agent_ready_toolkits; + use crate::openhuman::tinyflows::caps::fetch_live_toolkit_catalog; let terms: Vec = query .split_whitespace() @@ -545,31 +581,55 @@ pub(crate) fn search_curated_catalog( .collect(), }; - let mut out = Vec::new(); - for toolkit in toolkits { - let Some(catalog) = catalog_for_toolkit(&toolkit) else { - continue; - }; - for tool in catalog { + // Fetch every candidate toolkit's live catalog concurrently — a bare + // keyword query (no `toolkit` filter) fans out across every agent-ready + // toolkit, and fetching them one at a time would pay for each one's + // round trip back-to-back (the per-toolkit cache only helps repeats). + let fetched: Vec<( + String, + Option>, + )> = futures::future::join_all(toolkits.into_iter().map(|toolkit| async move { + let catalog = fetch_live_toolkit_catalog(config, &toolkit).await; + (toolkit, catalog) + })) + .await; + + let mut matches: Vec<(bool, Value)> = Vec::new(); + for (toolkit, catalog) in fetched { + let Some(catalog) = catalog else { continue }; + for tool in &catalog { let slug_lc = tool.slug.to_ascii_lowercase(); - // Every term must match either the slug or the toolkit name. - let matches = terms - .iter() - .all(|term| slug_lc.contains(term) || toolkit.contains(term)); - if !matches { + let desc_lc = tool + .description + .as_deref() + .unwrap_or_default() + .to_ascii_lowercase(); + let is_match = terms.iter().all(|term| { + slug_lc.contains(term) || toolkit.contains(term) || desc_lc.contains(term) + }); + if !is_match { continue; } - out.push(json!({ - "slug": tool.slug, - "toolkit": toolkit, - "scope": tool.scope.as_str(), - })); - if out.len() >= limit { - return out; - } + matches.push(( + tool.is_curated, + json!({ + "slug": tool.slug, + "toolkit": toolkit, + "description": tool.description, + "required_args": tool.required_args, + "output_fields": tool.output_fields, + "primary_array_path": tool.primary_array_path, + "featured": tool.is_curated, + }), + )); } } - out + + // Curated (`featured`) results first; stable sort preserves fetch order + // within each group. + matches.sort_by_key(|(is_curated, _)| std::cmp::Reverse(*is_curated)); + matches.truncate(limit); + matches.into_iter().map(|(_, v)| v).collect() } #[async_trait] @@ -579,17 +639,19 @@ impl Tool for SearchToolCatalogTool { } fn description(&self) -> &str { - "Search the curated Composio tool catalog for REAL action slugs to use on \ - `tool_call` nodes. Read-only. Query by keyword (e.g. 'send email', \ - 'slack message'); optionally scope to one `toolkit` (e.g. 'gmail'). \ - Returns matching { slug, toolkit, scope, response_fields } entries. \ - ALWAYS ground a tool_call node's `slug` in a real result here — do not \ - invent slugs. `response_fields` names the action's REAL top-level \ - output field names (from Composio's own schema) — use THOSE, not a \ - guess, when a downstream node reads this tool's output via \ - `=nodes..item.json.`. When `response_fields` is empty a \ - `response_fields_note` explains the output shape is unknown — \ - dry_run_workflow the binding to verify it resolves before proposing." + "Search the FULL LIVE Composio catalog for REAL action slugs to use on `tool_call` \ + nodes — every action for a named app, whether or not the user has connected it yet \ + and whether or not it's one of OpenHuman's hand-curated actions. Read-only. Query by \ + keyword (e.g. 'send email', 'slack message'); optionally scope to one `toolkit` (e.g. \ + 'gmail', or any Composio app name) to search that app specifically. Returns matching \ + { slug, toolkit, description, required_args, output_fields, primary_array_path, \ + featured } entries, curated (`featured: true`) matches ranked first. ALWAYS ground a \ + tool_call node's `slug` in a real result here — never invent one. Before wiring a \ + match's args or a downstream binding, call get_tool_contract { slug } for the FULL \ + contract (exact required_args, full input/output JSON Schema) — this search result is \ + enough to FIND the right slug, get_tool_contract is what grounds the WIRING. If the \ + app isn't connected yet, you can still build the node and use composio_connect (or \ + tell the user) — the flow will prompt for the connection at run time." } fn parameters_schema(&self) -> Value { @@ -598,11 +660,11 @@ impl Tool for SearchToolCatalogTool { "properties": { "query": { "type": "string", - "description": "Keywords to match against tool slugs (case-insensitive; all terms must match)." + "description": "Keywords to match against tool slugs/descriptions (case-insensitive; all terms must match)." }, "toolkit": { "type": "string", - "description": "Optional toolkit slug to scope the search (e.g. 'gmail', 'slack')." + "description": "Optional toolkit/app slug to scope the search (e.g. 'gmail', 'slack', or any named Composio app — connected or not)." } }, "required": ["query"], @@ -628,59 +690,9 @@ impl Tool for SearchToolCatalogTool { target: "flows", %query, toolkit = toolkit.unwrap_or("(any)"), - "[flows] search_tool_catalog: searching curated Composio catalog (read-only)" + "[flows] search_tool_catalog: searching the FULL LIVE Composio catalog (read-only)" ); - let mut results = search_curated_catalog(&query, toolkit, MAX_CATALOG_RESULTS); - - // Resolve each *distinct* slug's output fields concurrently rather - // than awaiting one `composio_response_fields` call per matched - // result in sequence — a broad query spanning several toolkits would - // otherwise pay for their catalog round trips back-to-back (the - // per-toolkit cache only helps repeat lookups, not the first one). - let mut unique_slugs: Vec = results - .iter() - .filter_map(|r| r.get("slug").and_then(Value::as_str).map(str::to_string)) - .collect(); - unique_slugs.sort(); - unique_slugs.dedup(); - let fetched = futures::future::join_all(unique_slugs.into_iter().map(|slug| { - let config = self.config.clone(); - async move { - let fields = - crate::openhuman::tinyflows::caps::composio_response_fields(&config, &slug) - .await; - (slug, fields) - } - })) - .await; - let response_fields_by_slug: std::collections::HashMap>> = - fetched.into_iter().collect(); - - for result in &mut results { - let Some(slug) = result - .get("slug") - .and_then(Value::as_str) - .map(str::to_string) - else { - continue; - }; - let response_fields = response_fields_by_slug.get(&slug).cloned().flatten(); - let Value::Object(map) = result else { - continue; - }; - match response_fields { - Some(fields) => { - map.insert("response_fields".to_string(), json!(fields)); - } - None => { - map.insert("response_fields".to_string(), json!(Vec::::new())); - map.insert( - "response_fields_note".to_string(), - json!("output shape unknown — dry-run to verify the binding resolves"), - ); - } - } - } + let results = search_live_catalog(&self.config, &query, toolkit, MAX_CATALOG_RESULTS).await; Ok(ToolResult::success(serde_json::to_string_pretty(&json!({ "query": query, "count": results.len(), @@ -689,6 +701,110 @@ impl Tool for SearchToolCatalogTool { } } +// ───────────────────────────────────────────────────────────────────────────── +// get_tool_contract — read-only: the FULL live contract for one action slug +// ───────────────────────────────────────────────────────────────────────────── + +/// `get_tool_contract`: fetch the FULL live [`ToolContract`](crate::openhuman::tinyflows::caps::ToolContract) +/// for one Composio action slug — the grounding step the builder MUST take +/// before wiring a `search_tool_catalog` match's args or a downstream +/// binding/`split_out.path` off it. Where `search_tool_catalog` is for +/// FINDING a real slug, this is for WIRING it correctly: exact +/// `required_args` (wire every one), the full `input_schema`/`output_schema`, +/// and `primary_array_path` (prefixed `json.` for a `split_out.path`). +pub struct GetToolContractTool { + config: Arc, +} + +impl GetToolContractTool { + pub fn new(config: Arc) -> Self { + Self { config } + } +} + +#[async_trait] +impl Tool for GetToolContractTool { + fn name(&self) -> &str { + "get_tool_contract" + } + + fn description(&self) -> &str { + "Fetch the FULL live contract for one Composio action slug (found via \ + search_tool_catalog) before wiring it into a tool_call node. Read-only. Returns { \ + slug, toolkit, description, required_args, input_schema, output_fields, \ + output_schema, primary_array_path, is_curated }. Use `required_args` for EVERY arg \ + you must wire in config.args; use `output_fields` for a downstream \ + `=nodes..item.json.` binding — never guess a field name; use \ + `primary_array_path` (prefixed with `json.`, e.g. \"json.data.messages\") verbatim as \ + a downstream split_out.path when you need to fan out over this action's result list. \ + Call this for every real slug right before you wire its args — search_tool_catalog's \ + summary is enough to find the slug, this is what grounds the wiring." + } + + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "slug": { + "type": "string", + "description": "The exact Composio action slug, e.g. 'GMAIL_SEND_EMAIL' (from search_tool_catalog)." + } + }, + "required": ["slug"], + "additionalProperties": false + }) + } + + fn permission_level(&self) -> PermissionLevel { + PermissionLevel::None + } + + fn external_effect(&self) -> bool { + false + } + + async fn execute(&self, args: Value) -> anyhow::Result { + let slug = match args.get("slug").and_then(Value::as_str).map(str::trim) { + Some(s) if !s.is_empty() => s.to_string(), + _ => return Ok(ToolResult::error("Missing 'slug' parameter".to_string())), + }; + let Some(toolkit) = + crate::openhuman::memory_sync::composio::providers::toolkit_from_slug(&slug) + else { + return Ok(ToolResult::error(format!( + "Could not extract a toolkit from slug '{slug}' — it must look like \ + '_' (e.g. 'GMAIL_SEND_EMAIL')." + ))); + }; + + tracing::debug!( + target: "flows", + %slug, + %toolkit, + "[flows] get_tool_contract: fetching the live contract (read-only)" + ); + + let Some(catalog) = + crate::openhuman::tinyflows::caps::fetch_live_toolkit_catalog(&self.config, &toolkit) + .await + else { + return Ok(ToolResult::error(format!( + "Could not fetch the live Composio catalog for toolkit '{toolkit}' (no backend \ + session, or a transient failure) — try again, or use search_tool_catalog to \ + confirm the toolkit is reachable." + ))); + }; + + match catalog.iter().find(|c| c.slug.eq_ignore_ascii_case(&slug)) { + Some(contract) => Ok(ToolResult::success(serde_json::to_string_pretty(contract)?)), + None => Ok(ToolResult::error(format!( + "'{slug}' is not a real action in the '{toolkit}' toolkit's live catalog — use \ + search_tool_catalog to find a real slug." + ))), + } + } +} + // ───────────────────────────────────────────────────────────────────────────── // list_agent_profiles — read-only: selectable agent kinds for an `agent` node // ───────────────────────────────────────────────────────────────────────────── @@ -1339,6 +1455,23 @@ impl Tool for SaveWorkflowTool { binding_errors.join("\n\n") ))); } + // Tool-contract enforcement gate (systemic tool-contract fix, Part 2): + // reject a `tool_call` node whose slug isn't a REAL action in the + // live Composio catalog, or whose real required args aren't all + // wired — before the graph is ever persisted. + let contract_errors = ops::validate_tool_contracts(&self.config, &graph).await; + if !contract_errors.is_empty() { + tracing::debug!( + target: "flows", + %flow_id, + error_count = contract_errors.len(), + "[flows] save_workflow: tool-contract check rejected the graph" + ); + return Ok(ToolResult::error(format!( + "{}\n\nFix these tool_call nodes and call save_workflow again.", + contract_errors.join("\n\n") + ))); + } // Author-time warnings (unfired trigger kinds + unwired REQUIRED // Composio args) were previously computed by propose/revise but never // surfaced again at save time — add them here so the agent sees any diff --git a/src/openhuman/flows/builder_tools_tests.rs b/src/openhuman/flows/builder_tools_tests.rs index d64ef409e..02a4a68a0 100644 --- a/src/openhuman/flows/builder_tools_tests.rs +++ b/src/openhuman/flows/builder_tools_tests.rs @@ -148,12 +148,38 @@ async fn list_flow_connections_is_read_only() { assert!(parsed["connections"].is_array()); } -// ── search_tool_catalog ────────────────────────────────────────────────────── +// ── search_tool_catalog / get_tool_contract ───────────────────────────────── +// The live-catalog cache is process-global (`LIVE_CATALOG_CACHE`) — every +// test below seeds the exact toolkit(s)/contract(s) it needs via +// `seed_live_catalog_cache` so none of this touches a live Composio backend, +// and keeps each toolkit's seeded contents self-consistent across tests that +// share a toolkit key (same discipline the pre-fix required-args/response- +// fields caches already required). -#[test] -fn search_curated_catalog_finds_real_gmail_slug() { - // Grounded search over the curated catalog returns a real slug/scope. - let results = search_curated_catalog("gmail", Some("gmail"), 40); +use crate::openhuman::tinyflows::caps::{seed_live_catalog_cache, ToolContract}; + +fn seeded_gmail_send_contract() -> ToolContract { + ToolContract { + slug: "GMAIL_SEND_EMAIL".to_string(), + toolkit: "gmail".to_string(), + description: Some("Send an email".to_string()), + required_args: vec!["to".to_string(), "body".to_string()], + input_schema: Some(json!({ "type": "object", "required": ["to", "body"] })), + output_fields: vec!["id".to_string(), "threadId".to_string()], + output_schema: Some(json!({ + "type": "object", + "properties": { "id": {"type": "string"}, "threadId": {"type": "string"} } + })), + primary_array_path: None, + is_curated: true, + } +} + +#[tokio::test] +async fn search_live_catalog_finds_a_seeded_real_gmail_slug() { + seed_live_catalog_cache("gmail", vec![seeded_gmail_send_contract()]); + let config = Config::default(); + let results = search_live_catalog(&config, "send", Some("gmail"), 40).await; assert!(!results.is_empty(), "gmail catalog should have entries"); for r in &results { assert_eq!(r["toolkit"], "gmail"); @@ -162,19 +188,45 @@ fn search_curated_catalog_finds_real_gmail_slug() { .unwrap() .to_ascii_uppercase() .starts_with("GMAIL")); - assert!(r["scope"].is_string()); + assert_eq!(r["featured"], true); } } -#[test] -fn search_curated_catalog_all_terms_must_match() { +#[tokio::test] +async fn search_live_catalog_all_terms_must_match() { + seed_live_catalog_cache("gmail", vec![seeded_gmail_send_contract()]); + let config = Config::default(); // A nonsense term matches nothing. - let results = search_curated_catalog("zzz_no_such_slug_zzz", None, 40); + let results = search_live_catalog(&config, "zzz_no_such_slug_zzz", Some("gmail"), 40).await; assert!(results.is_empty()); } +#[tokio::test] +async fn search_live_catalog_ranks_curated_before_uncurated_without_hiding_either() { + // Uses its own cache key (never `"gmail"`) — the process-global + // `LIVE_CATALOG_CACHE` is shared with every other `#[tokio::test]` in + // this file, most of which seed `"gmail"` with a single curated entry. + // This test's 2-item, exact-order assertion would be flaky if a + // concurrently-running test's `seed_live_catalog_cache("gmail", ..)` + // replaced the entry between this seed and the query below. + let mut uncurated = seeded_gmail_send_contract(); + uncurated.slug = "GMAIL_UNCURATED_SEND".to_string(); + uncurated.is_curated = false; + seed_live_catalog_cache( + "gmailranktest", + vec![uncurated, seeded_gmail_send_contract()], + ); + + let config = Config::default(); + let results = search_live_catalog(&config, "send", Some("gmailranktest"), 40).await; + assert_eq!(results.len(), 2, "a real, uncurated action is never hidden"); + assert_eq!(results[0]["featured"], true, "curated match ranks first"); + assert_eq!(results[1]["featured"], false); +} + #[tokio::test] async fn search_tool_catalog_tool_is_read_only_and_grounds() { + seed_live_catalog_cache("gmail", vec![seeded_gmail_send_contract()]); let tmp = TempDir::new().unwrap(); let tool = SearchToolCatalogTool::new(test_config(&tmp)); assert_eq!(tool.name(), "search_tool_catalog"); @@ -200,17 +252,12 @@ async fn search_tool_catalog_missing_query_is_error() { } #[tokio::test] -async fn search_tool_catalog_grounds_response_fields_from_seeded_output_schema() { - // A known action's output schema (seeded, standing in for a live - // Composio fetch) surfaces as real `response_fields` on the match. +async fn search_tool_catalog_grounds_output_fields_from_the_live_catalog() { + // A known action's real output schema (seeded, standing in for a live + // Composio fetch) surfaces as real `output_fields`/`required_args` on + // the match — no separate per-slug lookup needed anymore. + seed_live_catalog_cache("gmail", vec![seeded_gmail_send_contract()]); let tmp = TempDir::new().unwrap(); - let mut entries = std::collections::HashMap::new(); - entries.insert( - "GMAIL_SEND_EMAIL".to_string(), - vec!["id".to_string(), "threadId".to_string()], - ); - crate::openhuman::tinyflows::caps::seed_response_fields_cache("gmail", entries); - let tool = SearchToolCatalogTool::new(test_config(&tmp)); let result = tool .execute(json!({ "query": "send", "toolkit": "gmail" })) @@ -222,34 +269,45 @@ async fn search_tool_catalog_grounds_response_fields_from_seeded_output_schema() let send_email = results .iter() .find(|r| r["slug"] == "GMAIL_SEND_EMAIL") - .expect("GMAIL_SEND_EMAIL should be in the curated catalog"); - let fields: Vec<&str> = send_email["response_fields"] + .expect("GMAIL_SEND_EMAIL should be in the live catalog"); + let fields: Vec<&str> = send_email["output_fields"] .as_array() .unwrap() .iter() .map(|v| v.as_str().unwrap()) .collect(); assert_eq!(fields, vec!["id", "threadId"]); - assert!(send_email.get("response_fields_note").is_none()); + assert_eq!(send_email["required_args"], json!(["to", "body"])); } #[tokio::test] async fn search_tool_catalog_degrades_gracefully_when_output_schema_unknown() { - // No cache entry seeded for this toolkit and no live Composio backend - // configured in tests, so the fetch fails/returns nothing — the tool - // must still succeed, with an empty `response_fields` + explanatory note - // rather than erroring or blocking the search. - let tmp = TempDir::new().unwrap(); - // Seed an entry for a *different* action so the toolkit's cache is - // populated but this slug is absent from it — the "known toolkit, - // unknown action" branch of the degrade path. - let mut entries = std::collections::HashMap::new(); - entries.insert("SLACK_SOMETHING_ELSE".to_string(), vec!["ok".to_string()]); - crate::openhuman::tinyflows::caps::seed_response_fields_cache("slack", entries); + // The seeded action has no output schema — the tool must still succeed, + // with an empty `output_fields` list rather than erroring. Uses its own + // fictional toolkit key (never the real `"slack"` key) — `slack` is a + // statically-catalogued toolkit elsewhere in this test suite (e.g. + // `ops_tests.rs`'s `validate_tool_contracts` tests), and this fixture's + // `is_curated: false` would otherwise race with those tests over the + // shared process-global `LIVE_CATALOG_CACHE` entry for `"slack"`. + seed_live_catalog_cache( + "slackschematest", + vec![ToolContract { + slug: "SLACKSCHEMATEST_SEND_MESSAGE".to_string(), + toolkit: "slackschematest".to_string(), + description: None, + required_args: vec!["channel".to_string()], + input_schema: None, + output_fields: Vec::new(), + output_schema: None, + primary_array_path: None, + is_curated: false, + }], + ); + let tmp = TempDir::new().unwrap(); let tool = SearchToolCatalogTool::new(test_config(&tmp)); let result = tool - .execute(json!({ "query": "send", "toolkit": "slack" })) + .execute(json!({ "query": "send", "toolkit": "slackschematest" })) .await .unwrap(); assert!(!result.is_error, "{}", result.output()); @@ -257,14 +315,58 @@ async fn search_tool_catalog_degrades_gracefully_when_output_schema_unknown() { let results = parsed["results"].as_array().unwrap(); assert!(!results.is_empty(), "slack catalog should have entries"); for r in results { - assert!(r["response_fields"].as_array().unwrap().is_empty()); - assert_eq!( - r["response_fields_note"], - "output shape unknown — dry-run to verify the binding resolves" - ); + assert!(r["output_fields"].as_array().unwrap().is_empty()); + assert_eq!(r["featured"], false); } } +// ── get_tool_contract ──────────────────────────────────────────────────────── + +#[tokio::test] +async fn get_tool_contract_returns_the_full_seeded_contract() { + seed_live_catalog_cache("gmail", vec![seeded_gmail_send_contract()]); + let tmp = TempDir::new().unwrap(); + let tool = GetToolContractTool::new(test_config(&tmp)); + assert_eq!(tool.name(), "get_tool_contract"); + assert_eq!(tool.permission_level(), PermissionLevel::None); + assert!(!tool.external_effect()); + + let result = tool + .execute(json!({ "slug": "GMAIL_SEND_EMAIL" })) + .await + .unwrap(); + assert!(!result.is_error, "{}", result.output()); + let parsed: Value = serde_json::from_str(&result.output()).unwrap(); + assert_eq!(parsed["slug"], "GMAIL_SEND_EMAIL"); + assert_eq!(parsed["toolkit"], "gmail"); + assert_eq!(parsed["required_args"], json!(["to", "body"])); + assert_eq!(parsed["output_fields"], json!(["id", "threadId"])); + assert!(parsed["output_schema"].is_object()); + assert!(parsed["input_schema"].is_object()); +} + +#[tokio::test] +async fn get_tool_contract_missing_slug_is_error() { + let tmp = TempDir::new().unwrap(); + let tool = GetToolContractTool::new(test_config(&tmp)); + let result = tool.execute(json!({})).await.unwrap(); + assert!(result.is_error); + assert!(result.output().contains("Missing 'slug'")); +} + +#[tokio::test] +async fn get_tool_contract_rejects_a_hallucinated_slug() { + seed_live_catalog_cache("gmail", vec![seeded_gmail_send_contract()]); + let tmp = TempDir::new().unwrap(); + let tool = GetToolContractTool::new(test_config(&tmp)); + let result = tool + .execute(json!({ "slug": "GMAIL_DOES_NOT_EXIST" })) + .await + .unwrap(); + assert!(result.is_error); + assert!(result.output().contains("not a real action")); +} + // ── dry_run_workflow ───────────────────────────────────────────────────────── #[test] @@ -366,12 +468,7 @@ async fn dry_run_catches_unwired_required_composio_arg() { // NOTE: the cache is process-global and other tests seed the `gmail` // toolkit too — keep every seeding of GMAIL_SEND_EMAIL identical // (`to` + `body`) so test order can't change the outcome. - let mut entries = std::collections::HashMap::new(); - entries.insert( - "GMAIL_SEND_EMAIL".to_string(), - vec!["to".to_string(), "body".to_string()], - ); - crate::openhuman::tinyflows::caps::seed_required_args_cache("gmail", entries); + seed_live_catalog_cache("gmail", vec![seeded_gmail_send_contract()]); let tmp = TempDir::new().unwrap(); let tool = DryRunWorkflowTool::new(policy(AutonomyLevel::Supervised), test_config(&tmp)); @@ -561,12 +658,7 @@ async fn dry_run_flags_tool_call_error_when_on_error_is_route() { // got far enough to trace an `=`-expression before the preflight error). // Seed the same schema as `dry_run_catches_unwired_required_composio_arg` // (process-global cache; keep the arg list identical across tests). - let mut entries = std::collections::HashMap::new(); - entries.insert( - "GMAIL_SEND_EMAIL".to_string(), - vec!["to".to_string(), "body".to_string()], - ); - crate::openhuman::tinyflows::caps::seed_required_args_cache("gmail", entries); + seed_live_catalog_cache("gmail", vec![seeded_gmail_send_contract()]); let tool = DryRunWorkflowTool::new( policy(AutonomyLevel::Supervised), @@ -607,12 +699,7 @@ async fn dry_run_flags_tool_call_error_when_on_error_is_route() { async fn dry_run_flags_tool_call_error_when_on_error_is_continue() { // Same case as above, but `on_error: "continue"` — the other policy that // converts a node failure into routed data instead of failing the run. - let mut entries = std::collections::HashMap::new(); - entries.insert( - "GMAIL_SEND_EMAIL".to_string(), - vec!["to".to_string(), "body".to_string()], - ); - crate::openhuman::tinyflows::caps::seed_required_args_cache("gmail", entries); + seed_live_catalog_cache("gmail", vec![seeded_gmail_send_contract()]); let tool = DryRunWorkflowTool::new( policy(AutonomyLevel::Supervised), @@ -766,14 +853,16 @@ async fn dry_run_passes_when_agent_uses_input_context_instead_of_prompt_expressi ); } +/// (systemic tool-contract fix, Part 2b) A missing required Composio arg is +/// now a HARD REJECT at `revise_workflow` — `validate_tool_contracts` runs +/// ahead of the older advisory `graph_wiring_warnings` check and catches the +/// exact same condition first, so the graph never gets far enough to merely +/// warn about it. `graph_wiring_warnings`'s own required-arg warning (still +/// exercised directly in `ops_tests.rs`) stays as a defense-in-depth +/// fallback for any caller that doesn't also run `validate_tool_contracts`. #[tokio::test] -async fn revise_workflow_warns_on_unwired_required_composio_arg() { - let mut entries = std::collections::HashMap::new(); - entries.insert( - "GMAIL_SEND_EMAIL".to_string(), - vec!["to".to_string(), "body".to_string()], - ); - crate::openhuman::tinyflows::caps::seed_required_args_cache("gmail", entries); +async fn revise_workflow_rejects_a_missing_required_composio_arg() { + seed_live_catalog_cache("gmail", vec![seeded_gmail_send_contract()]); let tmp = TempDir::new().unwrap(); let tool = ReviseWorkflowTool::new(test_config(&tmp)); @@ -794,23 +883,15 @@ async fn revise_workflow_warns_on_unwired_required_composio_arg() { .await .unwrap(); - assert!(!result.is_error, "{}", result.output()); - let parsed: Value = serde_json::from_str(&result.output()).unwrap(); - let warnings = parsed["warnings"].as_array().unwrap(); assert!( - warnings.iter().any(|w| { - let w = w.as_str().unwrap_or_default(); - w.contains("`to`") && w.contains("send") - }), - "expected a warning naming node `send` and arg `to`: {warnings:?}" - ); - // `body` is wired (expression) — no warning for it. - assert!( - !warnings - .iter() - .any(|w| w.as_str().unwrap_or_default().contains("`body`")), - "wired arg must not warn: {warnings:?}" + result.is_error, + "a missing required arg must now hard-reject" ); + let output = result.output(); + assert!(output.contains("send"), "{output}"); + assert!(output.contains("`to`"), "{output}"); + // `body` is wired (expression) — never named as missing. + assert!(!output.contains("`body`"), "{output}"); } // ── save_workflow ──────────────────────────────────────────────────────────── diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index e5efbf1b8..0de76055e 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -210,6 +210,170 @@ pub(crate) async fn graph_wiring_warnings(config: &Config, graph: &WorkflowGraph )); } } + + warnings.extend(graph_output_field_warnings(config, graph).await); + warnings.extend(graph_split_out_path_warnings(config, graph).await); + warnings +} + +/// Author-time WARN (systemic tool-contract fix, Part 2c): any +/// `=nodes..item.json.` binding — anywhere in the graph, not just +/// `tool_call` args — whose `` names a `tool_call` node calling a REAL +/// Composio action with a KNOWN live output schema, but whose `` is +/// not one of that action's real `output_fields`. Advisory, not fatal: a +/// binding to an unknown field could still resolve to something useful at +/// runtime for an action whose output schema is incomplete, so this warns +/// rather than rejects — mirroring `graph_wiring_warnings`'s existing +/// required-arg warnings. +/// +/// Skipped entirely when the referenced action's output schema is +/// **unknown** (`ToolContract::output_schema` is `None`) — there is nothing +/// real to check the field against, so warning would just be noise (or a +/// false positive for a still-legitimate binding). Also skipped for a +/// binding that dereferences `.item.` without `.json` on an +/// enveloping node — that shape is already a HARD reject in +/// [`validate_binding_resolvability`], not a warning here. +async fn graph_output_field_warnings(config: &Config, graph: &WorkflowGraph) -> Vec { + use crate::openhuman::memory_sync::composio::providers::toolkit_from_slug; + use crate::openhuman::tinyflows::caps::fetch_live_toolkit_catalog; + + let mut warnings = Vec::new(); + for node in &graph.nodes { + for (location, expr) in collect_expressions(&node.config) { + let Some((ref_id, has_json, field)) = parse_node_binding(&expr) else { + continue; + }; + if !has_json { + continue; + } + let Some(ref_node) = graph.node(&ref_id) else { + continue; + }; + if ref_node.kind != NodeKind::ToolCall { + continue; + } + let Some(ref_slug) = ref_node.config.get("slug").and_then(Value::as_str) else { + continue; + }; + if ref_slug.starts_with('=') || ref_slug.starts_with("oh:") { + continue; + } + let Some(ref_toolkit) = toolkit_from_slug(ref_slug) else { + continue; + }; + let Some(catalog) = fetch_live_toolkit_catalog(config, &ref_toolkit).await else { + continue; + }; + let Some(contract) = catalog + .iter() + .find(|c| c.slug.eq_ignore_ascii_case(ref_slug)) + else { + continue; + }; + // Output schema unknown — nothing real to check `field` against. + if contract.output_schema.is_none() { + continue; + } + if !contract.output_fields.iter().any(|f| f == &field) { + tracing::warn!( + target: "flows", + node = %node.id, + %location, + ref_node = %ref_id, + ref_slug, + %field, + output_fields = ?contract.output_fields, + "[flows] wiring check: downstream binding reads a field not in the tool's real output_fields" + ); + warnings.push(format!( + "Node '{}': binding `{location}` (`{expr}`) reads field `{field}` off \ + tool_call `{ref_id}` (`{ref_slug}`), but that is not one of its real \ + output fields ({}) — call get_tool_contract {{ slug: \"{ref_slug}\" }} to \ + see the real output field names.", + node.id, + contract.output_fields.join(", "), + )); + } + } + } + warnings +} + +/// Author-time WARN/suggest (systemic tool-contract fix, Part 2d): a +/// `split_out` node whose direct predecessor is a `tool_call` calling a REAL +/// Composio action with a KNOWN `primary_array_path` (see +/// [`crate::openhuman::tinyflows::caps::compute_primary_array_path`]), but +/// whose configured `config.path` doesn't match the `json.` +/// convention (dereferencing the `{json,text,raw}` envelope's own `json` +/// field, then the action's real array property). Advisory: a mismatched +/// path degrades the fan-out (or silently produces one item instead of many) +/// rather than crashing, so this suggests the real path instead of rejecting. +/// +/// Skipped when the predecessor's action has no known `primary_array_path` +/// (nothing to suggest), or when `split_out`'s predecessor isn't a +/// `tool_call` at all (no envelope/array-path convention applies). +async fn graph_split_out_path_warnings(config: &Config, graph: &WorkflowGraph) -> Vec { + use crate::openhuman::memory_sync::composio::providers::toolkit_from_slug; + use crate::openhuman::tinyflows::caps::fetch_live_toolkit_catalog; + + let mut warnings = Vec::new(); + for node in &graph.nodes { + if node.kind != NodeKind::SplitOut { + continue; + } + let configured_path = node.config.get("path").and_then(Value::as_str); + + for edge in graph.edges.iter().filter(|e| e.to_node == node.id) { + let Some(pred) = graph.node(&edge.from_node) else { + continue; + }; + if pred.kind != NodeKind::ToolCall { + continue; + } + let Some(pred_slug) = pred.config.get("slug").and_then(Value::as_str) else { + continue; + }; + if pred_slug.starts_with('=') || pred_slug.starts_with("oh:") { + continue; + } + let Some(pred_toolkit) = toolkit_from_slug(pred_slug) else { + continue; + }; + let Some(catalog) = fetch_live_toolkit_catalog(config, &pred_toolkit).await else { + continue; + }; + let Some(contract) = catalog + .iter() + .find(|c| c.slug.eq_ignore_ascii_case(pred_slug)) + else { + continue; + }; + let Some(primary) = contract.primary_array_path.as_deref() else { + continue; + }; + let expected = format!("json.{primary}"); + if configured_path != Some(expected.as_str()) { + tracing::warn!( + target: "flows", + node = %node.id, + predecessor = %pred.id, + pred_slug, + configured_path, + %expected, + "[flows] wiring check: split_out.path does not match the predecessor tool's real array path" + ); + let configured_display = configured_path + .map(|p| format!("\"{p}\"")) + .unwrap_or_else(|| "unset".to_string()); + warnings.push(format!( + "Node '{}': split_out.path is {configured_display} but its predecessor \ + tool_call `{}` (`{pred_slug}`) wraps its real array at `{expected}` — set \ + config.path to \"{expected}\" to fan out over the actual response list.", + node.id, pred.id, + )); + } + } + } warnings } @@ -529,6 +693,156 @@ pub(crate) fn validate_binding_resolvability(graph: &WorkflowGraph) -> Vec Vec { + use crate::openhuman::memory_sync::composio::providers::{ + catalog_for_toolkit, get_provider, toolkit_from_slug, + }; + use crate::openhuman::tinyflows::caps::{fetch_live_toolkit_catalog, missing_required_args}; + + let mut errors = Vec::new(); + for node in &graph.nodes { + if node.kind != NodeKind::ToolCall { + continue; + } + let Some(slug) = node.config.get("slug").and_then(Value::as_str) else { + continue; + }; + // `=`-derived slugs resolve from upstream/trigger data at runtime — + // nothing to check statically. Native `oh:` tools have no Composio + // contract. + if slug.starts_with('=') || slug.starts_with("oh:") { + continue; + } + let Some(toolkit) = toolkit_from_slug(slug) else { + continue; + }; + let Some(catalog) = fetch_live_toolkit_catalog(config, &toolkit).await else { + tracing::debug!( + target: "flows", + node = %node.id, + %slug, + %toolkit, + "[flows] tool-contract check: live catalog fetch failed — skipping (best-effort, never false-rejects)" + ); + continue; + }; + + let Some(contract) = catalog.iter().find(|c| c.slug.eq_ignore_ascii_case(slug)) else { + tracing::warn!( + target: "flows", + node = %node.id, + %slug, + %toolkit, + "[flows] tool-contract check: slug is not a real action in the live catalog — rejecting" + ); + errors.push(format!( + "Node '{}': `{slug}` is not a real action in the `{toolkit}` toolkit's live \ + Composio catalog — use search_tool_catalog {{ query: ..., toolkit: \"{toolkit}\" \ + }} to find a real action slug.", + node.id + )); + continue; + }; + + // Mirror `flow_tool_allowed`'s Path A: a toolkit OpenHuman ships a + // static curated catalog for is a hard curated-only allowlist at + // RUNTIME — `find_curated` rejects any slug that isn't one of the + // curated actions, regardless of whether it's a real live action. + // `search_tool_catalog`/`get_tool_contract` deliberately surface + // real-but-uncurated actions too (ranking signal only, never + // hidden — see `ToolContract::is_curated`'s doc), so without this + // check a graph could pass authoring/save with a real-but-uncurated + // action on a curated toolkit and then fail every run with "tool + // not permitted". Hold authoring to the same bar the runtime gate + // enforces instead of loosening the runtime gate. + let has_static_catalog = get_provider(&toolkit) + .and_then(|p| p.curated_tools()) + .or_else(|| catalog_for_toolkit(&toolkit)) + .is_some(); + if has_static_catalog && !contract.is_curated { + tracing::warn!( + target: "flows", + node = %node.id, + %slug, + %toolkit, + "[flows] tool-contract check: slug is real but not curated for a statically-catalogued toolkit — rejecting to match the runtime allowlist" + ); + errors.push(format!( + "Node '{}': `{slug}` is a real `{toolkit}` action but not one of OpenHuman's \ + curated actions for `{toolkit}` — the runtime tool gate only allows curated \ + actions for toolkits with a curated catalog, so this would be rejected on \ + every run. Use search_tool_catalog {{ query: ..., toolkit: \"{toolkit}\" }} and \ + pick a result with `featured: true`.", + node.id + )); + continue; + } + + let args = node.config.get("args").cloned().unwrap_or(Value::Null); + let missing = missing_required_args(&contract.required_args, &args); + if !missing.is_empty() { + tracing::warn!( + target: "flows", + node = %node.id, + %slug, + ?missing, + "[flows] tool-contract check: required arg(s) missing or null — rejecting" + ); + let list = missing + .iter() + .map(|m| format!("`{m}`")) + .collect::>() + .join(", "); + errors.push(format!( + "Node '{}': tool_call `{slug}` is missing required arg(s) {list} — wire each \ + from an upstream node's output, e.g. \"{}\": \ + \"=nodes..item.json.\" (call get_tool_contract {{ slug: \ + \"{slug}\" }} for the exact required_args list).", + node.id, missing[0] + )); + } + } + errors +} + /// Validates a candidate graph without persisting it — the same /// migrate/validate path `flows_create` and `ProposeWorkflowTool` use — and /// reports structural errors alongside non-fatal trigger warnings diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index d6c4fe9b9..8f1370290 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -1823,6 +1823,298 @@ fn binding_to_agent_with_matching_schema_is_accepted() { ); } +// ── validate_tool_contracts (systemic tool-contract fix, Part 2) ─────────── +// +// The live-catalog cache is process-global (`LIVE_CATALOG_CACHE`) — every +// test below seeds the exact toolkit it needs via `seed_live_catalog_cache` +// so none of this touches a live Composio backend. + +use crate::openhuman::tinyflows::caps::{seed_live_catalog_cache, ToolContract}; + +fn seeded_slack_send_contract() -> ToolContract { + ToolContract { + slug: "SLACK_SEND_MESSAGE".to_string(), + toolkit: "slack".to_string(), + description: None, + required_args: vec!["channel".to_string(), "text".to_string()], + input_schema: None, + output_fields: vec!["ts".to_string(), "channel".to_string()], + output_schema: Some(json!({ + "type": "object", + "properties": { "ts": {"type": "string"}, "channel": {"type": "string"} } + })), + primary_array_path: None, + // `slack` ships a static curated catalog (`catalog_for_toolkit`), so + // `validate_tool_contracts` now enforces the same curated-only bar + // `flow_tool_allowed`'s Path A does at runtime (Codex feedback on + // this PR) — this fixture models a real curated Slack action, not + // an uncurated one, since these tests exercise the required-arg / + // hallucinated-slug checks rather than the curation gate itself. + is_curated: true, + } +} + +#[tokio::test] +async fn validate_tool_contracts_rejects_a_hallucinated_slug() { + seed_live_catalog_cache("slack", vec![seeded_slack_send_contract()]); + let config = Config::default(); + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "SLACK_POST_MESSAGE_TO_CHANNEL", + "args": { "channel": "#general", "text": "hi" } } } + ], + "edges": [ { "from_node": "t", "to_node": "post" } ] + })); + let errors = validate_tool_contracts(&config, &g).await; + assert_eq!(errors.len(), 1, "{errors:?}"); + assert!(errors[0].contains("post"), "{}", errors[0]); + assert!( + errors[0].contains("SLACK_POST_MESSAGE_TO_CHANNEL"), + "{}", + errors[0] + ); + assert!(errors[0].contains("search_tool_catalog"), "{}", errors[0]); +} + +#[tokio::test] +async fn validate_tool_contracts_rejects_a_missing_required_arg() { + seed_live_catalog_cache("slack", vec![seeded_slack_send_contract()]); + let config = Config::default(); + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "SLACK_SEND_MESSAGE", + "args": { "channel": "#general" } } } + ], + "edges": [ { "from_node": "t", "to_node": "post" } ] + })); + let errors = validate_tool_contracts(&config, &g).await; + assert_eq!(errors.len(), 1, "{errors:?}"); + assert!(errors[0].contains("`text`"), "{}", errors[0]); + assert!(errors[0].contains("get_tool_contract"), "{}", errors[0]); +} + +#[tokio::test] +async fn validate_tool_contracts_passes_a_fully_wired_real_slug() { + seed_live_catalog_cache("slack", vec![seeded_slack_send_contract()]); + let config = Config::default(); + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "SLACK_SEND_MESSAGE", + "args": { "channel": "#general", "text": "hi" } } } + ], + "edges": [ { "from_node": "t", "to_node": "post" } ] + })); + let errors = validate_tool_contracts(&config, &g).await; + assert!(errors.is_empty(), "{errors:?}"); +} + +/// (Codex feedback on this PR) `notion` ships a static curated catalog +/// (`catalog_for_toolkit`), so at RUNTIME `flow_tool_allowed`'s Path A +/// hard-rejects any slug `find_curated` doesn't recognize — even a real, +/// live action. Without this check, a real-but-uncurated action for a +/// statically-catalogued toolkit would pass authoring/save here and then +/// fail every single run as "tool not permitted". Uses its own toolkit key +/// (`notion`, not `slack`/`gmail`) since it seeds different `is_curated` +/// content than every other test sharing those keys. +#[tokio::test] +async fn validate_tool_contracts_rejects_a_real_but_uncurated_action_on_a_statically_catalogued_toolkit( +) { + seed_live_catalog_cache( + "notion", + vec![ToolContract { + slug: "NOTION_UNCURATED_ACTION".to_string(), + toolkit: "notion".to_string(), + description: None, + required_args: vec![], + input_schema: None, + output_fields: vec![], + output_schema: None, + primary_array_path: None, + // Real (a live catalog fetch found it), but NOT one of + // OpenHuman's curated Notion actions. + is_curated: false, + }], + ); + let config = Config::default(); + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "NOTION_UNCURATED_ACTION", "args": {} } } + ], + "edges": [ { "from_node": "t", "to_node": "post" } ] + })); + let errors = validate_tool_contracts(&config, &g).await; + assert_eq!(errors.len(), 1, "{errors:?}"); + assert!( + errors[0].contains("NOTION_UNCURATED_ACTION"), + "{}", + errors[0] + ); + assert!(errors[0].contains("curated"), "{}", errors[0]); +} + +#[tokio::test] +async fn validate_tool_contracts_skips_expression_derived_and_native_slugs() { + let config = Config::default(); + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "dynamic", "kind": "tool_call", "name": "Dynamic", + "config": { "slug": "=item.tool", "args": {} } }, + { "id": "native", "kind": "tool_call", "name": "Native", + "config": { "slug": "oh:web_search", "args": {} } } + ], + "edges": [ + { "from_node": "t", "to_node": "dynamic" }, + { "from_node": "t", "to_node": "native" } + ] + })); + let errors = validate_tool_contracts(&config, &g).await; + assert!(errors.is_empty(), "{errors:?}"); +} + +#[tokio::test] +async fn validate_tool_contracts_skips_rather_than_rejects_when_the_catalog_is_unreachable() { + // No seed for this toolkit and no live backend configured — the fetch + // fails, and the node must be SKIPPED (never false-rejected). + let config = Config::default(); + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "SOMEUNSEEDEDTOOLKIT_DO_THING", "args": {} } } + ], + "edges": [ { "from_node": "t", "to_node": "post" } ] + })); + let errors = validate_tool_contracts(&config, &g).await; + assert!( + errors.is_empty(), + "a live-catalog fetch failure must skip, not reject: {errors:?}" + ); +} + +// ── graph_wiring_warnings: required-arg advisory + output-field/split_out.path +// advisories (Part 2c/2d) ──────────────────────────────────────────────── + +/// `graph_wiring_warnings`'s own required-arg check, exercised DIRECTLY +/// (rather than through `revise_workflow`/`save_workflow`, where the newer +/// `validate_tool_contracts` hard-rejects the identical condition first — +/// see `revise_workflow_rejects_a_missing_required_composio_arg` in +/// `builder_tools_tests.rs`). Keeps this advisory code path covered for any +/// caller that consults `graph_wiring_warnings` without also running the +/// hard gate first. +#[tokio::test] +async fn graph_wiring_warnings_flags_a_missing_required_arg() { + seed_live_catalog_cache("slack", vec![seeded_slack_send_contract()]); + let config = Config::default(); + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "SLACK_SEND_MESSAGE", + "args": { "channel": "#general" } } } + ], + "edges": [ { "from_node": "t", "to_node": "post" } ] + })); + let warnings = graph_wiring_warnings(&config, &g).await; + assert!( + warnings + .iter() + .any(|w| w.contains("`text`") && w.contains("post")), + "{warnings:?}" + ); +} + +#[tokio::test] +async fn graph_wiring_warnings_flags_a_downstream_field_not_in_output_fields() { + seed_live_catalog_cache("slack", vec![seeded_slack_send_contract()]); + let config = Config::default(); + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "SLACK_SEND_MESSAGE", + "args": { "channel": "#general", "text": "hi" } } }, + { "id": "xform", "kind": "transform", "name": "Log", + // Reads a field that isn't in SLACK_SEND_MESSAGE's real + // output_fields (`ts`/`channel`) — must WARN, not reject. + "config": { "set": { "note": "=nodes.post.item.json.not_a_real_field" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "post" }, + { "from_node": "post", "to_node": "xform" } + ] + })); + let warnings = graph_wiring_warnings(&config, &g).await; + assert!( + warnings + .iter() + .any(|w| w.contains("not_a_real_field") && w.contains("post")), + "{warnings:?}" + ); +} + +#[tokio::test] +async fn graph_wiring_warnings_is_silent_when_the_downstream_field_is_real() { + seed_live_catalog_cache("slack", vec![seeded_slack_send_contract()]); + let config = Config::default(); + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "SLACK_SEND_MESSAGE", + "args": { "channel": "#general", "text": "hi" } } }, + { "id": "xform", "kind": "transform", "name": "Log", + "config": { "set": { "note": "=nodes.post.item.json.ts" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "post" }, + { "from_node": "post", "to_node": "xform" } + ] + })); + let warnings = graph_wiring_warnings(&config, &g).await; + assert!( + !warnings.iter().any(|w| w.contains("not in")), + "a real output field must not warn: {warnings:?}" + ); +} + +#[tokio::test] +async fn graph_wiring_warnings_suggests_the_real_split_out_path() { + let mut contract = seeded_slack_send_contract(); + contract.slug = "SLACKFANOUT_SEND_MESSAGE".to_string(); + contract.toolkit = "slackfanout".to_string(); + contract.primary_array_path = Some("data.messages".to_string()); + seed_live_catalog_cache("slackfanout", vec![contract]); + let config = Config::default(); + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "SLACKFANOUT_SEND_MESSAGE", + "args": { "channel": "#general", "text": "hi" } } }, + { "id": "split", "kind": "split_out", "name": "Split", + "config": { "path": "items" } } + ], + "edges": [ + { "from_node": "t", "to_node": "post" }, + { "from_node": "post", "to_node": "split" } + ] + })); + let warnings = graph_wiring_warnings(&config, &g).await; + assert!( + warnings.iter().any(|w| w.contains("json.data.messages")), + "{warnings:?}" + ); +} + // ───────────────────────────────────────────────────────────────────────────── // degrade_completed_status (PR2 — run honesty) // ───────────────────────────────────────────────────────────────────────────── diff --git a/src/openhuman/flows/tools.rs b/src/openhuman/flows/tools.rs index 0a0ebf960..2272dc433 100644 --- a/src/openhuman/flows/tools.rs +++ b/src/openhuman/flows/tools.rs @@ -23,7 +23,9 @@ use serde_json::{json, Value}; use tinyflows::model::{Node, NodeKind, WorkflowGraph}; use crate::openhuman::config::Config; -use crate::openhuman::flows::ops::{validate_and_migrate_graph, validate_binding_resolvability}; +use crate::openhuman::flows::ops::{ + validate_and_migrate_graph, validate_binding_resolvability, validate_tool_contracts, +}; use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult}; /// Max characters kept for a `config_hint` before truncation, so a long @@ -195,6 +197,24 @@ impl Tool for ProposeWorkflowTool { ))); } + // Tool-contract enforcement gate (systemic tool-contract fix, Part 2): + // reject a `tool_call` node whose slug isn't a REAL action in the + // live Composio catalog, or whose real required args aren't all + // wired — before the user ever reviews the proposal. + let contract_errors = validate_tool_contracts(&self.config, &graph).await; + if !contract_errors.is_empty() { + tracing::debug!( + target: "flows", + %name, + error_count = contract_errors.len(), + "[flows] propose_workflow: tool-contract check rejected the graph" + ); + return Ok(ToolResult::error(format!( + "{}\n\nFix these tool_call nodes and call propose_workflow again.", + contract_errors.join("\n\n") + ))); + } + let summary = build_summary(&graph); // Author-time warnings: unfired trigger kinds + unwired REQUIRED // Composio args (see `ops::graph_wiring_warnings`) — surfaced on the diff --git a/src/openhuman/tinyflows/caps.rs b/src/openhuman/tinyflows/caps.rs index a4999ba55..919ec1c80 100644 --- a/src/openhuman/tinyflows/caps.rs +++ b/src/openhuman/tinyflows/caps.rs @@ -25,7 +25,7 @@ use tinyflows::model::WorkflowGraph; use crate::openhuman::agent::harness::definition::SandboxMode; use crate::openhuman::composio::client::{ - create_composio_client, direct_execute, ComposioClientKind, + create_composio_client, direct_execute, direct_list_tools, ComposioClientKind, }; use crate::openhuman::config::{Config, HttpRequestConfig}; use crate::openhuman::credentials::{HttpCredential, HttpCredentialsStore}; @@ -1108,17 +1108,35 @@ pub(crate) fn http_cred_name(conn: &str) -> Option<&str> { /// // unchanged (a connected-but-uncurated action on a cataloged toolkit is /// // still rejected — the catalog is the tighter allowlist there). /// -/// Returns whether `slug` may be invoked as a flow `tool_call`, given (only when -/// needed) the user's live connected-toolkit slug set. +/// // (systemic tool-contract fix, PR2) Path B is now further tightened rather +/// // than loosened: on top of the (0.3) connected-toolkit check, the SLUG +/// // ITSELF must be a genuine action in that toolkit's LIVE Composio catalog +/// // (`fetch_live_toolkit_catalog`) — previously any string sharing the +/// // connected toolkit's prefix passed (e.g. a hallucinated/typo'd +/// // `STRIPE_DOES_NOT_EXIST` for a connected `stripe`), with no per-user +/// // read/write/admin scope check at all. Now: existence is broadened to the +/// // real catalog (a real-but-uncurated action is allowed), but scope gating +/// // is ADDED via [`classify_unknown`] — strictly narrower than before, never +/// // looser. /// -/// Split out from [`is_curated_flow_tool`] as a pure function so the two decision -/// paths are unit-testable without a live Composio backend: `connected_toolkits` -/// is `None` when the toolkit has a static catalog (the connected set is never -/// consulted then) or when the connected set could not be fetched (fail-closed). -async fn flow_tool_allowed(slug: &str, connected_toolkits: Option<&[String]>) -> bool { +/// Returns whether `slug` may be invoked as a flow `tool_call`, given (only when +/// needed) the user's live connected-toolkit slug set. `config` is only used by +/// Path B's live-catalog fetch (fed through [`fetch_live_toolkit_catalog`], +/// which is itself cached — a seeded test cache never touches the network). +/// +/// Split out from [`is_curated_flow_tool`] as a (mostly) pure function so the +/// two decision paths are unit-testable without a live Composio backend: +/// `connected_toolkits` is `None` when the toolkit has a static catalog (the +/// connected set is never consulted then) or when the connected set could not +/// be fetched (fail-closed). +async fn flow_tool_allowed( + config: &Config, + slug: &str, + connected_toolkits: Option<&[String]>, +) -> bool { use crate::openhuman::memory_sync::composio::providers::{ - catalog_for_toolkit, find_curated, get_provider, load_user_scope_or_default, - toolkit_from_slug, + catalog_for_toolkit, classify_unknown, find_curated, get_provider, + load_user_scope_or_default, toolkit_from_slug, }; let Some(toolkit) = toolkit_from_slug(slug) else { @@ -1142,19 +1160,47 @@ async fn flow_tool_allowed(slug: &str, connected_toolkits: Option<&[String]>) -> return allowed; } - // Path B (0.3): no static catalog — allow iff the user has a live ACTIVE - // Composio connection for this toolkit. Made-up toolkits are never connected. - match connected_toolkits { - Some(toolkits) => { - let connected = toolkits.iter().any(|t| t.eq_ignore_ascii_case(&toolkit)); - tracing::debug!(target: "flows", %slug, %toolkit, connected, "[flows] tool_call curation: live connected-toolkit allowlist decision"); - connected - } + // Path B: no static catalog. First, the (0.3) toolkit-level gate — allow + // only when the user has a live ACTIVE Composio connection for it. A + // made-up toolkit is never connected, so it rejects right here without + // ever reaching the live-catalog fetch below. + let connected = match connected_toolkits { + Some(toolkits) => toolkits.iter().any(|t| t.eq_ignore_ascii_case(&toolkit)), None => { tracing::warn!(target: "flows", %slug, %toolkit, "[flows] tool_call curation: reject — no static catalog and the connected-toolkit set was unavailable (fail-closed)"); false } + }; + if !connected { + tracing::debug!(target: "flows", %slug, %toolkit, "[flows] tool_call curation: reject — toolkit has no static catalog and is not connected"); + return false; } + + // Second, the (systemic tool-contract fix) slug-existence gate — the + // exact slug must be a genuine action in the toolkit's LIVE Composio + // catalog, not merely share its prefix. A fetch failure fails closed + // (never falls back to "any slug with the right prefix passes"). + let Some(live_catalog) = fetch_live_toolkit_catalog(config, &toolkit).await else { + tracing::warn!(target: "flows", %slug, %toolkit, "[flows] tool_call curation: reject — connected but the live catalog fetch failed (fail-closed)"); + return false; + }; + if live_catalog + .iter() + .find(|c| c.slug.eq_ignore_ascii_case(slug)) + .is_none() + { + tracing::debug!(target: "flows", %slug, %toolkit, "[flows] tool_call curation: reject — slug is not a real action in this toolkit's live catalog"); + return false; + } + + // Finally, scope-gate the same way a curated action is — via the + // classify_unknown heuristic (mirrors + // `providers::is_action_visible_with_pref`'s uncurated branch), which the + // pre-fix Path B never applied at all. + let pref = load_user_scope_or_default(&toolkit).await; + let allowed = pref.allows(classify_unknown(slug)); + tracing::debug!(target: "flows", %slug, %toolkit, allowed, "[flows] tool_call curation: live catalog + scope decision"); + allowed } /// Whether `slug`'s toolkit lacks a static curated catalog, i.e. the curation @@ -1222,7 +1268,7 @@ async fn is_curated_flow_tool(config: &Config, slug: &str) -> bool { } else { None }; - flow_tool_allowed(slug, connected.as_deref()).await + flow_tool_allowed(config, slug, connected.as_deref()).await } /// Finds the connected account a Composio `connection_id` refers to within a @@ -1311,175 +1357,305 @@ pub struct OpenHumanTools { /// search / media generation / file / shell / etc. — the full toolset. pub(crate) const NATIVE_TOOL_PREFIX: &str = "oh:"; -/// Process-level cache for [`composio_required_args`]: toolkit → (uppercase -/// action slug → required top-level arg names). One `list_tools` fetch per -/// toolkit per process; schemas are effectively static within a session. -static REQUIRED_ARGS_CACHE: std::sync::OnceLock< - std::sync::Mutex< - std::collections::HashMap>>, - >, +/// One Composio action's LIVE, ground-truth contract — the source of truth +/// [Part 1 of the systemic tool-contract fix] grounds the Workflow builder +/// against, replacing the old "guess a slug/arg/field/path and hope" +/// authoring flow. +/// +/// Everything on this type comes straight from Composio's own v3 `/tools` +/// listing (`ComposioToolFunction` — `parameters`/`output_parameters`), never +/// from OpenHuman's static curated catalog: `required_args`/`input_schema` +/// are the action's real input contract, `output_fields`/`output_schema`/ +/// `primary_array_path` are its real output contract. `is_curated` is the +/// ONE field that cross-references the static catalog — purely for ranking +/// (curated matches first in `search_tool_catalog`), never for filtering: +/// a real, uncurated action still produces a full `ToolContract`. +#[derive(Debug, Clone, serde::Serialize)] +pub struct ToolContract { + /// The Composio action slug, e.g. `"GMAIL_SEND_EMAIL"`. + pub slug: String, + /// The lowercase toolkit slug this action belongs to, e.g. `"gmail"`. + pub toolkit: String, + /// Human-readable description shown to the model, when Composio + /// publishes one for this action. + pub description: Option, + /// Required top-level input argument names (`input_schema`'s + /// `required` array). Empty when the action takes no required args — + /// NOT the same as "schema unknown" (there is no such state here: an + /// action always has SOME input schema, even if empty). + pub required_args: Vec, + /// The action's full input JSON Schema, verbatim from Composio. + pub input_schema: Option, + /// Top-level output/response field names — empty when + /// [`Self::output_schema`] is `None` (unknown) OR when it's `Some` but + /// names no top-level properties; check `output_schema` to tell those + /// two apart. + pub output_fields: Vec, + /// The action's full output JSON Schema, when Composio publishes one. + /// `None` means "unknown to this listing", not "empty" — mirrors + /// [`composio_response_fields`]'s long-standing contract. + pub output_schema: Option, + /// Dotted path (relative to the envelope's own `json` field — prefix + /// with `"json."` for a `split_out.path`, e.g. `"json.data.messages"`) + /// to the first array-typed property in [`Self::output_schema`], via + /// [`compute_primary_array_path`]. `None` when the output schema is + /// unknown or names no array property. + pub primary_array_path: Option, + /// Whether this action is ALSO one of OpenHuman's hand-curated actions + /// for its toolkit (`catalog_for_toolkit` / + /// `ComposioProvider::curated_tools`) — ranking signal only; a `false` + /// here never hides a real action, it only sorts it after curated ones. + pub is_curated: bool, +} + +/// Process-level cache backing [`fetch_live_toolkit_catalog`]: lowercase +/// toolkit slug → every [`ToolContract`] the LIVE Composio catalog published +/// for it. One fetch per toolkit per process — schemas are effectively +/// static within a session. +/// +/// Replaces the narrower `REQUIRED_ARGS_CACHE` / `RESPONSE_FIELDS_CACHE` +/// pair (single-purpose, args-only / fields-only) that predated this fix: +/// [`composio_required_args`] and [`composio_response_fields`] now both +/// delegate to this one cache/fetch instead of each running its own +/// independent `composio_list_tools` round trip. +static LIVE_CATALOG_CACHE: std::sync::OnceLock< + std::sync::Mutex>>, > = std::sync::OnceLock::new(); -/// Seeds the required-args cache for a toolkit — test hook so preflight -/// behavior can be exercised without a live Composio backend. +/// Seeds the live-catalog cache for a toolkit — test hook so preflight / +/// search / contract-validation behavior can be exercised without a live +/// Composio backend. Replaces the narrower `seed_required_args_cache` / +/// `seed_response_fields_cache` test hooks this fix removes. #[cfg(test)] -pub(crate) fn seed_required_args_cache( - toolkit: &str, - entries: std::collections::HashMap>, -) { - REQUIRED_ARGS_CACHE +pub(crate) fn seed_live_catalog_cache(toolkit: &str, contracts: Vec) { + LIVE_CATALOG_CACHE .get_or_init(Default::default) .lock() - .expect("required-args cache poisoned") - .insert(toolkit.to_string(), entries); + .expect("live catalog cache poisoned") + .insert(toolkit.trim().to_ascii_lowercase(), contracts); +} + +/// Fetches a toolkit's tool schemas STRAIGHT from the Composio client, +/// deliberately bypassing `composio::ops::composio_list_tools`'s curated- +/// whitelist filter (Direct mode's `filter_list_tools_response_for_direct` — +/// Backend mode's branch of `composio_list_tools` never filters at all, so +/// this is behavior-identical to it there) — so [`fetch_live_toolkit_catalog`] +/// grounds against the FULL live catalog (every real action, connected or +/// not, curated or not), not the narrower curated subset the pre-fix +/// `search_tool_catalog` searched. +/// +/// - **Backend mode** calls [`crate::openhuman::composio::client::ComposioClient::list_tools`] +/// directly — already unfiltered (`composio_list_tools`'s backend branch +/// applies no filter either), so this is not a behavior change there. +/// - **Direct mode** calls [`direct_list_tools`] directly instead of going +/// through `composio_list_tools`'s direct branch, which DOES apply +/// `filter_list_tools_response_for_direct` — that's the filter this +/// function exists to skip. `direct_list_tools` itself never filters; the +/// curation is layered on entirely by its `composio_list_tools` caller. +/// +/// Returns `None` on any client-construction or network failure — callers +/// degrade to "catalog unknown" rather than blocking. +async fn fetch_raw_toolkit_tools( + config: &Config, + toolkit: &str, +) -> Option { + let kind = create_composio_client(config) + .map_err(|e| { + tracing::debug!(target: "flows", %toolkit, error = %e, "[flows] live catalog: composio client unavailable — skipping"); + e + }) + .ok()?; + match kind { + ComposioClientKind::Backend(client) => client + .list_tools(Some(&[toolkit.to_string()]), None) + .await + .map_err(|e| { + tracing::debug!(target: "flows", %toolkit, error = %e, "[flows] live catalog: backend fetch failed — skipping"); + e + }) + .ok(), + ComposioClientKind::Direct(tool) => direct_list_tools(&tool, &[toolkit.to_string()], None) + .await + .map_err(|e| { + tracing::debug!(target: "flows", %toolkit, error = %e, "[flows] live catalog: direct fetch failed — skipping"); + e + }) + .ok(), + } +} + +/// Fetches (or returns the cached) FULL LIVE Composio catalog for one +/// toolkit — every real action Composio publishes for it, mapped into +/// [`ToolContract`]s — regardless of OpenHuman's curated whitelist or the +/// user's connection state. This is the ground-truth source the Workflow +/// builder's discovery (`search_tool_catalog`/`get_tool_contract`) and +/// enforcement (`ops::validate_tool_contracts`) both consult. +/// +/// Degrades gracefully when an action's listing carries no +/// `output_parameters` (unknown to this crate, or genuinely unpublished by +/// Composio for it) — `output_fields` is empty, `primary_array_path` is +/// `None`, and `output_schema` stays `None` so callers can distinguish "no +/// fields" from "schema unknown". Applies identically whether the listing +/// came from Direct mode (which threads `output_parameters` through +/// natively) or Backend mode (whatever its own proxy response carries under +/// the same field — may legitimately be absent). +/// +/// `None` when the fetch itself failed (no client, network error) — +/// distinct from `Some(vec![])`, which means the toolkit is real but +/// currently publishes zero actions. +pub(crate) async fn fetch_live_toolkit_catalog( + config: &Config, + toolkit: &str, +) -> Option> { + use crate::openhuman::memory_sync::composio::providers::{ + catalog_for_toolkit, find_curated, get_provider, + }; + + let key = toolkit.trim().to_ascii_lowercase(); + if key.is_empty() { + return None; + } + + if let Some(cached) = LIVE_CATALOG_CACHE + .get_or_init(Default::default) + .lock() + .ok()? + .get(&key) + { + return Some(cached.clone()); + } + + tracing::debug!(target: "flows", toolkit = %key, "[flows] live catalog: fetching (cache miss)"); + let resp = fetch_raw_toolkit_tools(config, &key).await?; + + let curated_catalog = get_provider(&key) + .and_then(|p| p.curated_tools()) + .or_else(|| catalog_for_toolkit(&key)); + + let contracts: Vec = resp + .tools + .iter() + .map(|tool| { + let slug = tool.function.name.clone(); + let required_args = tool + .function + .parameters + .as_ref() + .and_then(|p| p.get("required")) + .and_then(Value::as_array) + .map(|arr| { + arr.iter() + .filter_map(|v| v.as_str().map(str::to_string)) + .collect() + }) + .unwrap_or_default(); + let output_fields = + response_fields_from_schema(tool.function.output_parameters.as_ref()); + let primary_array_path = + compute_primary_array_path(tool.function.output_parameters.as_ref()); + let is_curated = curated_catalog.is_some_and(|cat| find_curated(cat, &slug).is_some()); + ToolContract { + slug, + toolkit: key.clone(), + description: tool.function.description.clone(), + required_args, + input_schema: tool.function.parameters.clone(), + output_fields, + output_schema: tool.function.output_parameters.clone(), + primary_array_path, + is_curated, + } + }) + .collect(); + + if let Ok(mut cache) = LIVE_CATALOG_CACHE.get_or_init(Default::default).lock() { + cache.insert(key, contracts.clone()); + } + Some(contracts) +} + +/// Walks an output JSON Schema breadth-first for the first `type: "array"` +/// property, returning its dotted path relative to the schema's own root +/// (e.g. `"data.messages"` for a Gmail-list-shaped `{data: {messages: +/// [...]}}` schema, or `"messages"` for a flatter `{messages: [...]}` +/// schema). `None` when `schema` is absent or no array property is found at +/// any depth. +/// +/// Breadth-first (not depth-first): when a schema nests more than one array +/// property, the SHALLOWEST one wins, since that is virtually always the one +/// a `split_out` node should fan out over. +pub(crate) fn compute_primary_array_path(schema: Option<&Value>) -> Option { + let root = schema?; + let mut queue: std::collections::VecDeque<(String, &Value)> = std::collections::VecDeque::new(); + queue.push_back((String::new(), root)); + + while let Some((path, node)) = queue.pop_front() { + let Some(props) = node.get("properties").and_then(Value::as_object) else { + continue; + }; + // Check every property at THIS level for an array before descending + // to the next level — guarantees the shallowest match wins. + for (key, prop_schema) in props { + if prop_schema.get("type").and_then(Value::as_str) == Some("array") { + let prop_path = if path.is_empty() { + key.clone() + } else { + format!("{path}.{key}") + }; + return Some(prop_path); + } + } + for (key, prop_schema) in props { + let prop_path = if path.is_empty() { + key.clone() + } else { + format!("{path}.{key}") + }; + queue.push_back((prop_path, prop_schema)); + } + } + None } /// Best-effort lookup of a Composio action's **required** top-level parameter -/// names, from the toolkit's tool schemas (`parameters.required`). +/// names — a thin projection over [`fetch_live_toolkit_catalog`]'s +/// [`ToolContract`]s (this used to run its own independent +/// `REQUIRED_ARGS_CACHE`-backed fetch; existing callers — the required-arg +/// preflight, `graph_wiring_warnings` — keep this exact signature). /// /// Returns `None` when the schema is unavailable — unknown toolkit, client -/// construction failure, or a failed/empty listing — so callers can skip the -/// preflight rather than block execution on a catalog hiccup. Results are -/// cached per toolkit for the life of the process. +/// construction failure, a failed/empty listing, or the slug isn't present +/// in the toolkit's live catalog — so callers can skip the preflight rather +/// than block execution on a catalog hiccup. pub(crate) async fn composio_required_args(config: &Config, slug: &str) -> Option> { let toolkit = crate::openhuman::memory_sync::composio::providers::toolkit_from_slug(slug)?; - let slug_key = slug.to_ascii_uppercase(); - - if let Some(by_slug) = REQUIRED_ARGS_CACHE - .get_or_init(Default::default) - .lock() - .ok()? - .get(&toolkit) - { - return by_slug.get(&slug_key).cloned(); - } - - tracing::debug!(target: "flows", %toolkit, %slug, "[flows] preflight: fetching tool schemas for toolkit"); - let resp = crate::openhuman::composio::ops::composio_list_tools( - config, - Some(vec![toolkit.clone()]), - None, - ) - .await - .map_err(|e| { - tracing::debug!(target: "flows", %toolkit, error = %e, "[flows] preflight: schema fetch failed — skipping check"); - e - }) - .ok()?; - - let mut by_slug = std::collections::HashMap::new(); - for tool in &resp.value.tools { - let required: Vec = tool - .function - .parameters - .as_ref() - .and_then(|p| p.get("required")) - .and_then(Value::as_array) - .map(|arr| { - arr.iter() - .filter_map(|v| v.as_str().map(str::to_string)) - .collect() - }) - .unwrap_or_default(); - by_slug.insert(tool.function.name.to_ascii_uppercase(), required); - } - let found = by_slug.get(&slug_key).cloned(); - if let Ok(mut cache) = REQUIRED_ARGS_CACHE.get_or_init(Default::default).lock() { - cache.insert(toolkit, by_slug); - } - found -} - -/// Process-level cache for [`composio_response_fields`]: toolkit → (uppercase -/// action slug → top-level output/response field names). A sibling of -/// [`REQUIRED_ARGS_CACHE`] — same keying and one-fetch-per-toolkit-per-process -/// lifetime — but kept as its own `OnceLock` so seeding one cache in tests -/// never leaks into the other. -static RESPONSE_FIELDS_CACHE: std::sync::OnceLock< - std::sync::Mutex< - std::collections::HashMap>>, - >, -> = std::sync::OnceLock::new(); - -/// Seeds the response-fields cache for a toolkit — test hook so -/// `search_tool_catalog`'s grounding can be exercised without a live Composio -/// backend. Mirrors [`seed_required_args_cache`]. -#[cfg(test)] -pub(crate) fn seed_response_fields_cache( - toolkit: &str, - entries: std::collections::HashMap>, -) { - RESPONSE_FIELDS_CACHE - .get_or_init(Default::default) - .lock() - .expect("response-fields cache poisoned") - .insert(toolkit.to_string(), entries); + let contracts = fetch_live_toolkit_catalog(config, &toolkit).await?; + contracts + .iter() + .find(|c| c.slug.eq_ignore_ascii_case(slug)) + .map(|c| c.required_args.clone()) } /// Best-effort lookup of a Composio action's **response/output** top-level -/// field names — the output-side analogue of [`composio_required_args`]'s -/// input-side lookup, so `search_tool_catalog` can ground a downstream -/// binding (`=nodes..item.json.`) in a real field name instead of -/// guessing one. -/// -/// Source: Composio v3 `/tools` publishes an `output_parameters` JSON Schema -/// per action alongside `input_parameters` — documented as "Schema -/// definition of return values from the tool" -/// (). -/// `direct_list_tools` (Composio Direct mode) threads that schema through as -/// [`crate::openhuman::composio::types::ComposioToolFunction::output_parameters`]. -/// The backend-proxied path forwards whatever its own -/// `/agent-integrations/composio/tools` response carries under the same -/// field — opaque to this crate, so it may legitimately be absent there. +/// field names — the output-side analogue of [`composio_required_args`], +/// now a thin projection over [`fetch_live_toolkit_catalog`]'s +/// [`ToolContract`]s (replaces the standalone `RESPONSE_FIELDS_CACHE`-backed +/// fetch; `search_tool_catalog`'s grounding keeps this exact signature). /// /// Returns `None` when no output schema is known for the slug — unknown -/// toolkit, client construction failure, a failed/empty listing, or an -/// action whose listing doesn't publish `output_parameters` — so callers -/// degrade to "output shape unknown" (e.g. suggest a dry-run) rather than -/// blocking or guessing. `Some(vec![])` means the schema was found but names -/// no top-level properties. Cached per toolkit for the life of the process. +/// toolkit, client construction failure, a failed/empty listing, the slug +/// isn't in the live catalog, or a real action whose listing doesn't +/// publish `output_parameters` — so callers degrade to "output shape +/// unknown" rather than blocking or guessing. `Some(vec![])` means the +/// schema was found but names no top-level properties. pub(crate) async fn composio_response_fields(config: &Config, slug: &str) -> Option> { let toolkit = crate::openhuman::memory_sync::composio::providers::toolkit_from_slug(slug)?; - let slug_key = slug.to_ascii_uppercase(); - - if let Some(by_slug) = RESPONSE_FIELDS_CACHE - .get_or_init(Default::default) - .lock() - .ok()? - .get(&toolkit) - { - return by_slug.get(&slug_key).cloned(); - } - - tracing::debug!(target: "flows", %toolkit, %slug, "[flows] catalog: fetching output schemas for toolkit"); - let resp = crate::openhuman::composio::ops::composio_list_tools( - config, - Some(vec![toolkit.clone()]), - None, - ) - .await - .map_err(|e| { - tracing::debug!(target: "flows", %toolkit, error = %e, "[flows] catalog: output-schema fetch failed — skipping"); - e - }) - .ok()?; - - let mut by_slug = std::collections::HashMap::new(); - for tool in &resp.value.tools { - // Only cache an entry when the listing actually published an output - // schema — an absent `output_parameters` must stay "unknown" (no - // entry, so lookups fall through to `None`) rather than collapsing - // into `Some(vec![])`, which would mean "schema present, no fields". - if let Some(schema) = tool.function.output_parameters.as_ref() { - let fields = response_fields_from_schema(Some(schema)); - by_slug.insert(tool.function.name.to_ascii_uppercase(), fields); - } - } - let found = by_slug.get(&slug_key).cloned(); - if let Ok(mut cache) = RESPONSE_FIELDS_CACHE.get_or_init(Default::default).lock() { - cache.insert(toolkit, by_slug); - } - found + let contracts = fetch_live_toolkit_catalog(config, &toolkit).await?; + let contract = contracts + .iter() + .find(|c| c.slug.eq_ignore_ascii_case(slug))?; + contract.output_schema.as_ref()?; + Some(contract.output_fields.clone()) } /// Extracts top-level field names from a Composio `output_parameters` JSON @@ -2750,22 +2926,32 @@ mod tests { use crate::openhuman::memory_sync::composio::providers::{ catalog_for_toolkit, get_provider, }; + let config = Config::default(); // Precondition: `flowstestkit` is genuinely uncatalogued, so the decision // flows through the connected-set path (not the static curated path). assert!(catalog_for_toolkit("flowstestkit").is_none()); assert!(get_provider("flowstestkit").is_none()); // No connected set at all → fail-closed reject. - assert!(!flow_tool_allowed("FLOWSTESTKIT_DO_THING", None).await); + assert!(!flow_tool_allowed(&config, "FLOWSTESTKIT_DO_THING", None).await); // Connected set present but does not include this toolkit → reject. - assert!(!flow_tool_allowed("FLOWSTESTKIT_DO_THING", Some(&["gmail".to_string()])).await); + assert!( + !flow_tool_allowed( + &config, + "FLOWSTESTKIT_DO_THING", + Some(&["gmail".to_string()]) + ) + .await + ); // A blank slug is always rejected. - assert!(!flow_tool_allowed("", Some(&["flowstestkit".to_string()])).await); + assert!(!flow_tool_allowed(&config, "", Some(&["flowstestkit".to_string()])).await); } /// A real Composio toolkit OpenHuman ships no static catalog for now PASSES - /// once the user has an ACTIVE connection for it (the TODO(0.3) fix) — the - /// exact same slug that rejects above. + /// once the user has an ACTIVE connection for it (the TODO(0.3) fix) AND + /// the slug is a genuine action in its LIVE catalog (systemic tool-contract + /// fix) — seeded here so the test never touches a live Composio backend. + /// The exact same slug rejects above without a connection. #[tokio::test] async fn connected_uncatalogued_toolkit_now_passes() { use crate::openhuman::memory_sync::composio::providers::{ @@ -2774,12 +2960,77 @@ mod tests { assert!(catalog_for_toolkit("flowstestkit").is_none()); assert!(get_provider("flowstestkit").is_none()); + let config = Config::default(); + seed_live_catalog_cache( + "flowstestkit", + vec![ToolContract { + slug: "FLOWSTESTKIT_DO_THING".to_string(), + toolkit: "flowstestkit".to_string(), + description: None, + required_args: Vec::new(), + input_schema: None, + output_fields: Vec::new(), + output_schema: None, + primary_array_path: None, + is_curated: false, + }], + ); + assert!( - flow_tool_allowed("FLOWSTESTKIT_DO_THING", Some(&["flowstestkit".to_string()])).await + flow_tool_allowed( + &config, + "FLOWSTESTKIT_DO_THING", + Some(&["flowstestkit".to_string()]) + ) + .await ); // Case-insensitive match on the toolkit slug. assert!( - flow_tool_allowed("FLOWSTESTKIT_DO_THING", Some(&["FlowsTestKit".to_string()])).await + flow_tool_allowed( + &config, + "FLOWSTESTKIT_DO_THING", + Some(&["FlowsTestKit".to_string()]) + ) + .await + ); + } + + /// A CONNECTED but uncatalogued toolkit still rejects a slug that shares + /// its prefix but isn't a genuine action in the LIVE catalog — the + /// systemic tool-contract fix's tightening: connection alone is no longer + /// sufficient, the slug itself must be real. + #[tokio::test] + async fn connected_uncatalogued_toolkit_rejects_a_hallucinated_slug() { + use crate::openhuman::memory_sync::composio::providers::{ + catalog_for_toolkit, get_provider, + }; + assert!(catalog_for_toolkit("flowstestkit").is_none()); + assert!(get_provider("flowstestkit").is_none()); + + let config = Config::default(); + seed_live_catalog_cache( + "flowstestkit", + vec![ToolContract { + slug: "FLOWSTESTKIT_DO_THING".to_string(), + toolkit: "flowstestkit".to_string(), + description: None, + required_args: Vec::new(), + input_schema: None, + output_fields: Vec::new(), + output_schema: None, + primary_array_path: None, + is_curated: false, + }], + ); + + assert!( + !flow_tool_allowed( + &config, + "FLOWSTESTKIT_MADE_UP_ACTION", + Some(&["flowstestkit".to_string()]) + ) + .await, + "a hallucinated slug for a connected-but-uncurated toolkit must still reject" ); } @@ -3219,4 +3470,204 @@ mod tests { assert!(response_fields_from_schema(Some(&json!("not an object"))).is_empty()); assert!(response_fields_from_schema(Some(&json!({}))).is_empty()); } + + // ── compute_primary_array_path ────────────────────────────────────────── + + #[test] + fn compute_primary_array_path_finds_a_top_level_array_property() { + let schema = json!({ + "type": "object", + "properties": { "items": { "type": "array" }, "count": { "type": "integer" } } + }); + assert_eq!( + compute_primary_array_path(Some(&schema)), + Some("items".to_string()) + ); + } + + #[test] + fn compute_primary_array_path_finds_a_nested_array_property() { + // Gmail-shaped: the array lives two levels down, under `data.messages`. + let schema = json!({ + "type": "object", + "properties": { + "data": { + "type": "object", + "properties": { + "messages": { "type": "array" }, + "nextPageToken": { "type": "string" } + } + } + } + }); + assert_eq!( + compute_primary_array_path(Some(&schema)), + Some("data.messages".to_string()) + ); + } + + #[test] + fn compute_primary_array_path_prefers_the_shallowest_array() { + // A top-level array (`items`) must win over a deeper one + // (`data.nested`) even though `data` is declared first. + let schema = json!({ + "type": "object", + "properties": { + "data": { + "type": "object", + "properties": { "nested": { "type": "array" } } + }, + "items": { "type": "array" } + } + }); + assert_eq!( + compute_primary_array_path(Some(&schema)), + Some("items".to_string()) + ); + } + + #[test] + fn compute_primary_array_path_none_when_absent_or_no_array_property() { + assert_eq!(compute_primary_array_path(None), None); + assert_eq!( + compute_primary_array_path(Some(&json!({ "type": "object" }))), + None + ); + assert_eq!( + compute_primary_array_path(Some( + &json!({ "type": "object", "properties": { "id": { "type": "string" } } }) + )), + None + ); + } + + // ── fetch_live_toolkit_catalog / composio_required_args / + // composio_response_fields delegation ───────────────────────────────── + + fn contract( + slug: &str, + toolkit: &str, + required: &[&str], + output_fields: &[&str], + ) -> ToolContract { + let output_schema = if output_fields.is_empty() { + None + } else { + Some(json!({ + "type": "object", + "properties": output_fields + .iter() + .map(|f| (f.to_string(), json!({ "type": "string" }))) + .collect::>() + })) + }; + ToolContract { + slug: slug.to_string(), + toolkit: toolkit.to_string(), + description: None, + required_args: required.iter().map(|s| s.to_string()).collect(), + input_schema: None, + output_fields: output_fields.iter().map(|s| s.to_string()).collect(), + output_schema, + primary_array_path: None, + is_curated: false, + } + } + + #[tokio::test] + async fn fetch_live_toolkit_catalog_returns_the_seeded_cache_without_a_network_call() { + let config = Config::default(); + seed_live_catalog_cache( + "flowscatalogkit", + vec![contract( + "FLOWSCATALOGKIT_DO_THING", + "flowscatalogkit", + &["to"], + &["id", "threadId"], + )], + ); + + let catalog = fetch_live_toolkit_catalog(&config, "flowscatalogkit") + .await + .expect("seeded catalog must be returned without a network call"); + assert_eq!(catalog.len(), 1); + assert_eq!(catalog[0].slug, "FLOWSCATALOGKIT_DO_THING"); + + // Case/whitespace-insensitive on the toolkit key. + let same = fetch_live_toolkit_catalog(&config, " FlowsCatalogKit ") + .await + .expect("cache lookup is case/whitespace-insensitive"); + assert_eq!(same.len(), 1); + } + + #[tokio::test] + async fn composio_required_args_and_response_fields_delegate_to_the_live_catalog() { + let config = Config::default(); + seed_live_catalog_cache( + "flowsreqkit", + vec![contract( + "FLOWSREQKIT_SEND", + "flowsreqkit", + &["to", "body"], + &["id", "threadId"], + )], + ); + + assert_eq!( + composio_required_args(&config, "FLOWSREQKIT_SEND").await, + Some(vec!["to".to_string(), "body".to_string()]) + ); + assert_eq!( + composio_response_fields(&config, "FLOWSREQKIT_SEND").await, + Some(vec!["id".to_string(), "threadId".to_string()]) + ); + + // An unknown slug within a known/seeded toolkit yields None (not a + // panic, not an empty-vec false positive). + assert_eq!( + composio_required_args(&config, "FLOWSREQKIT_UNKNOWN_ACTION").await, + None + ); + assert_eq!( + composio_response_fields(&config, "FLOWSREQKIT_UNKNOWN_ACTION").await, + None + ); + } + + #[tokio::test] + async fn composio_response_fields_distinguishes_unknown_schema_from_empty_fields() { + let config = Config::default(); + + // Schema KNOWN but empty (`properties: {}`) → `Some(vec![])`. + seed_live_catalog_cache( + "flowsschemaempty", + vec![{ + let mut c = contract("FLOWSSCHEMAEMPTY_ACTION", "flowsschemaempty", &[], &[]); + c.output_schema = Some(json!({ "type": "object", "properties": {} })); + c + }], + ); + assert_eq!( + composio_response_fields(&config, "FLOWSSCHEMAEMPTY_ACTION").await, + Some(Vec::new()), + "schema known but empty must be Some(vec![]), not None" + ); + + // Schema UNKNOWN (`output_schema: None`, the degrade-gracefully case) + // → `None`, even though the slug itself is found in the catalog. + seed_live_catalog_cache( + "flowsschemaunknown", + vec![contract( + "FLOWSSCHEMAUNKNOWN_ACTION", + "flowsschemaunknown", + &[], + &[], + )], + ); + assert_eq!( + composio_response_fields(&config, "FLOWSSCHEMAUNKNOWN_ACTION").await, + None, + "an action with no published output schema must be None, not Some(vec![])" + ); + } } diff --git a/src/openhuman/tinyflows/tests.rs b/src/openhuman/tinyflows/tests.rs index 72ce1a92e..285aae2e1 100644 --- a/src/openhuman/tinyflows/tests.rs +++ b/src/openhuman/tinyflows/tests.rs @@ -397,22 +397,51 @@ fn missing_required_args_flags_absent_and_null() { assert!(super::caps::missing_required_args(&required, &full).is_empty()); } +/// Minimal seeded [`ToolContract`](super::caps::ToolContract) for the tests +/// below — only `required_args` matters for the preflight, so every other +/// field is left at its "unknown" default. +fn seeded_required_args_contract( + slug: &str, + toolkit: &str, + required: &[&str], +) -> super::caps::ToolContract { + super::caps::ToolContract { + slug: slug.to_string(), + toolkit: toolkit.to_string(), + description: None, + required_args: required.iter().map(|s| s.to_string()).collect(), + input_schema: None, + output_fields: Vec::new(), + output_schema: None, + primary_array_path: None, + is_curated: false, + } +} + #[tokio::test] async fn preflight_fails_before_dispatch_naming_the_missing_field() { let tmp = TempDir::new().unwrap(); let config = test_config(&tmp); - // Seed the schema cache so no live Composio backend is needed. - let mut entries = std::collections::HashMap::new(); - entries.insert( - "GMAIL_SEND_EMAIL".to_string(), - vec!["to".to_string(), "subject".to_string(), "body".to_string()], + // Seed the schema cache so no live Composio backend is needed. Uses its + // own toolkit/slug (never `"gmail"`) — the process-global + // `LIVE_CATALOG_CACHE` is shared with every `#[tokio::test]` in this + // file, and `preflight_invoker_gates_the_mock_tool_path` below seeds a + // DIFFERENT required-args list under the same slug; under parallel + // execution one test could overwrite the other's cache entry between + // seed and assert, making both flaky. + super::caps::seed_live_catalog_cache( + "preflightfailtest", + vec![seeded_required_args_contract( + "PREFLIGHTFAILTEST_SEND_EMAIL", + "preflightfailtest", + &["to", "subject", "body"], + )], ); - super::caps::seed_required_args_cache("gmail", entries); // `to` resolved to null (the classic mis-wired agent → tool_call case). let err = super::caps::preflight_composio_args( &config, - "GMAIL_SEND_EMAIL", + "PREFLIGHTFAILTEST_SEND_EMAIL", &json!({ "to": null, "subject": "hi", "body": "text" }), ) .await @@ -431,7 +460,7 @@ async fn preflight_fails_before_dispatch_naming_the_missing_field() { // Fully-wired args pass. super::caps::preflight_composio_args( &config, - "GMAIL_SEND_EMAIL", + "PREFLIGHTFAILTEST_SEND_EMAIL", &json!({ "to": "a@b.com", "subject": "hi", "body": "text" }), ) .await @@ -455,9 +484,17 @@ async fn preflight_invoker_gates_the_mock_tool_path() { let tmp = TempDir::new().unwrap(); let config = test_config(&tmp); - let mut entries = std::collections::HashMap::new(); - entries.insert("GMAIL_SEND_EMAIL".to_string(), vec!["to".to_string()]); - super::caps::seed_required_args_cache("gmail", entries); + // Own toolkit/slug (never `"gmail"`) — see the comment in + // `preflight_fails_before_dispatch_naming_the_missing_field` above for + // why sharing the `"gmail"` cache key across parallel tests is flaky. + super::caps::seed_live_catalog_cache( + "preflightgatetest", + vec![seeded_required_args_contract( + "PREFLIGHTGATETEST_SEND_EMAIL", + "preflightgatetest", + &["to"], + )], + ); let mock = tinyflows::caps::mock::mock_capabilities(); let invoker = super::caps::PreflightToolInvoker { @@ -468,17 +505,21 @@ async fn preflight_invoker_gates_the_mock_tool_path() { // Unwired required arg: fails with the named field even though the inner // mock would echo anything. let err = invoker - .invoke("GMAIL_SEND_EMAIL", json!({ "to": null }), None) + .invoke("PREFLIGHTGATETEST_SEND_EMAIL", json!({ "to": null }), None) .await .expect_err("dry-run preflight must catch the unwired arg"); assert!(err.to_string().contains("`to`")); // Wired arg: delegates to the mock echo. let ok = invoker - .invoke("GMAIL_SEND_EMAIL", json!({ "to": "a@b.com" }), None) + .invoke( + "PREFLIGHTGATETEST_SEND_EMAIL", + json!({ "to": "a@b.com" }), + None, + ) .await .expect("wired arg passes through to the mock"); - assert_eq!(ok["tool"], "GMAIL_SEND_EMAIL"); + assert_eq!(ok["tool"], "PREFLIGHTGATETEST_SEND_EMAIL"); // Native `oh:` slugs bypass the Composio preflight (no Composio schema). // The mock echoes them unchecked. diff --git a/src/openhuman/tools/ops.rs b/src/openhuman/tools/ops.rs index 9063f944b..c573eb83e 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -286,6 +286,11 @@ pub fn all_tools_with_runtime( Box::new(GetFlowRunTool::new(config.clone())), Box::new(ListFlowConnectionsTool::new(config.clone())), Box::new(SearchToolCatalogTool::new(config.clone())), + // Full live contract (schemas, real required_args/output_fields, + // primary_array_path) for one action slug found via + // search_tool_catalog — the grounding step before WIRING a node's + // args/downstream bindings (systemic tool-contract fix, Part 1). + Box::new(GetToolContractTool::new(config.clone())), // Ground an `agent` node's `agent_ref` in real registered agent-kind ids // (researcher / code_executor / …) — the agent analogue of // search_tool_catalog. Read-only.