diff --git a/src/openhuman/agent_registry/agents/loader.rs b/src/openhuman/agent_registry/agents/loader.rs index ba7d7a013..3733acaac 100644 --- a/src/openhuman/agent_registry/agents/loader.rs +++ b/src/openhuman/agent_registry/agents/loader.rs @@ -1031,7 +1031,7 @@ mod tests { // + confirmed test-run + save-onto-existing belt — no flow creation // (flows_create/set_enabled), no shell, no file writes, no channel // sends, and no composio_execute. It can list toolkits/connections, - // raise the inline connect card, `run_workflow` a flow the user already + // raise the inline connect card, `run_flow` a flow the user already // SAVED to test it (a real run the prompt gates behind user // confirmation), and `save_workflow` a built graph onto a flow the host // ALREADY created (the prompt bar's instant-create path) — but it can @@ -1065,7 +1065,7 @@ mod tests { "search_tool_catalog", "list_agent_profiles", "dry_run_workflow", - "run_workflow", + "run_flow", "composio_list_toolkits", "composio_list_connections", "composio_connect", diff --git a/src/openhuman/composio/client.rs b/src/openhuman/composio/client.rs index 563ebf3c3..e4b72faaa 100644 --- a/src/openhuman/composio/client.rs +++ b/src/openhuman/composio/client.rs @@ -1090,6 +1090,7 @@ pub(super) async fn direct_list_tools( name: item.slug, description: item.description, parameters: item.input_parameters, + output_parameters: item.output_parameters, }, }) .collect(); diff --git a/src/openhuman/composio/tools/direct.rs b/src/openhuman/composio/tools/direct.rs index 885ce09f3..1e35a0f2a 100644 --- a/src/openhuman/composio/tools/direct.rs +++ b/src/openhuman/composio/tools/direct.rs @@ -1030,6 +1030,15 @@ struct ComposioV3Tool { /// the same model-callable schema backend mode surfaces. #[serde(default, alias = "parameters")] input_parameters: Option, + /// JSON schema for the tool's OUTPUT/return value, per Composio v3 + /// `/tools`'s `output_parameters` field ("Schema definition of return + /// values from the tool" — + /// ). + /// Re-emitted as `ComposioToolFunction::output_parameters` so callers + /// can ground a downstream binding in the tool's real output field + /// names instead of guessing them. + #[serde(default)] + output_parameters: Option, } #[derive(Debug, Clone, Deserialize)] @@ -1090,6 +1099,9 @@ pub struct ComposioToolSchemaV3 { pub description: Option, pub toolkit_slug: Option, pub input_parameters: Option, + /// See [`ComposioV3Tool::output_parameters`] — Composio v3's schema for + /// the action's return value, when published. + pub output_parameters: Option, } impl ComposioToolSchemaV3 { @@ -1109,6 +1121,7 @@ impl ComposioToolSchemaV3 { description: item.description.or(item.name), toolkit_slug, input_parameters: item.input_parameters, + output_parameters: item.output_parameters, } } } diff --git a/src/openhuman/composio/tools/direct_tests.rs b/src/openhuman/composio/tools/direct_tests.rs index 51d7d3c20..05779019f 100644 --- a/src/openhuman/composio/tools/direct_tests.rs +++ b/src/openhuman/composio/tools/direct_tests.rs @@ -726,6 +726,7 @@ fn map_v3_tools_uses_name_when_slug_missing() { app_name: Some("myapp".into()), toolkit: None, input_parameters: None, + output_parameters: None, }]; let actions = map_v3_tools_to_actions(items); assert_eq!(actions.len(), 1); @@ -742,6 +743,7 @@ fn map_v3_tools_skips_items_without_slug_or_name() { app_name: None, toolkit: None, input_parameters: None, + output_parameters: None, }]; let actions = map_v3_tools_to_actions(items); assert!( @@ -762,6 +764,7 @@ fn map_v3_tools_prefers_toolkit_slug_over_app_name() { name: None, }), input_parameters: None, + output_parameters: None, }]; let actions = map_v3_tools_to_actions(items); assert_eq!(actions[0].app_name.as_deref(), Some("preferred-app")); diff --git a/src/openhuman/composio/tools_tests.rs b/src/openhuman/composio/tools_tests.rs index 22e89b67f..ea0a5ea88 100644 --- a/src/openhuman/composio/tools_tests.rs +++ b/src/openhuman/composio/tools_tests.rs @@ -641,6 +641,7 @@ fn render_tools_markdown_groups_by_toolkit_and_drops_schemas() { }, "required": ["to", "subject", "body"], })), + output_parameters: None, }, }, ComposioToolSchema { @@ -653,6 +654,7 @@ fn render_tools_markdown_groups_by_toolkit_and_drops_schemas() { "properties": { "title": {} }, "required": ["title"], })), + output_parameters: None, }, }, ], @@ -707,6 +709,7 @@ fn retain_connected_tools_drops_unconnected_toolkits_case_insensitively() { name: "GMAIL_SEND_EMAIL".into(), description: None, parameters: None, + output_parameters: None, }, }, ComposioToolSchema { @@ -715,6 +718,7 @@ fn retain_connected_tools_drops_unconnected_toolkits_case_insensitively() { name: "NOTION_CREATE_PAGE".into(), description: None, parameters: None, + output_parameters: None, }, }, ComposioToolSchema { @@ -723,6 +727,7 @@ fn retain_connected_tools_drops_unconnected_toolkits_case_insensitively() { name: "GMAIL_LIST_THREADS".into(), description: None, parameters: None, + output_parameters: None, }, }, ], diff --git a/src/openhuman/composio/types.rs b/src/openhuman/composio/types.rs index 21249f217..eb181fdeb 100644 --- a/src/openhuman/composio/types.rs +++ b/src/openhuman/composio/types.rs @@ -251,9 +251,20 @@ pub struct ComposioToolFunction { /// Human-readable description shown to the model. #[serde(default)] pub description: Option, - /// JSON schema for the tool parameters. + /// JSON schema for the tool's INPUT parameters. #[serde(default)] pub parameters: Option, + /// JSON schema describing the tool's OUTPUT/return-value shape, when the + /// upstream listing publishes one. Composio's v3 `/tools` endpoint calls + /// this `output_parameters` — documented as "Schema definition of return + /// values from the tool" + /// () — + /// alongside `input_parameters`. `None` means "unknown" (not "empty"): + /// the backend-proxied `/agent-integrations/composio/tools` path is + /// opaque to this crate and may not forward it, and not every Composio + /// action publishes an output schema. + #[serde(default)] + pub output_parameters: Option, } /// Response body of `GET /agent-integrations/composio/tools`. diff --git a/src/openhuman/flows/agents/workflow_builder/agent.toml b/src/openhuman/flows/agents/workflow_builder/agent.toml index 5a0f3342c..e8da3b39b 100644 --- a/src/openhuman/flows/agents/workflow_builder/agent.toml +++ b/src/openhuman/flows/agents/workflow_builder/agent.toml @@ -8,7 +8,7 @@ iteration_policy = "extended" max_result_chars = 12000 # No sandbox filter needed: the belt is propose/read/dry-run plus two # explicitly-bounded writes (save_workflow onto an existing flow; -# run_workflow of a saved flow behind a confirm-first prompt rule), and +# run_flow of a saved flow behind a confirm-first prompt rule), and # dry_run_workflow runs against tinyflows MOCK capabilities (no real effects). sandbox_mode = "none" omit_identity = true @@ -33,9 +33,11 @@ hint = "reasoning" # create or enable a flow, or perform a real integration action directly. # Composio access is limited to LISTING toolkits/connections and raising the # inline CONNECT card (an approval-gated OAuth hand-off). -# `run_workflow` executes a flow the user has ALREADY saved to test it — a real +# `run_flow` executes a flow the user has ALREADY saved to test it — a real # run, but the prompt requires asking the user to confirm first, and the flow's -# own approval gate still pauses outbound-action nodes. +# own approval gate still pauses outbound-action nodes. (Named `run_flow`, not +# `run_workflow` — that name belongs to the unrelated legacy skills-workflow +# runner tool, which is registered in the same default tool registry.) named = [ "propose_workflow", "revise_workflow", @@ -47,7 +49,7 @@ named = [ "search_tool_catalog", "list_agent_profiles", "dry_run_workflow", - "run_workflow", + "run_flow", "composio_list_toolkits", "composio_list_connections", "composio_connect", diff --git a/src/openhuman/flows/agents/workflow_builder/builder_prompt.rs b/src/openhuman/flows/agents/workflow_builder/builder_prompt.rs index 6d31152b6..d7d0443d7 100644 --- a/src/openhuman/flows/agents/workflow_builder/builder_prompt.rs +++ b/src/openhuman/flows/agents/workflow_builder/builder_prompt.rs @@ -48,7 +48,7 @@ pub struct BuilderRequest { #[serde(default)] pub graph: Option, /// The saved flow's id (required for `build`; optional elsewhere so the - /// agent may `run_workflow` it to test after confirming). + /// agent may `run_flow` it to test after confirming). #[serde(default)] pub flow_id: Option, /// The failed run id (== thread id) for `repair`, so the agent can @@ -93,13 +93,13 @@ const DIRECTIVE_PROPOSE: &str = const DIRECTIVE_REVISE: &str = "Revise this tinyflows automation and return the revised proposal. Do not save \ unless I explicitly ask you to (when I do, use save_workflow on the saved flow id), and never enable or \ - disable anything. You may run_workflow the SAVED flow to test it, but ONLY if I ask and only after you \ + disable anything. You may run_flow the SAVED flow to test it, but ONLY if I ask and only after you \ confirm with me first."; const DIRECTIVE_BUILD_AND_SAVE: &str = "Build this tinyflows automation END-TO-END. The flow already exists \ (created blank just now) — design the graph, verify it with dry_run_workflow, return the workflow \ proposal, then SAVE it onto the flow id below with save_workflow. Do not enable or disable anything, and \ - do not run_workflow a real test unless I explicitly confirm first. Tell me what you saved when you are done."; + do not run_flow a real test unless I explicitly confirm first. Tell me what you saved when you are done."; /// Serialize a graph compactly for injection as agent context. fn serialize_graph(graph: &Value) -> String { @@ -131,7 +131,7 @@ pub fn render_prompt(req: &BuilderRequest) -> String { lines.push(String::new()); lines.push(format!( "This workflow is saved with flow id `{flow_id}` — if I ask you to run/test it, you \ - may run_workflow that id, but confirm with me first." + may run_flow that id, but confirm with me first." )); } lines.push(String::new()); diff --git a/src/openhuman/flows/agents/workflow_builder/prompt.md b/src/openhuman/flows/agents/workflow_builder/prompt.md index 49b5610b3..b8d6616a6 100644 --- a/src/openhuman/flows/agents/workflow_builder/prompt.md +++ b/src/openhuman/flows/agents/workflow_builder/prompt.md @@ -43,19 +43,19 @@ Never `save_workflow` onto a flow the user did NOT ask you to build/update — editing some other saved flow requires their explicit ask naming it. It cannot create flows, and it never changes `enabled` or the approval gate. -## Testing a saved flow: `run_workflow` (ask first!) +## Testing a saved flow: `run_flow` (ask first!) -Once the user has **saved** a flow, you can `run_workflow { flow_id }` to test it +Once the user has **saved** a flow, you can `run_flow { flow_id }` to test it end-to-end. Unlike `dry_run_workflow`, this is a **real run** — real effects can fire (the flow's own approval gate still pauses outbound-action nodes, but treat it as real). Rules: -1. **Only a saved flow.** `run_workflow` needs a `flow_id`; if the graph isn't +1. **Only a saved flow.** `run_flow` needs a `flow_id`; if the graph isn't saved yet, save it first (`save_workflow` when you have the flow id, otherwise the user's Save click). You can't run a draft — use `dry_run_workflow` for a draft wiring check. 2. **ALWAYS ask for confirmation and wait for an explicit "yes"** before calling - `run_workflow`. Say what it will do ("This will run the flow for real and may + `run_flow`. Say what it will do ("This will run the flow for real and may send/act on live data — run it now?") and only proceed once they agree. Never run a workflow unprompted or as a surprise side effect of another request. 3. After a run, read the result (status + any nodes paused for approval) and @@ -73,6 +73,9 @@ it as real). Rules: - `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). - `list_flows` / `get_flow` → reuse or clone an existing flow instead of duplicating one. - **Missing the integration the workflow needs?** See "Connecting @@ -156,6 +159,14 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. — 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. + - **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 — + don't ship a guessed field name. - **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 0386af050..f3e936129 100644 --- a/src/openhuman/flows/builder_tools.rs +++ b/src/openhuman/flows/builder_tools.rs @@ -501,17 +501,19 @@ impl Tool for ListFlowConnectionsTool { /// `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). -pub struct SearchToolCatalogTool; - -impl SearchToolCatalogTool { - pub fn new() -> Self { - Self - } +/// +/// 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. +pub struct SearchToolCatalogTool { + config: Arc, } -impl Default for SearchToolCatalogTool { - fn default() -> Self { - Self::new() +impl SearchToolCatalogTool { + pub fn new(config: Arc) -> Self { + Self { config } } } @@ -580,8 +582,14 @@ impl Tool for SearchToolCatalogTool { "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 } entries. ALWAYS ground a \ - tool_call node's `slug` in a real result here — do not invent slugs." + 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." } fn parameters_schema(&self) -> Value { @@ -622,7 +630,57 @@ impl Tool for SearchToolCatalogTool { toolkit = toolkit.unwrap_or("(any)"), "[flows] search_tool_catalog: searching curated Composio catalog (read-only)" ); - let results = search_curated_catalog(&query, toolkit, MAX_CATALOG_RESULTS); + 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"), + ); + } + } + } Ok(ToolResult::success(serde_json::to_string_pretty(&json!({ "query": query, "count": results.len(), diff --git a/src/openhuman/flows/builder_tools_tests.rs b/src/openhuman/flows/builder_tools_tests.rs index 7814b9a49..7f5b027c1 100644 --- a/src/openhuman/flows/builder_tools_tests.rs +++ b/src/openhuman/flows/builder_tools_tests.rs @@ -175,7 +175,8 @@ fn search_curated_catalog_all_terms_must_match() { #[tokio::test] async fn search_tool_catalog_tool_is_read_only_and_grounds() { - let tool = SearchToolCatalogTool::new(); + let tmp = TempDir::new().unwrap(); + let tool = SearchToolCatalogTool::new(test_config(&tmp)); assert_eq!(tool.name(), "search_tool_catalog"); assert_eq!(tool.permission_level(), PermissionLevel::None); assert!(!tool.external_effect()); @@ -191,12 +192,79 @@ async fn search_tool_catalog_tool_is_read_only_and_grounds() { #[tokio::test] async fn search_tool_catalog_missing_query_is_error() { - let tool = SearchToolCatalogTool::new(); + let tmp = TempDir::new().unwrap(); + let tool = SearchToolCatalogTool::new(test_config(&tmp)); let result = tool.execute(json!({})).await.unwrap(); assert!(result.is_error); assert!(result.output().contains("Missing 'query'")); } +#[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. + 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" })) + .await + .unwrap(); + assert!(!result.is_error, "{}", result.output()); + let parsed: Value = serde_json::from_str(&result.output()).unwrap(); + let results = parsed["results"].as_array().unwrap(); + 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"] + .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()); +} + +#[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); + + let tool = SearchToolCatalogTool::new(test_config(&tmp)); + let result = tool + .execute(json!({ "query": "send", "toolkit": "slack" })) + .await + .unwrap(); + assert!(!result.is_error, "{}", result.output()); + let parsed: Value = serde_json::from_str(&result.output()).unwrap(); + 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" + ); + } +} + // ── dry_run_workflow ───────────────────────────────────────────────────────── #[test] diff --git a/src/openhuman/flows/schemas.rs b/src/openhuman/flows/schemas.rs index dc5428d55..9151858ec 100644 --- a/src/openhuman/flows/schemas.rs +++ b/src/openhuman/flows/schemas.rs @@ -779,7 +779,7 @@ pub fn schemas(function: &str) -> ControllerSchema { name: "flow_id", ty: TypeSchema::Option(Box::new(TypeSchema::String)), comment: "Saved flow id — required for `build` (save target); optional \ - elsewhere (lets the agent run_workflow it to test, with confirmation).", + elsewhere (lets the agent run_flow it to test, with confirmation).", required: false, }, FieldSchema { diff --git a/src/openhuman/flows/tools.rs b/src/openhuman/flows/tools.rs index 0a89470e6..0a0ebf960 100644 --- a/src/openhuman/flows/tools.rs +++ b/src/openhuman/flows/tools.rs @@ -245,7 +245,15 @@ impl RunFlowTool { #[async_trait] impl Tool for RunFlowTool { fn name(&self) -> &str { - "run_workflow" + // NOTE: deliberately `run_flow`, not `run_workflow` — the latter + // name is already taken by the unrelated legacy skills-workflow + // runner (`crate::openhuman::agent::tools::run_workflow`, + // `RUN_WORKFLOW_TOOL_NAME`), which is registered in the same + // default tool registry (`tools::ops::all_tools_with_runtime`). + // Both tools were previously named identically, which + // `all_tools_default_registry_has_no_duplicate_tool_names` caught + // as a duplicate-tool-name registry bug. + "run_flow" } fn description(&self) -> &str { @@ -284,21 +292,22 @@ impl Tool for RunFlowTool { } async fn execute(&self, args: Value) -> anyhow::Result { - let flow_id = - match args.get("flow_id").and_then(Value::as_str).map(str::trim) { - Some(id) if !id.is_empty() => id.to_string(), - _ => return Ok(ToolResult::error( - "Missing 'flow_id' — run_workflow only works on a SAVED flow. Ask the user \ + let flow_id = match args.get("flow_id").and_then(Value::as_str).map(str::trim) { + Some(id) if !id.is_empty() => id.to_string(), + _ => { + return Ok(ToolResult::error( + "Missing 'flow_id' — run_flow only works on a SAVED flow. Ask the user \ to Save the workflow first, then run it by id." .to_string(), - )), - }; + )) + } + }; let input = args.get("input").cloned().unwrap_or_else(|| json!({})); tracing::info!( target: "flows", %flow_id, - "[flows] run_workflow: agent-initiated test run starting" + "[flows] run_flow: agent-initiated test run starting" ); match crate::openhuman::flows::ops::flows_run( @@ -315,7 +324,7 @@ impl Tool for RunFlowTool { "result": outcome.value, }))?)), Err(e) => { - tracing::debug!(target: "flows", %flow_id, error = %e, "[flows] run_workflow: failed"); + tracing::debug!(target: "flows", %flow_id, error = %e, "[flows] run_flow: failed"); Ok(ToolResult::error(format!( "Could not run flow '{flow_id}': {e}" ))) diff --git a/src/openhuman/tinyflows/caps.rs b/src/openhuman/tinyflows/caps.rs index f7f10167b..da86ab504 100644 --- a/src/openhuman/tinyflows/caps.rs +++ b/src/openhuman/tinyflows/caps.rs @@ -1294,6 +1294,132 @@ pub(crate) async fn composio_required_args(config: &Config, slug: &str) -> Optio 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); +} + +/// 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. +/// +/// 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. +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 +} + +/// Extracts top-level field names from a Composio `output_parameters` JSON +/// Schema value. Composio shapes this as a standard object schema — +/// `{"type": "object", "properties": {...}}` — same convention as +/// `input_parameters`, so this reads `.properties`'s keys when present. Falls +/// back to the schema's own top-level keys (minus common JSON-Schema +/// keywords) for a looser/legacy shape. Empty when the schema is +/// absent/unrecognized — never fails. +fn response_fields_from_schema(schema: Option<&Value>) -> Vec { + const SCHEMA_KEYWORDS: &[&str] = &[ + "type", + "required", + "additionalProperties", + "$schema", + "description", + "title", + "examples", + ]; + + let Some(obj) = schema.and_then(Value::as_object) else { + return Vec::new(); + }; + let mut fields: Vec = + if let Some(props) = obj.get("properties").and_then(Value::as_object) { + props.keys().cloned().collect() + } else { + obj.keys() + .filter(|k| !SCHEMA_KEYWORDS.contains(&k.as_str())) + .cloned() + .collect() + }; + fields.sort(); + fields +} + /// Returns the names in `required` that are absent or `null` in `args`. pub(crate) fn missing_required_args(required: &[String], args: &Value) -> Vec { required @@ -2792,4 +2918,63 @@ mod tests { other => panic!("expected a capability error, got: {other:?}"), } } + + // ── response_fields_from_schema ───────────────────────────────────────── + // Direct unit tests for the pure schema-extraction step inside + // `composio_response_fields`'s live-fetch loop — cheaper and more + // targeted than exercising the whole `composio_list_tools` round trip, + // and covers the schema shapes that loop actually has to handle. + + #[test] + fn response_fields_from_schema_reads_standard_properties_object() { + let schema = json!({ + "type": "object", + "properties": { "id": {"type": "string"}, "threadId": {"type": "string"} } + }); + assert_eq!( + response_fields_from_schema(Some(&schema)), + vec!["id".to_string(), "threadId".to_string()] + ); + } + + #[test] + fn response_fields_from_schema_reads_nested_data_error_wrapper_as_top_level_keys() { + // A `{data, error}` envelope has no special unwrapping — the function + // documents (and this test locks in) that it reports the schema's own + // top-level property names, not the fields nested inside `data`. + let schema = json!({ + "type": "object", + "properties": { + "data": {"type": "object", "properties": {"id": {"type": "string"}}}, + "error": {"type": "string"} + } + }); + assert_eq!( + response_fields_from_schema(Some(&schema)), + vec!["data".to_string(), "error".to_string()] + ); + } + + #[test] + fn response_fields_from_schema_falls_back_to_top_level_keys_minus_schema_keywords() { + // Legacy/loose shape with no `properties` wrapper: falls back to the + // schema object's own keys, filtering out JSON-Schema keywords. + let schema = json!({ + "type": "object", + "description": "legacy shape", + "id": {"type": "string"}, + "threadId": {"type": "string"} + }); + assert_eq!( + response_fields_from_schema(Some(&schema)), + vec!["id".to_string(), "threadId".to_string()] + ); + } + + #[test] + fn response_fields_from_schema_empty_for_none_or_non_object() { + assert!(response_fields_from_schema(None).is_empty()); + assert!(response_fields_from_schema(Some(&json!("not an object"))).is_empty()); + assert!(response_fields_from_schema(Some(&json!({}))).is_empty()); + } } diff --git a/src/openhuman/tools/ops.rs b/src/openhuman/tools/ops.rs index 5e659442b..9063f944b 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -285,7 +285,7 @@ pub fn all_tools_with_runtime( Box::new(GetFlowTool::new(config.clone())), Box::new(GetFlowRunTool::new(config.clone())), Box::new(ListFlowConnectionsTool::new(config.clone())), - Box::new(SearchToolCatalogTool::new()), + Box::new(SearchToolCatalogTool::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.