From 0663750f04b1476cef7ff837a5070f6a2e8a456e Mon Sep 17 00:00:00 2001 From: Cyrus Gray <144336577+graycyrus@users.noreply.github.com> Date: Tue, 7 Jul 2026 19:42:55 +0530 Subject: [PATCH] fix(flows): account for Composio's data-wrapper in tool output paths (B1) (#4645) --- .../flows/agents/workflow_builder/prompt.md | 26 +++- src/openhuman/flows/builder_tools.rs | 16 ++- src/openhuman/flows/ops.rs | 132 ++++++++++++++---- src/openhuman/flows/ops_tests.rs | 112 ++++++++++++++- src/openhuman/tinyflows/caps.rs | 118 +++++++++++++++- 5 files changed, 354 insertions(+), 50 deletions(-) diff --git a/src/openhuman/flows/agents/workflow_builder/prompt.md b/src/openhuman/flows/agents/workflow_builder/prompt.md index 186f60e9d..dafa23544 100644 --- a/src/openhuman/flows/agents/workflow_builder/prompt.md +++ b/src/openhuman/flows/agents/workflow_builder/prompt.md @@ -220,14 +220,18 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. - **Wiring a DOWNSTREAM node off THIS tool's output?** Don't guess the field name (e.g. assuming `GMAIL_FETCH_EMAILS` returns `.messages`) — `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. + output field names. **A Composio tool_call's result is wrapped in + `data`** (`ComposioExecuteResponse`), one level DEEPER than the engine's + own `{json,text,raw}` envelope — so bind + `=nodes..item.json.data.` (not `.item.json.`) + to one of those `output_fields`. 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. + `"path": "json.data.messages"` — as the downstream `split_out.path`. + `primary_array_path` already includes the `data.` segment above, so + just prefix `json.` — don't guess 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 @@ -285,7 +289,15 @@ Use expressions to thread data between steps (a `transform`'s `set`, an kinds is that envelope, NOT the structured value itself: - Structured fields live under **`.json`** — `"=nodes..item.json."` - (jq: `"=.nodes[\"\"].items[0].json."`). + (jq: `"=.nodes[\"\"].items[0].json."`) — **except a Composio + `tool_call`**, whose real output nests one level DEEPER, under `data`: + `"=nodes..item.json.data."`. That's Composio's own execute- + response wrapper (`{data, successful, error, costUsd, …}`), stacked + underneath the engine's `{json,text,raw}` envelope — `agent` and + `http_request` nodes carry no such wrapper and keep the plain + `.item.json.` form. A native `oh:`-prefixed tool_call also has no + `data` wrapper (it isn't a Composio call) — this only applies to a + `tool_call` whose `slug` is a real Composio action. - Prose lives under **`.text`** — `"=nodes..item.text"`. - `code`, `transform`, `split_out`, `merge`, `output_parser`, `sub_workflow`, and `trigger` nodes do **NOT** envelope — their output is addressed directly, diff --git a/src/openhuman/flows/builder_tools.rs b/src/openhuman/flows/builder_tools.rs index c6fa619d2..4c1ec8210 100644 --- a/src/openhuman/flows/builder_tools.rs +++ b/src/openhuman/flows/builder_tools.rs @@ -734,11 +734,17 @@ impl Tool for GetToolContractTool { 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." + `=nodes..item.json.data.` binding — note the `data.` segment: a Composio \ + tool_call's real runtime output wraps its payload in `data` \ + (`ComposioExecuteResponse`), so `output_fields` names fields INSIDE that wrapper, not \ + top-level envelope keys — never guess a field name, and never drop the `data.` \ + segment (`.item.json.` with no `data.` resolves null even when `` is a \ + real output field). Use `primary_array_path` (prefixed with `json.`, e.g. \ + \"json.data.messages\" — the `data.` segment is already baked into the value) 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 { diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index 8c977492a..78fe323b9 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -217,14 +217,21 @@ pub(crate) async fn graph_wiring_warnings(config: &Config, graph: &WorkflowGraph } /// 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. +/// `=nodes..item.json.data.` 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`. Also warns (a distinct +/// message) when the binding is missing the `data.` segment entirely — a +/// Composio `tool_call`'s real runtime output always wraps its payload in +/// `data` (`ComposioExecuteResponse`; see +/// [`crate::openhuman::tinyflows::caps::ToolContract::output_fields`]'s doc), +/// so `=nodes..item.json.` (no `data.`) is GUARANTEED to resolve +/// `null` even when `` names a real output field — that used to be +/// silently accepted here (B1: the exact bug that produces a hollow run). +/// 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 @@ -233,6 +240,16 @@ pub(crate) async fn graph_wiring_warnings(config: &Config, graph: &WorkflowGraph /// 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. +/// +/// Also skipped for a binding that addresses the whole payload +/// (`=nodes..item.json.data`, e.g. as an agent `input_context`) or one +/// of `ComposioExecuteResponse`'s OTHER top-level envelope fields — +/// `successful`, `error`, `costUsd`, `markdownFormatted` — which live +/// alongside `data`, not inside it. `OpenHumanTools::invoke` serializes the +/// whole `ComposioExecuteResponse` verbatim, so these ARE real +/// `.item.json.` fields with no `data.` prefix; flagging them as +/// "missing the `data.` segment" would rewire an already-correct binding to +/// a nonsense path (e.g. suggesting `.item.json.data.successful`). 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; @@ -240,7 +257,7 @@ async fn graph_output_field_warnings(config: &Config, graph: &WorkflowGraph) -> 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 { + let Some((ref_id, has_json, field_path)) = parse_node_binding(&expr) else { continue; }; if !has_json { @@ -270,11 +287,51 @@ async fn graph_output_field_warnings(config: &Config, graph: &WorkflowGraph) -> else { continue; }; - // Output schema unknown — nothing real to check `field` against. + // Output schema unknown — nothing real to check `field_path` against. if contract.output_schema.is_none() { continue; } - if !contract.output_fields.iter().any(|f| f == &field) { + + // Whole-payload access (`.item.json.data`, e.g. an agent's + // `input_context`) or one of `ComposioExecuteResponse`'s OTHER + // top-level envelope fields — these live alongside `data`, not + // inside it, and are real fields regardless of this action's + // `output_fields` (see this fn's doc). Not a "missing `data.`" + // mistake. + const COMPOSIO_ENVELOPE_METADATA_FIELDS: &[&str] = + &["successful", "error", "costUsd", "markdownFormatted"]; + if field_path == "data" + || COMPOSIO_ENVELOPE_METADATA_FIELDS + .contains(&field_path.split('.').next().unwrap_or(&field_path)) + { + continue; + } + + // A real Composio tool_call's payload is always nested one level + // under `data` (see this fn's doc) — a binding missing that + // segment is wrong regardless of whether the rest of the path + // happens to name a real field. + let Some(field) = field_path.strip_prefix("data.") else { + tracing::warn!( + target: "flows", + node = %node.id, + %location, + ref_node = %ref_id, + ref_slug, + %field_path, + "[flows] wiring check: downstream binding is missing the Composio `data.` wrapper segment" + ); + warnings.push(format!( + "Node '{}': binding `{location}` (`{expr}`) reads `.item.json.{field_path}` off \ + tool_call `{ref_id}` (`{ref_slug}`), but a Composio tool_call's real output \ + wraps its payload in `data` — this resolves null at runtime. Bind via \ + `=nodes.{ref_id}.item.json.data.{field_path}` instead.", + node.id + )); + continue; + }; + let field = field.split('.').next().unwrap_or(field); + if !contract.output_fields.iter().any(|f| f == field) { tracing::warn!( target: "flows", node = %node.id, @@ -302,12 +359,15 @@ async fn graph_output_field_warnings(config: &Config, graph: &WorkflowGraph) -> /// 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. +/// [`crate::openhuman::tinyflows::caps::compute_composio_array_path`] — this +/// already bakes in the `data.` segment Composio's execute-response wrapper +/// adds, so `expected` below comes out `"json.data.<…>"` for a real action +/// with no extra handling needed here), 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 @@ -443,10 +503,18 @@ fn collect_expressions(value: &Value) -> Vec<(String, String)> { } /// Matches the dotted-path form of a node-output binding — -/// `=nodes..item[.json].` — returning `(ref_id, has_json, -/// field)`. `has_json` is `true` when the expression dereferenced the -/// `{json,text,raw}` envelope wrapper (`.item.json.`) rather than the -/// item directly (`.item.`). +/// `=nodes..item[.json].` — returning `(ref_id, has_json, +/// field_path)`. `has_json` is `true` when the expression dereferenced the +/// `{json,text,raw}` envelope wrapper (`.item.json.`) rather than +/// the item directly (`.item.`). +/// +/// `field_path` captures the FULL remaining dotted path, not just its first +/// segment — e.g. `"data.messages"` for `.item.json.data.messages`. This +/// matters for a Composio `tool_call` ref, whose real output additionally +/// wraps the field in `data` (see [`crate::openhuman::tinyflows::caps::ToolContract::output_fields`]'s +/// doc): callers that need to check field membership against a schema with +/// no such wrapper (e.g. an `agent` node's `output_parser.schema`) should +/// compare against just `field_path`'s first segment. /// /// Only the dotted-path form is recognized here — the equivalent jq form /// (e.g. `=.nodes["ref"].items[0].field`) is an arbitrary jq program, not a @@ -458,7 +526,7 @@ fn parse_node_binding(expr: &str) -> Option<(String, bool, String)> { static RE: std::sync::OnceLock = std::sync::OnceLock::new(); RE.get_or_init(|| { regex::Regex::new( - r"^=nodes\.([A-Za-z_][A-Za-z0-9_]*)\.item(?:\.(json))?\.([A-Za-z_][A-Za-z0-9_]*)", + r"^=nodes\.([A-Za-z_][A-Za-z0-9_]*)\.item(?:\.(json))?\.([A-Za-z_][A-Za-z0-9_.]*)", ) .expect("static regex is valid") }) @@ -466,8 +534,11 @@ fn parse_node_binding(expr: &str) -> Option<(String, bool, String)> { let caps = node_binding_regex().captures(expr)?; let ref_id = caps.get(1)?.as_str().to_string(); let has_json = caps.get(2).is_some(); - let field = caps.get(3)?.as_str().to_string(); - Some((ref_id, has_json, field)) + let field_path = caps.get(3)?.as_str().trim_end_matches('.').to_string(); + if field_path.is_empty() { + return None; + } + Some((ref_id, has_json, field_path)) } /// Human-readable label for a [`NodeKind`], for @@ -650,7 +721,7 @@ pub(crate) fn validate_binding_resolvability(graph: &WorkflowGraph) -> Vec Vec Vec Vec` + /// (the engine's own `{json,text,raw}` envelope, THEN Composio's + /// `data` wrapper), never the bare `.item.json.` an agent/ + /// `http_request` output would use. pub output_fields: Vec, /// The action's full output JSON Schema, when Composio publishes one. /// `None` means "unknown to this listing", not "empty" — mirrors @@ -1397,9 +1408,13 @@ pub struct ToolContract { 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. + /// to the first array-typed property in the tool's real runtime output, + /// via [`compute_composio_array_path`]. Already accounts for Composio's + /// `data` wrapper (see [`Self::output_fields`]'s doc) — this is NOT the + /// bare [`compute_primary_array_path`] walk over [`Self::output_schema`], + /// which is relative to the unwrapped payload and would be missing the + /// leading `data.` segment. `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` / @@ -1552,7 +1567,7 @@ pub(crate) async fn fetch_live_toolkit_catalog( 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()); + compute_composio_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, @@ -1576,11 +1591,17 @@ pub(crate) async fn fetch_live_toolkit_catalog( /// 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: [...]}` +/// (e.g. `"data.messages"` for a schema that itself is shaped `{data: +/// {messages: [...]}}`, or `"messages"` for a flatter `{messages: [...]}` /// schema). `None` when `schema` is absent or no array property is found at /// any depth. /// +/// Pure schema walker — relative to `schema`'s own root, nothing else. A +/// real Composio `output_parameters` schema is normally shaped like the +/// flatter example (it describes the tool's payload, not the runtime +/// envelope around it) — [`compute_composio_array_path`] is the caller that +/// adjusts for that envelope; this function has no opinion on it. +/// /// 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. @@ -1617,6 +1638,32 @@ pub(crate) fn compute_primary_array_path(schema: Option<&Value>) -> Option, successful, error, costUsd, …}` — but the schema +/// Composio publishes as `output_parameters` (what [`compute_primary_array_path`] +/// walks) describes only ``, the content of that `data` field, not +/// the envelope around it. So the bare walk's result (e.g. `"messages"`) is +/// missing the `data.` segment a real `split_out.path`/downstream binding +/// needs (`"data.messages"`) — this wrapper adds it, UNCONDITIONALLY. +/// +/// There is no escape hatch for a payload schema that itself happens to +/// declare a top-level `data` property (e.g. a provider whose real payload +/// shape is `{data: {messages: [...]}}`, unrelated to Composio's own +/// wrapper) — `output_parameters` describes the payload only, per the +/// invariant documented on [`ToolContract::output_fields`], so the real +/// runtime path in that case is `data.data.messages`, not `data.messages`. +/// Treating a payload-level `data` key as "this schema already models the +/// envelope" silently drops a real wrapper segment and points a downstream +/// binding / `split_out.path` at the wrong (non-existent) array. +pub(crate) fn compute_composio_array_path(schema: Option<&Value>) -> Option { + let path = compute_primary_array_path(schema)?; + Some(format!("data.{path}")) +} + /// Best-effort lookup of a Composio action's **required** top-level parameter /// names — a thin projection over [`fetch_live_toolkit_catalog`]'s /// [`ToolContract`]s (this used to run its own independent @@ -3541,6 +3588,65 @@ mod tests { ); } + // ── compute_composio_array_path (B1: the `data` wrapper prefix) ───────── + + #[test] + fn compute_composio_array_path_prefixes_data_for_an_unwrapped_payload_schema() { + // The real shape: Composio's `output_parameters` for GMAIL_FETCH_EMAILS + // describes the payload directly — no `data` key in the schema — but + // the tool_call's real runtime output nests that payload one level + // deeper under `data` (`ComposioExecuteResponse`). The array path must + // account for that even though the schema itself never mentions `data`. + let schema = json!({ + "type": "object", + "properties": { + "messages": { "type": "array" }, + "nextPageToken": { "type": "string" } + } + }); + assert_eq!( + compute_composio_array_path(Some(&schema)), + Some("data.messages".to_string()) + ); + } + + #[test] + fn compute_composio_array_path_still_prefixes_data_when_the_payload_schema_itself_has_a_data_key( + ) { + // A payload whose own real shape happens to have a top-level `data` + // key (unrelated to Composio's wrapper — e.g. a provider that + // itself returns `{data: {messages: [...]}}`) must NOT be mistaken + // for "this schema already models the envelope". `output_parameters` + // always describes the payload only (see `ToolContract::output_fields`'s + // doc) — the real runtime path still needs the wrapper's `data.` + // prefix stacked on top, landing on `data.data.messages`, not + // `data.messages`. + let schema = json!({ + "type": "object", + "properties": { + "data": { + "type": "object", + "properties": { "messages": { "type": "array" } } + } + } + }); + assert_eq!( + compute_composio_array_path(Some(&schema)), + Some("data.data.messages".to_string()) + ); + } + + #[test] + fn compute_composio_array_path_none_when_the_bare_walk_finds_nothing() { + assert_eq!(compute_composio_array_path(None), None); + assert_eq!( + compute_composio_array_path(Some( + &json!({ "type": "object", "properties": { "id": { "type": "string" } } }) + )), + None + ); + } + // ── fetch_live_toolkit_catalog / composio_required_args / // composio_response_fields delegation ─────────────────────────────────