feat(flows): search_tool_catalog surfaces real tool output field names (#4591)

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",

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();

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<serde_json::Value>,
+    /// 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" —
+    /// <https://docs.composio.dev/reference/api-reference/tools/getTools>).
+    /// 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<serde_json::Value>,
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -1090,6 +1099,9 @@ pub struct ComposioToolSchemaV3 {
     pub description: Option<String>,
     pub toolkit_slug: Option<String>,
     pub input_parameters: Option<serde_json::Value>,
+    /// See [`ComposioV3Tool::output_parameters`] — Composio v3's schema for
+    /// the action's return value, when published.
+    pub output_parameters: Option<serde_json::Value>,
 }
 
 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,
         }
     }
 }

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"));

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,
                 },
             },
         ],

src/openhuman/composio/types.rs

@@ -251,9 +251,20 @@ pub struct ComposioToolFunction {
     /// Human-readable description shown to the model.
     #[serde(default)]
     pub description: Option<String>,
-    /// JSON schema for the tool parameters.
+    /// JSON schema for the tool's INPUT parameters.
     #[serde(default)]
     pub parameters: Option<serde_json::Value>,
+    /// 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"
+    /// (<https://docs.composio.dev/reference/api-reference/tools/getTools>) —
+    /// 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<serde_json::Value>,
 }
 
 /// Response body of `GET /agent-integrations/composio/tools`.

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",

src/openhuman/flows/agents/workflow_builder/builder_prompt.rs

@@ -48,7 +48,7 @@ pub struct BuilderRequest {
     #[serde(default)]
     pub graph: Option<Value>,
     /// 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<String>,
     /// 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());

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.<tool_call_id>.item.json.<field>` 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:<tool_name>` (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`.

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.<id>.item.json.<field>`) can be wired to a
+/// field that actually exists instead of a guessed one.
+pub struct SearchToolCatalogTool {
+    config: Arc<Config>,
 }
 
-impl Default for SearchToolCatalogTool {
-    fn default() -> Self {
-        Self::new()
+impl SearchToolCatalogTool {
+    pub fn new(config: Arc<Config>) -> 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.<id>.item.json.<field>`. 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<String> = 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<String, Option<Vec<String>>> =
+            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::<String>::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(),

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]

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 {

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<ToolResult> {
-        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}"
                 )))

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<String, std::collections::HashMap<String, Vec<String>>>,
+    >,
+> = 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<String, Vec<String>>,
+) {
+    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.<id>.item.json.<field>`) 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"
+/// (<https://docs.composio.dev/reference/api-reference/tools/getTools>).
+/// `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<Vec<String>> {
+    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<String> {
+    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<String> =
+        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<String> {
     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());
+    }
 }

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.