From 9830c90d5a530187bf26fc2b7986209061657d22 Mon Sep 17 00:00:00 2001 From: Cyrus Gray <144336577+graycyrus@users.noreply.github.com> Date: Wed, 8 Jul 2026 18:21:15 +0530 Subject: [PATCH] fix(flows): real-output probe to derive the array path when the tool schema is absent (B16) (#4702) --- src/openhuman/agent_registry/agents/loader.rs | 11 +- .../flows/agents/workflow_builder/agent.toml | 15 +- .../flows/agents/workflow_builder/prompt.md | 11 + src/openhuman/flows/builder_tools.rs | 156 ++++- src/openhuman/flows/builder_tools_tests.rs | 101 ++- src/openhuman/flows/ops.rs | 172 +++-- src/openhuman/flows/ops_tests.rs | 237 ++++++- src/openhuman/tinyflows/caps.rs | 597 ++++++++++++++++++ src/openhuman/tools/ops.rs | 9 + 9 files changed, 1257 insertions(+), 52 deletions(-) diff --git a/src/openhuman/agent_registry/agents/loader.rs b/src/openhuman/agent_registry/agents/loader.rs index 4ec63225d..d6e0cb067 100644 --- a/src/openhuman/agent_registry/agents/loader.rs +++ b/src/openhuman/agent_registry/agents/loader.rs @@ -1053,9 +1053,13 @@ mod tests { // 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 - // never create/enable a flow or perform a raw integration action. This - // pins the invariant in the agent definition itself, not just the tool - // implementations. + // never create/enable a flow or perform an arbitrary raw integration + // action. One narrow, deliberate carve-out (B12): `get_tool_output_sample` + // DOES make a real Composio call, but only ever a Read-scope one + // (hard-refused otherwise, regardless of the user's scope preference) + // against an already-connected toolkit — see `builder_tools.rs`'s + // module doc. This pins the invariant in the agent definition itself, + // not just the tool implementations. let def = find("workflow_builder"); assert_eq!(def.agent_tier, AgentTier::Worker); assert_eq!(def.delegate_name.as_deref(), Some("build_workflow")); @@ -1082,6 +1086,7 @@ mod tests { "list_flow_connections", "search_tool_catalog", "get_tool_contract", + "get_tool_output_sample", "list_agent_profiles", "dry_run_workflow", "run_flow", diff --git a/src/openhuman/flows/agents/workflow_builder/agent.toml b/src/openhuman/flows/agents/workflow_builder/agent.toml index 2f0e31672..6b09db83c 100644 --- a/src/openhuman/flows/agents/workflow_builder/agent.toml +++ b/src/openhuman/flows/agents/workflow_builder/agent.toml @@ -30,14 +30,22 @@ hint = "reasoning" # a confirmed real test-run of a SAVED flow + save_workflow (persist a graph # onto an EXISTING flow only). NO shell, NO file writes, NO channel sends, NO # composio_execute, and NO flows_create/set_enabled — the agent can never -# 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). +# create or enable a flow, or perform an arbitrary real integration action +# directly. Composio access is limited to LISTING toolkits/connections, +# raising the inline CONNECT card (an approval-gated OAuth hand-off), and one +# narrow carve-out below. # `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. (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.) +# `get_tool_output_sample` (B12) IS a real Composio call — the one deliberate +# carve-out — but it is hard-gated to Read-scope actions only (regardless of +# the user's scope preference) against an already-connected toolkit, and +# exists solely to sample the real output shape of a source tool whose live +# listing publishes no output schema at all (verified for every GitHub +# action), so a downstream split_out.path is never guessed. See +# `builder_tools.rs`'s module doc. named = [ "propose_workflow", "revise_workflow", @@ -48,6 +56,7 @@ named = [ "list_flow_connections", "search_tool_catalog", "get_tool_contract", + "get_tool_output_sample", "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 169835dbe..4d5c5c715 100644 --- a/src/openhuman/flows/agents/workflow_builder/prompt.md +++ b/src/openhuman/flows/agents/workflow_builder/prompt.md @@ -256,6 +256,17 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. `"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. + **If `get_tool_contract` returns `primary_array_path: null` for a source + tool you plan to `split_out` (its live listing has no output schema at + all — this is genuinely true for every GitHub action, e.g. + `GITHUB_LIST_REPOSITORY_ISSUES`), do NOT default to `"json.data"`** — that + targets the WHOLE payload container (e.g. `{issues: [...]}` itself), so + the split yields exactly ONE item instead of one per real result. Instead + call `get_tool_output_sample { slug, args }` (the SAME `args` you're + wiring into the real node) to make one bounded, read-only, real call and + get the ACTUAL array path (e.g. `"data.issues"`, not `"data.items"`) — + it only works on an already-connected, Read-scope action, so if the + toolkit isn't connected yet, note that to the user instead of guessing. - **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 diff --git a/src/openhuman/flows/builder_tools.rs b/src/openhuman/flows/builder_tools.rs index 22c6ba861..906b0d457 100644 --- a/src/openhuman/flows/builder_tools.rs +++ b/src/openhuman/flows/builder_tools.rs @@ -13,6 +13,7 @@ //! | [`ListFlowConnectionsTool`] | `None` | read: connection refs (ids/names only) | //! | [`SearchToolCatalogTool`] | `None` | read: real Composio tool slugs (live catalog) | //! | [`GetToolContractTool`] | `None` | read: one action's FULL live contract | +//! | [`GetToolOutputSampleTool`] | `ReadOnly` | ONE bounded real Composio call (Read-scope only, connected toolkit only) | //! | [`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 | @@ -32,8 +33,18 @@ //! `composio_list_toolkits`, `composio_list_connections`, `composio_connect` //! (defined in `composio/tools.rs`) — so the builder can link an app the //! workflow needs before proposing. Those stay within the invariant: connect -//! is an approval-gated OAuth hand-off, and `composio_execute` (running a real -//! action) remains deliberately OUT of scope. +//! is an approval-gated OAuth hand-off, and `composio_execute` (running an +//! arbitrary real action, any scope) remains deliberately OUT of scope. +//! +//! **One narrow, deliberate carve-out (B12):** [`GetToolOutputSampleTool`] +//! (`get_tool_output_sample`) DOES perform a real Composio call — but only +//! ever a `Read`-scope one (hard-refused otherwise, regardless of the user's +//! per-toolkit scope preference), and only against a toolkit the user has +//! ALREADY connected. It exists because some actions' live listings publish +//! no output schema at all (verified for every GitHub action), leaving +//! `get_tool_contract` with no ground truth for a downstream `split_out.path` +//! — this makes exactly one bounded real read to observe the actual shape +//! instead. It can never send/create/update/delete anything. use std::sync::Arc; @@ -802,7 +813,20 @@ impl Tool for GetToolContractTool { }; match catalog.iter().find(|c| c.slug.eq_ignore_ascii_case(&slug)) { - Some(contract) => Ok(ToolResult::success(serde_json::to_string_pretty(contract)?)), + Some(contract) => { + // B12: a prior real-output probe (get_tool_output_sample) for + // this exact slug is ACTUAL observed data and always wins + // over the schema-derived hint — most relevant for an action + // whose live listing publishes no output schema at all (e.g. + // every GitHub action verified live as of this fix), where + // `contract.primary_array_path` would otherwise be + // permanently `None`. + let contract = + crate::openhuman::tinyflows::caps::apply_probe_override(contract.clone()); + 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." @@ -811,6 +835,132 @@ impl Tool for GetToolContractTool { } } +// ───────────────────────────────────────────────────────────────────────────── +// get_tool_output_sample — READ-ONLY real Composio call: the B12 output probe +// ───────────────────────────────────────────────────────────────────────────── + +/// `get_tool_output_sample`: make ONE bounded, READ-ONLY, REAL Composio call +/// for `slug` and derive its `primary_array_path`/`output_fields` from the +/// ACTUAL response, overriding `get_tool_contract`'s schema-derived hint for +/// this slug from then on (see +/// [`crate::openhuman::tinyflows::caps::apply_probe_override`]). +/// +/// **Exists because a schema-derived hint sometimes doesn't exist at all**: +/// Composio's live listing genuinely omits `output_parameters` for some +/// actions — verified live for every GitHub action, including the curated +/// `GITHUB_LIST_REPOSITORY_ISSUES` — leaving `get_tool_contract`'s +/// `primary_array_path` permanently `null`. Without ground truth the builder +/// has been observed guessing the whole-payload `"json.data"` as a +/// `split_out.path` (live flow "funny reminders v2": one item — the +/// `{issues:[...]}` container itself — instead of the real per-item list), +/// silently degrading a fan-out to a single item. +/// +/// **This is a deliberate, narrow carve-out of the workflow-builder agent's +/// "propose/read only, no composio_execute" invariant** (see this module's +/// top doc): unlike `composio_execute`, this tool can ONLY ever perform a +/// `Read`-scope action (gated by +/// [`crate::openhuman::tinyflows::caps::probe_tool_output_sample`]'s scope +/// check, which ignores the user's per-toolkit scope preference — a probe +/// must never perform a real mutation no matter what the user has toggled +/// on) against a toolkit the user has ALREADY connected. No message is sent, +/// no record created/updated/deleted, ever. +/// +/// Pass the SAME `args` you intend to wire into the real `tool_call` node — +/// this samples THAT call, not a generic fixture. Omit `args` (or pass `{}`) +/// for a zero-required-arg action. +pub struct GetToolOutputSampleTool { + config: Arc, +} + +impl GetToolOutputSampleTool { + pub fn new(config: Arc) -> Self { + Self { config } + } +} + +#[async_trait] +impl Tool for GetToolOutputSampleTool { + fn name(&self) -> &str { + "get_tool_output_sample" + } + + fn description(&self) -> &str { + "Make ONE bounded, READ-ONLY, REAL call to a Composio action and derive its real \ + `primary_array_path`/`output_fields` from the ACTUAL response — use this when \ + get_tool_contract returns `output_schema: null` / `primary_array_path: null` for a \ + source tool you plan to `split_out` (e.g. every GitHub action, verified live), so a \ + downstream split_out.path never fans out over the whole-payload container by mistake. \ + Only ever performs a Read action (refuses Write/Admin actions unconditionally, \ + regardless of the user's scope preference) against an ALREADY-CONNECTED toolkit — never \ + sends, creates, updates, or deletes anything. Pass the SAME args you intend to wire into \ + the real tool_call node — this samples THAT exact call. Call get_tool_contract again \ + afterward (or trust this tool's own `primary_array_path`/`output_fields`) to see the \ + override applied. Real actions only, not `oh:` or `=`-derived slugs." + } + + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "slug": { + "type": "string", + "description": "The exact Composio action slug, e.g. 'GITHUB_LIST_REPOSITORY_ISSUES'." + }, + "args": { + "type": "object", + "description": "Arguments for the real call — the SAME ones you intend to wire into the tool_call node (e.g. {\"owner\": \"acme\", \"repo\": \"widgets\"}). Omit for a zero-required-arg action." + } + }, + "required": ["slug"], + "additionalProperties": false + }) + } + + fn permission_level(&self) -> PermissionLevel { + PermissionLevel::ReadOnly + } + + 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 call_args = args.get("args").cloned().unwrap_or(json!({})); + + tracing::debug!( + target: "flows", + %slug, + "[flows] get_tool_output_sample: tool invoked" + ); + + match crate::openhuman::tinyflows::caps::probe_tool_output_sample( + &self.config, + &slug, + call_args, + ) + .await + { + Ok(sample) => { + let primary_array_path_for_split_out = sample + .primary_array_path + .as_ref() + .map(|p| format!("json.{p}")); + Ok(ToolResult::success(serde_json::to_string_pretty(&json!({ + "slug": slug, + "primary_array_path": sample.primary_array_path, + "split_out_path": primary_array_path_for_split_out, + "output_fields": sample.output_fields, + }))?)) + } + Err(e) => Ok(ToolResult::error(e)), + } + } +} + // ───────────────────────────────────────────────────────────────────────────── // list_agent_profiles — read-only: selectable agent kinds for an `agent` node // ───────────────────────────────────────────────────────────────────────────── diff --git a/src/openhuman/flows/builder_tools_tests.rs b/src/openhuman/flows/builder_tools_tests.rs index 675160771..bd4e6dcb0 100644 --- a/src/openhuman/flows/builder_tools_tests.rs +++ b/src/openhuman/flows/builder_tools_tests.rs @@ -190,7 +190,9 @@ async fn list_flow_connections_is_read_only() { // share a toolkit key (same discipline the pre-fix required-args/response- // fields caches already required). -use crate::openhuman::tinyflows::caps::{seed_live_catalog_cache, ToolContract}; +use crate::openhuman::tinyflows::caps::{ + seed_live_catalog_cache, seed_probe_cache, ProbedOutputSample, ToolContract, +}; fn seeded_gmail_send_contract() -> ToolContract { ToolContract { @@ -401,6 +403,103 @@ async fn get_tool_contract_rejects_a_hallucinated_slug() { assert!(result.output().contains("not a real action")); } +/// B12: a cached real-output probe overrides `get_tool_contract`'s +/// schema-derived `primary_array_path`/`output_fields` — most relevant for a +/// slug whose live listing (like every GitHub action, verified live) has NO +/// output schema at all, so the schema-derived fields would otherwise be +/// permanently empty/null. +#[tokio::test] +async fn get_tool_contract_applies_a_cached_probe_override() { + let contract = ToolContract { + slug: "PROBEOVERRIDETEST_LIST_REPOSITORY_ISSUES".to_string(), + toolkit: "probeoverridetest".to_string(), + description: None, + required_args: vec!["owner".to_string(), "repo".to_string()], + input_schema: None, + output_fields: vec![], + output_schema: None, + primary_array_path: None, + is_curated: true, + }; + seed_live_catalog_cache("probeoverridetest", vec![contract]); + seed_probe_cache( + "PROBEOVERRIDETEST_LIST_REPOSITORY_ISSUES", + ProbedOutputSample { + primary_array_path: Some("data.issues".to_string()), + output_fields: vec!["issues".to_string(), "total_count".to_string()], + sample: json!({ "data": { "issues": [], "total_count": 0 } }), + }, + ); + let tmp = TempDir::new().unwrap(); + let tool = GetToolContractTool::new(test_config(&tmp)); + let result = tool + .execute(json!({ "slug": "PROBEOVERRIDETEST_LIST_REPOSITORY_ISSUES" })) + .await + .unwrap(); + assert!(!result.is_error, "{}", result.output()); + let parsed: Value = serde_json::from_str(&result.output()).unwrap(); + assert_eq!(parsed["primary_array_path"], "data.issues"); + assert_eq!(parsed["output_fields"], json!(["issues", "total_count"])); + // The schema-derived field stays null — the probe overrides the HINT + // fields, it doesn't fabricate a schema that was never published. + assert!(parsed["output_schema"].is_null()); +} + +// ── get_tool_output_sample (B12: the real-output probe) ───────────────────── + +#[test] +fn get_tool_output_sample_is_read_only_permission_with_no_external_effect() { + let tmp = TempDir::new().unwrap(); + let tool = GetToolOutputSampleTool::new(test_config(&tmp)); + assert_eq!(tool.name(), "get_tool_output_sample"); + assert_eq!(tool.permission_level(), PermissionLevel::ReadOnly); + assert!(!tool.external_effect()); +} + +#[tokio::test] +async fn get_tool_output_sample_missing_slug_is_error() { + let tmp = TempDir::new().unwrap(); + let tool = GetToolOutputSampleTool::new(test_config(&tmp)); + let result = tool.execute(json!({})).await.unwrap(); + assert!(result.is_error); + assert!(result.output().contains("Missing 'slug'")); +} + +/// The scope gate runs BEFORE any client/network call, so a Write-scope +/// action is refused entirely offline — this must never depend on a live +/// Composio backend to prove the probe can't perform a real mutation. +#[tokio::test] +async fn get_tool_output_sample_refuses_a_write_scope_action() { + let tmp = TempDir::new().unwrap(); + let tool = GetToolOutputSampleTool::new(test_config(&tmp)); + let result = tool + .execute(json!({ "slug": "GMAIL_SEND_EMAIL" })) + .await + .unwrap(); + assert!(result.is_error); + assert!(result.output().contains("READ-only"), "{}", result.output()); +} + +/// The connected-toolkit gate runs before the real call too — in a test +/// environment with no backend session, `fetch_connected_integrations` +/// degrades to empty (best-effort, per its own doc), so a Read-scope action +/// against an unconnected toolkit is refused without ever reaching a client. +#[tokio::test] +async fn get_tool_output_sample_refuses_an_unconnected_toolkit() { + let tmp = TempDir::new().unwrap(); + let tool = GetToolOutputSampleTool::new(test_config(&tmp)); + let result = tool + .execute(json!({ "slug": "GITHUB_LIST_REPOSITORY_ISSUES" })) + .await + .unwrap(); + assert!(result.is_error); + assert!( + result.output().contains("not connected") || result.output().contains("no active"), + "{}", + result.output() + ); +} + // ── dry_run_workflow ───────────────────────────────────────────────────────── #[test] diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index 7c47e2af7..e4ab5c7fe 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -288,8 +288,15 @@ async fn graph_output_field_warnings(config: &Config, graph: &WorkflowGraph) -> else { continue; }; - // Output schema unknown — nothing real to check `field_path` against. - if contract.output_schema.is_none() { + // B12: a real-output probe (`get_tool_output_sample`) for this + // exact slug overrides the schema-derived `output_fields` — most + // relevant for an action whose live listing publishes no output + // schema at all (e.g. every GitHub action, verified live). + let contract = + crate::openhuman::tinyflows::caps::apply_probe_override(contract.clone()); + // Nothing real to check `field_path` against — schema unknown AND + // no probed output fields either. + if contract.output_schema.is_none() && contract.output_fields.is_empty() { continue; } @@ -357,25 +364,74 @@ async fn graph_output_field_warnings(config: &Config, graph: &WorkflowGraph) -> 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_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. +/// Given a Composio action's payload-only `output_schema` (see +/// [`crate::openhuman::tinyflows::caps::ToolContract::output_fields`]'s doc — +/// NEVER includes the runtime `data` envelope) and a `split_out.path` +/// addressed relative to the ENVELOPE (`json.`, e.g. +/// `"json.data"` or `"json.data.issues"`), resolves whether the path lands on +/// something that is DEFINITELY not an array. /// -/// 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). +/// `Some(true)` — non-array (an object or scalar): a `split_out` over this +/// path fans out over exactly ONE item, the classic "wrong array path" +/// signal [`graph_split_out_path_warnings`]'s generic enforcement flags. +/// `Some(false)` — array: the path is fine. `None` — the path can't be +/// resolved against the schema at all (an unpublished/unknown nested field, +/// or a path missing the `data.` segment entirely) — stay silent rather than +/// guess; that's a distinct failure mode from "resolves to a non-array". +fn schema_says_path_is_non_array(output_schema: &Value, configured_path: &str) -> Option { + let relative = configured_path + .strip_prefix("json.") + .unwrap_or(configured_path); + if relative == "data" { + // Whole-payload access (`json.data`) — non-array unless the payload's + // own root schema type is literally "array" (a bare-array response, + // e.g. a REST endpoint that returns `[...]` directly), in which case + // `json.data` legitimately IS the real list. + let ty = output_schema.get("type").and_then(Value::as_str)?; + return Some(ty != "array"); + } + let rest = relative.strip_prefix("data.").filter(|r| !r.is_empty())?; + let mut node = output_schema; + for seg in rest.split('.') { + node = node.get("properties")?.get(seg)?; + } + let ty = node.get("type").and_then(Value::as_str)?; + Some(ty != "array") +} + +/// Author-time WARN/suggest (systemic tool-contract fix, Part 2d, extended by +/// B12): a `split_out` node whose direct predecessor is a `tool_call` calling +/// a REAL Composio action, checked two ways: +/// +/// 1. **KNOWN `primary_array_path`** (see +/// [`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.<…>"` with no +/// extra handling needed here — and, via +/// [`crate::openhuman::tinyflows::caps::apply_probe_override`], a real +/// `get_tool_output_sample` probe for this slug overrides a schema that +/// never named an array at all): if the configured `config.path` doesn't match the +/// `json.` convention, suggest the real path. +/// 2. **UNKNOWN `primary_array_path`, but a KNOWN `output_schema`/probe that +/// proves the configured path is definitely NOT an array** (B12 +/// enforcement, "regardless" of whether a correct path can be suggested — +/// catches the class at build time even when nothing to suggest is +/// derivable): warn generically. This is exactly the live bug this fix +/// closes — `GITHUB_LIST_REPOSITORY_ISSUES` publishes no output schema at +/// all, so a builder without a probe guessed the whole-payload +/// `"json.data"`, silently fanning out over ONE item (the `{issues: +/// [...]}` container) instead of the real per-issue list. +/// +/// Both are advisory: a mismatched/non-array path degrades the fan-out (or +/// silently produces one item instead of many) rather than crashing. +/// +/// Skipped entirely when `split_out`'s predecessor isn't a `tool_call` at all +/// (no envelope/array-path convention applies), or when NEITHER a +/// `primary_array_path` NOR an `output_schema` is known (truly nothing to +/// check against). 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; + use crate::openhuman::tinyflows::caps::{apply_probe_override, fetch_live_toolkit_catalog}; let mut warnings = Vec::new(); for node in &graph.nodes { @@ -409,29 +465,63 @@ async fn graph_split_out_path_warnings(config: &Config, graph: &WorkflowGraph) - 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, - )); + // B12: a real-output probe overrides the schema-derived + // `primary_array_path` for this exact slug when one is cached. + let contract = apply_probe_override(contract.clone()); + + match contract.primary_array_path.as_deref() { + Some(primary) => { + 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, + )); + } + } + // No known array anywhere in this action's real output — the + // generic non-array enforcement is the only thing left that + // can catch a wrong path here (nothing to suggest, but a + // known-non-array hit is still a strong signal). + None => { + let Some(cp) = configured_path else { continue }; + let Some(schema) = contract.output_schema.as_ref() else { + continue; + }; + if schema_says_path_is_non_array(schema, cp) == Some(true) { + tracing::warn!( + target: "flows", + node = %node.id, + predecessor = %pred.id, + pred_slug, + configured_path = cp, + "[flows] wiring check: split_out.path resolves to a non-array — likely the wrong array path" + ); + warnings.push(format!( + "Node '{}': split_out.path is \"{cp}\" but tool_call `{}` (`{pred_slug}`)'s \ + known real output does not name an array at that path (or names no array \ + property at all) — this fans out over a single object instead of a real \ + list. If the action's real output nests the list under a named field (e.g. \ + `data.issues`), call get_tool_output_sample {{ slug: \"{pred_slug}\" }} to \ + sample the real response, then re-check with get_tool_contract.", + node.id, pred.id, + )); + } + } } } } diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index f69fcabba..9b1b93e37 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -1829,7 +1829,9 @@ fn binding_to_agent_with_matching_schema_is_accepted() { // 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}; +use crate::openhuman::tinyflows::caps::{ + seed_live_catalog_cache, seed_probe_cache, ProbedOutputSample, ToolContract, +}; fn seeded_slack_send_contract() -> ToolContract { ToolContract { @@ -2369,6 +2371,239 @@ async fn graph_wiring_warnings_suggests_the_real_split_out_path() { ); } +/// B12 enforcement: a `split_out.path` that resolves to a NON-array (an +/// object, here) against a KNOWN output schema is flagged even though the +/// action names no array anywhere (`primary_array_path` is `None`) — there +/// is nothing to *suggest*, but a definite non-array hit is still a strong +/// "wrong array path" signal worth catching at build time. +#[tokio::test] +async fn graph_wiring_warnings_flags_a_split_out_path_that_resolves_to_a_non_array() { + // seeded_slack_send_contract's output_schema names only scalar fields + // (ts/channel) — a real, known schema with no array in it anywhere. + let mut contract = seeded_slack_send_contract(); + contract.slug = "NONARRAYFANOUT_SEND_MESSAGE".to_string(); + contract.toolkit = "nonarrayfanout".to_string(); + seed_live_catalog_cache("nonarrayfanout", 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": "NONARRAYFANOUT_SEND_MESSAGE", + "args": { "channel": "#general", "text": "hi" } } }, + { "id": "split", "kind": "split_out", "name": "Split", + "config": { "path": "json.data" } } + ], + "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("split") && w.contains("does not name an array")), + "{warnings:?}" + ); +} + +/// The non-array enforcement stays SILENT when the action's output schema is +/// genuinely unknown (not just "known but arrayless") — nothing real to check +/// the path against, so no false positive. +#[tokio::test] +async fn graph_wiring_warnings_is_silent_on_split_out_when_schema_is_wholly_unknown() { + let contract = ToolContract { + slug: "UNKNOWNSCHEMA_DO_THING".to_string(), + toolkit: "unknownschema".to_string(), + description: None, + required_args: vec![], + input_schema: None, + output_fields: vec![], + output_schema: None, + primary_array_path: None, + is_curated: true, + }; + seed_live_catalog_cache("unknownschema", 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": "UNKNOWNSCHEMA_DO_THING", "args": {} } }, + { "id": "split", "kind": "split_out", "name": "Split", + "config": { "path": "json.data" } } + ], + "edges": [ + { "from_node": "t", "to_node": "post" }, + { "from_node": "post", "to_node": "split" } + ] + })); + assert!( + graph_wiring_warnings(&config, &g).await.is_empty(), + "{:?}", + graph_wiring_warnings(&config, &g).await + ); +} + +/// B12 end-to-end: the EXACT live bug shape (flow "funny reminders v2"). +/// `GITHUB_LIST_REPOSITORY_ISSUES`-equivalent contract has NO schema at all +/// (`output_schema: None`, `primary_array_path: None` — verified live for +/// every GitHub action), so before a probe the enforcement above has nothing +/// to check the configured `"json.data"` against and stays silent. Once +/// `get_tool_output_sample` has probed the slug (seeded here via +/// `seed_probe_cache`, standing in for a real bounded call), the cached +/// `primary_array_path` overrides the schema-derived (absent) hint and the +/// EXISTING mismatch-suggestion path fires with the real nested path. +#[tokio::test] +async fn graph_wiring_warnings_suggests_the_probed_split_out_path_when_schema_is_unknown() { + let contract = ToolContract { + slug: "GHPROBEFANOUT_LIST_REPOSITORY_ISSUES".to_string(), + toolkit: "ghprobefanout".to_string(), + description: None, + required_args: vec!["owner".to_string(), "repo".to_string()], + input_schema: None, + output_fields: vec![], + output_schema: None, + primary_array_path: None, + is_curated: true, + }; + seed_live_catalog_cache("ghprobefanout", vec![contract]); + seed_probe_cache( + "GHPROBEFANOUT_LIST_REPOSITORY_ISSUES", + ProbedOutputSample { + primary_array_path: Some("data.issues".to_string()), + output_fields: vec!["issues".to_string(), "total_count".to_string()], + sample: json!({ "data": { "issues": [], "total_count": 0 } }), + }, + ); + let config = Config::default(); + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "GHPROBEFANOUT_LIST_REPOSITORY_ISSUES", + "args": { "owner": "acme", "repo": "widgets" } } }, + // The exact wrong guess observed live: whole-payload access + // instead of the real nested `data.issues`. + { "id": "split", "kind": "split_out", "name": "Split", + "config": { "path": "json.data" } } + ], + "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.issues")), + "{warnings:?}" + ); + + // Fixed: once config.path matches the probed real path, the warning + // clears. + let fixed = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "GHPROBEFANOUT_LIST_REPOSITORY_ISSUES", + "args": { "owner": "acme", "repo": "widgets" } } }, + { "id": "split", "kind": "split_out", "name": "Split", + "config": { "path": "json.data.issues" } } + ], + "edges": [ + { "from_node": "t", "to_node": "post" }, + { "from_node": "post", "to_node": "split" } + ] + })); + assert!( + graph_wiring_warnings(&config, &fixed).await.is_empty(), + "{:?}", + graph_wiring_warnings(&config, &fixed).await + ); +} + +/// CodeRabbit (PR #4702 review): parity coverage for the probe-override path +/// in `graph_output_field_warnings` — mirrors +/// `graph_wiring_warnings_suggests_the_probed_split_out_path_when_schema_is_unknown` +/// above, but for a downstream FIELD binding rather than `split_out.path`. +/// With no schema at all (`output_schema: None`, `output_fields: []`), the +/// field-not-in-output_fields check would otherwise stay silent (nothing +/// real to check against) — once `get_tool_output_sample` has probed the +/// slug, the probed `output_fields` become the ground truth: a binding to a +/// probed-real field is silent, and a binding to a field NOT in the probed +/// set is flagged, exactly like the schema-known case already covers. +#[tokio::test] +async fn graph_wiring_warnings_uses_the_probed_output_fields_when_schema_is_unknown() { + let contract = ToolContract { + slug: "GHPROBEFIELDS_LIST_REPOSITORY_ISSUES".to_string(), + toolkit: "ghprobefields".to_string(), + description: None, + required_args: vec!["owner".to_string(), "repo".to_string()], + input_schema: None, + output_fields: vec![], + output_schema: None, + primary_array_path: None, + is_curated: true, + }; + seed_live_catalog_cache("ghprobefields", vec![contract]); + seed_probe_cache( + "GHPROBEFIELDS_LIST_REPOSITORY_ISSUES", + ProbedOutputSample { + primary_array_path: Some("data.issues".to_string()), + output_fields: vec!["issues".to_string(), "total_count".to_string()], + sample: json!({ "data": { "issues": [], "total_count": 0 } }), + }, + ); + let config = Config::default(); + + // A binding to a field the probe actually observed — silent. + let real_field = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "GHPROBEFIELDS_LIST_REPOSITORY_ISSUES", + "args": { "owner": "acme", "repo": "widgets" } } }, + { "id": "xform", "kind": "transform", "name": "Log", + "config": { "set": { "note": "=nodes.post.item.json.data.total_count" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "post" }, + { "from_node": "post", "to_node": "xform" } + ] + })); + assert!( + graph_wiring_warnings(&config, &real_field).await.is_empty(), + "a probed-real field must not warn: {:?}", + graph_wiring_warnings(&config, &real_field).await + ); + + // A binding to a field the probe did NOT observe — flagged, using the + // probed output_fields as ground truth even though the schema itself is + // unknown. + let fake_field = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "GHPROBEFIELDS_LIST_REPOSITORY_ISSUES", + "args": { "owner": "acme", "repo": "widgets" } } }, + { "id": "xform", "kind": "transform", "name": "Log", + "config": { "set": { "note": "=nodes.post.item.json.data.not_a_probed_field" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "post" }, + { "from_node": "post", "to_node": "xform" } + ] + })); + let warnings = graph_wiring_warnings(&config, &fake_field).await; + assert!( + warnings + .iter() + .any(|w| w.contains("not_a_probed_field") && w.contains("post")), + "{warnings:?}" + ); +} + // ───────────────────────────────────────────────────────────────────────────── // degrade_completed_status (PR2 — run honesty) // ───────────────────────────────────────────────────────────────────────────── diff --git a/src/openhuman/tinyflows/caps.rs b/src/openhuman/tinyflows/caps.rs index d975f2a85..30877158b 100644 --- a/src/openhuman/tinyflows/caps.rs +++ b/src/openhuman/tinyflows/caps.rs @@ -1776,6 +1776,387 @@ pub(crate) fn compute_composio_array_path(schema: Option<&Value>) -> Option Option { + let mut queue: std::collections::VecDeque<(String, &Value)> = std::collections::VecDeque::new(); + queue.push_back((String::new(), value)); + + while let Some((path, node)) = queue.pop_front() { + let Some(obj) = node.as_object() else { + continue; + }; + for (key, v) in obj { + if path.is_empty() && COMPOSIO_ENVELOPE_META_KEYS_AT_ROOT.contains(&key.as_str()) { + continue; + } + if v.is_array() { + let prop_path = if path.is_empty() { + key.clone() + } else { + format!("{path}.{key}") + }; + return Some(prop_path); + } + } + for (key, v) in obj { + if path.is_empty() && COMPOSIO_ENVELOPE_META_KEYS_AT_ROOT.contains(&key.as_str()) { + continue; + } + let prop_path = if path.is_empty() { + key.clone() + } else { + format!("{path}.{key}") + }; + queue.push_back((prop_path, v)); + } + } + None +} + +/// One real, LIVE-sampled Composio action result — [`probe_tool_output_sample`]'s +/// cached ground truth, keyed by action slug (uppercased). Distinct from +/// (and takes priority over, via [`apply_probe_override`]) the schema-derived +/// fields on [`ToolContract`]: a probe is an ACTUAL observed response, not a +/// published schema Composio may or may not provide. +#[derive(Debug, Clone, serde::Serialize)] +pub(crate) struct ProbedOutputSample { + /// Dotted path (relative to the envelope's own `json` field — prefix with + /// `"json."` for a `split_out.path`, same convention as + /// [`ToolContract::primary_array_path`]) to the first array found in the + /// real response. `None` when the real response named no array at all. + pub primary_array_path: Option, + /// Top-level field names of the real response's `data` payload — the + /// probed analogue of [`ToolContract::output_fields`]. + pub output_fields: Vec, + /// The full envelope-shaped sample value the probe observed, verbatim — + /// returned to `probe_tool_output_sample`'s IMMEDIATE caller only for + /// this one call. **Never persisted** into [`PROBE_CACHE`] + /// ([`cache_probe_result`] redacts it to `Value::Null` before inserting) + /// — the process-wide cache is keyed by slug alone, and a real probe + /// response can carry one user/connection/args' actual private data + /// (repo issues, messages, …); nothing else in the process reads a + /// CACHED sample (only the derived `primary_array_path`/`output_fields` + /// do), so retaining the full payload there would be pure unnecessary + /// exposure (see PR #4702 review). + pub sample: Value, +} + +/// Process-level cache backing [`probe_tool_output_sample`]: action slug +/// (uppercased) → the [`ProbedOutputSample`] it produced. One real probe per +/// slug per process — mirrors [`LIVE_CATALOG_CACHE`]'s one-fetch-per-process +/// shape, and for the same reason: a probe is a real, potentially +/// rate-limited/billed external call, not something to repeat every turn. +/// +/// Entries here always have `sample` redacted to `Value::Null` — see +/// [`cache_probe_result`] and [`ProbedOutputSample::sample`]'s doc. +static PROBE_CACHE: std::sync::OnceLock< + std::sync::Mutex>, +> = std::sync::OnceLock::new(); + +/// Seeds the probe cache for a slug — test hook so [`apply_probe_override`] +/// and the enforcement checks that consult a probe can be exercised without a +/// live Composio backend. Unlike [`cache_probe_result`], does NOT redact +/// `sample` — tests seed only small synthetic fixtures, never real user data. +#[cfg(test)] +pub(crate) fn seed_probe_cache(slug: &str, sample: ProbedOutputSample) { + PROBE_CACHE + .get_or_init(Default::default) + .lock() + .expect("probe cache poisoned") + .insert(slug.trim().to_ascii_uppercase(), sample); +} + +/// Caches the DERIVED metadata from a real probe — never the raw `sample` +/// payload itself (redacted to `Value::Null` here). See +/// [`ProbedOutputSample::sample`]'s doc for why: a real probe response can +/// contain one user/connection/args' actual private data, and nothing that +/// reads from the cache (only [`apply_probe_override`]) ever needs the raw +/// payload — only the derived `primary_array_path`/`output_fields`. +fn cache_probe_result(slug: &str, sample: ProbedOutputSample) { + let cached = ProbedOutputSample { + sample: Value::Null, + ..sample + }; + if let Ok(mut cache) = PROBE_CACHE.get_or_init(Default::default).lock() { + cache.insert(slug.trim().to_ascii_uppercase(), cached); + } +} + +/// Looks up a cached [`ProbedOutputSample`] for `slug`, or `None` when +/// [`probe_tool_output_sample`] has never successfully probed it this +/// process. +pub(crate) fn probed_output_sample(slug: &str) -> Option { + PROBE_CACHE + .get_or_init(Default::default) + .lock() + .ok()? + .get(&slug.trim().to_ascii_uppercase()) + .cloned() +} + +/// Overlays a cached [`probed_output_sample`] (if any) onto a schema-derived +/// [`ToolContract`] — the probe, being an ACTUAL observed response, always +/// wins over the schema-derived hint when both are present. A contract with +/// no cached probe passes through unchanged. Called everywhere a +/// [`ToolContract`] is consulted for wiring (`get_tool_contract`, +/// `graph_output_field_warnings`, `graph_split_out_path_warnings`) so a +/// probe the builder already ran is never shadowed by a stale/absent schema. +/// +/// `primary_array_path` is overlaid UNCONDITIONALLY (including `None`) once a +/// probe exists: a probe's `None` is itself meaningful ("the real response +/// named no array anywhere"), not "no opinion" — leaving a stale +/// schema-derived path in place after a real observation disproves it would +/// let a since-confirmed-wrong `split_out.path` keep looking supported (see +/// PR #4702 review). `output_fields` only overlays when non-empty since an +/// empty probe result there genuinely means "unknown", not "confirmed empty". +pub(crate) fn apply_probe_override(mut contract: ToolContract) -> ToolContract { + if let Some(probe) = probed_output_sample(&contract.slug) { + contract.primary_array_path = probe.primary_array_path; + if !probe.output_fields.is_empty() { + contract.output_fields = probe.output_fields; + } + } + contract +} + +/// Best-effort, but FAIL-CLOSED, classification of a Composio action slug's +/// [`ToolScope`] — mirrors [`flow_tool_allowed`]'s Path A / Path B split +/// rather than trusting `classify_unknown`'s verb heuristic unconditionally: +/// +/// - Toolkit has no extractable prefix at all (`toolkit_from_slug` fails): +/// `None` — nothing to confirm a scope against. +/// - Toolkit HAS a static curated catalog (`get_provider().curated_tools()` +/// or `catalog_for_toolkit`): the slug's scope is authoritative ONLY if the +/// slug is actually one of that catalog's curated entries. An uncurated +/// slug on a cataloged toolkit resolves to `None` — it must NOT fall +/// through to the verb heuristic, which can misclassify an uncurated write +/// action (e.g. a connected GitHub/Gmail action not in the curated list) +/// as `Read` by name alone (see PR #4702 review — this exact hole would +/// otherwise let the probe execute a real write). +/// - Toolkit has NO static catalog at all: falls back to `classify_unknown` +/// — the same authority [`flow_tool_allowed`]'s Path B accepts once it has +/// already confirmed (via its own connected + live-catalog checks) the +/// slug is real; here it's just the scope signal, gated further below. +/// +/// Used exclusively by [`probe_tool_output_sample`] to hard-refuse anything +/// that isn't a CONFIDENTLY CONFIRMED `Read` action REGARDLESS of the user's +/// per-toolkit scope preference — unlike [`flow_tool_allowed`], which honors +/// a user's opt-in to Write/Admin for a real `tool_call` node, a +/// schema-discovery probe must never perform a real mutation no matter what +/// the user has toggled on, and must never rely on a heuristic guess to +/// decide that: the builder never asked for (and the user never approved) +/// THIS specific write. `None` means "refuse — no confirmed Read scope", not +/// "assume Read". +fn resolve_composio_action_scope( + slug: &str, +) -> Option { + use crate::openhuman::memory_sync::composio::providers::{ + catalog_for_toolkit, classify_unknown, find_curated, get_provider, toolkit_from_slug, + }; + + let toolkit = toolkit_from_slug(slug)?; + match get_provider(&toolkit) + .and_then(|p| p.curated_tools()) + .or_else(|| catalog_for_toolkit(&toolkit)) + { + // A static catalog exists for this toolkit — only a genuinely + // curated entry's scope is trustworthy; an uncurated slug fails + // closed rather than being guessed via the verb heuristic. + Some(catalog) => find_curated(catalog, slug).map(|curated| curated.scope), + // No static catalog anywhere for this toolkit — the heuristic is + // the only available signal (still gated by the connected-toolkit + // check below in `probe_tool_output_sample`). + None => Some(classify_unknown(slug)), + } +} + +/// `get_tool_output_sample`'s implementation — see the module comment above +/// this section for why it exists. Gates, in order (fail closed on any): +/// +/// 1. **Scope**: [`resolve_composio_action_scope`] must CONFIRM `slug` as +/// `Read` (`None` — no confirmed scope, e.g. an uncurated slug on a +/// cataloged toolkit — refuses exactly like a confirmed non-`Read` scope +/// does; it is never treated as "assume Read"). +/// 2. **Connected**: the slug's toolkit must have an active Composio +/// connection. +/// +/// On success, derives + caches a [`ProbedOutputSample`] (process-lifetime, +/// keyed by slug) and returns it. `args` is forwarded verbatim to the real +/// call — the builder should pass the SAME arguments it intends to wire into +/// the real `tool_call` node (this is a sample of THAT call, not a generic +/// fixture); omitted/`null` becomes `{}`, which is fine for a +/// zero-required-arg action. +pub(crate) async fn probe_tool_output_sample( + config: &Config, + slug: &str, + args: Value, +) -> std::result::Result { + let slug = slug.trim(); + if slug.is_empty() { + return Err("get_tool_output_sample: slug must not be empty".to_string()); + } + + match resolve_composio_action_scope(slug) { + Some(crate::openhuman::memory_sync::composio::providers::ToolScope::Read) => {} + Some(other) => { + tracing::warn!( + target: "flows", + %slug, + scope = other.as_str(), + "[flows] get_tool_output_sample: refused — not a Read-scope action" + ); + return Err(format!( + "get_tool_output_sample refuses `{slug}`: classified as {} — this probe is \ + READ-only and never performs a real mutation, regardless of the user's scope \ + preference. Use get_tool_contract for its schema-derived (possibly unknown) \ + output shape instead.", + other.as_str() + )); + } + None => { + tracing::warn!( + target: "flows", + %slug, + "[flows] get_tool_output_sample: refused — no confirmed Read scope (either no \ + extractable toolkit, or an uncurated slug on a toolkit with a static curated \ + catalog — fails closed rather than guessing via the verb heuristic)" + ); + return Err(format!( + "get_tool_output_sample refuses `{slug}`: could not confirm this is a Read-scope \ + action. Either no toolkit could be extracted from the slug, or its toolkit ships \ + a static curated catalog and this slug is not one of its curated actions — this \ + probe never falls back to a verb-name heuristic in that case, since an uncurated \ + action on a cataloged toolkit could really be a write. Use get_tool_contract for \ + its schema-derived (possibly unknown) output shape instead." + )); + } + } + + let Some(toolkit) = crate::openhuman::memory_sync::composio::providers::toolkit_from_slug(slug) + else { + return Err(format!( + "get_tool_output_sample: could not extract a toolkit from slug '{slug}' — it must \ + look like '_'." + )); + }; + + let integrations = crate::openhuman::composio::fetch_connected_integrations(config).await; + let connected = integrations + .iter() + .any(|i| i.connected && i.toolkit.eq_ignore_ascii_case(&toolkit)); + if !connected { + tracing::warn!(target: "flows", %slug, %toolkit, "[flows] get_tool_output_sample: refused — toolkit not connected"); + return Err(format!( + "get_tool_output_sample refuses `{slug}`: the '{toolkit}' toolkit has no active \ + Composio connection for this user — connect it first (composio_connect), or fall \ + back to get_tool_contract's schema-derived hint." + )); + } + + tracing::debug!( + target: "flows", + %slug, + %toolkit, + "[flows] get_tool_output_sample: probing the real live response (read-only, bounded, one call)" + ); + + let kind = create_composio_client(config).map_err(|e| e.to_string())?; + let args_opt = if args.is_null() { None } else { Some(args) }; + let resp = match kind { + ComposioClientKind::Backend(client) => client + .execute_tool(slug, args_opt) + .await + .map_err(|e| format!("get_tool_output_sample: real call to `{slug}` failed: {e}"))?, + ComposioClientKind::Direct(tool) => { + direct_execute(&tool, slug, args_opt, &config.composio.entity_id, None) + .await + .map_err(|e| format!("get_tool_output_sample: real call to `{slug}` failed: {e}"))? + } + }; + + if !resp.successful { + let detail = resp + .error + .as_deref() + .map(str::trim) + .filter(|e| !e.is_empty()) + .unwrap_or("no error detail returned by the provider"); + return Err(format!( + "get_tool_output_sample: `{slug}` reported failure at the connected provider: {detail}" + )); + } + + let envelope = serde_json::to_value(&resp).map_err(|e| { + format!("get_tool_output_sample: could not serialize the real response: {e}") + })?; + let primary_array_path = compute_primary_array_path_from_value(&envelope); + let output_fields = resp + .data + .as_object() + .map(|obj| obj.keys().cloned().collect()) + .unwrap_or_default(); + + let sample = ProbedOutputSample { + primary_array_path, + output_fields, + sample: envelope, + }; + cache_probe_result(slug, sample.clone()); + tracing::info!( + target: "flows", + %slug, + primary_array_path = ?sample.primary_array_path, + "[flows] get_tool_output_sample: probed + cached the real output shape" + ); + Ok(sample) +} + /// 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 @@ -4034,6 +4415,222 @@ mod tests { ); } + // ── compute_primary_array_path_from_value (B12: the real-output probe) ── + + #[test] + fn compute_primary_array_path_from_value_finds_a_named_array_under_data() { + // The exact GITHUB_LIST_REPOSITORY_ISSUES shape observed live: the + // real array lives at `data.issues` (a NAMED field), not `data.items` + // — and there is no schema at all to derive this from (verified live: + // `output_schema: null` for this action), so only a real-value probe + // can find it. + let value = json!({ + "data": { "issues": [ { "id": 1 }, { "id": 2 } ], "total_count": 2 }, + "successful": true, + "error": null, + "costUsd": 0.0, + "markdownFormatted": null + }); + assert_eq!( + compute_primary_array_path_from_value(&value), + Some("data.issues".to_string()) + ); + } + + #[test] + fn compute_primary_array_path_from_value_skips_envelope_metadata_at_the_root() { + // None of the envelope's OTHER top-level fields are ever arrays in + // practice, but the skip-list is explicit so one never wins a + // shallowest-wins tie against a real nested array. + let value = json!({ + "successful": true, + "error": null, + "costUsd": 0.0, + "markdownFormatted": null, + "data": { "messages": ["a", "b"] } + }); + assert_eq!( + compute_primary_array_path_from_value(&value), + Some("data.messages".to_string()) + ); + } + + #[test] + fn compute_primary_array_path_from_value_none_when_no_array_anywhere() { + let value = json!({ + "data": { "id": "abc123", "name": "octocat" }, + "successful": true + }); + assert_eq!(compute_primary_array_path_from_value(&value), None); + assert_eq!(compute_primary_array_path_from_value(&json!(null)), None); + assert_eq!( + compute_primary_array_path_from_value(&json!("scalar")), + None + ); + } + + // ── apply_probe_override (B12) ─────────────────────────────────────────── + + fn bare_contract(slug: &str) -> ToolContract { + ToolContract { + slug: slug.to_string(), + toolkit: "github".to_string(), + description: None, + required_args: vec![], + input_schema: None, + output_fields: vec![], + output_schema: None, + primary_array_path: None, + is_curated: true, + } + } + + #[test] + fn apply_probe_override_overlays_a_cached_probe_onto_a_schemaless_contract() { + seed_probe_cache( + "PROBETEST_LIST_REPOSITORY_ISSUES", + ProbedOutputSample { + primary_array_path: Some("data.issues".to_string()), + output_fields: vec!["issues".to_string(), "total_count".to_string()], + sample: json!({ "data": { "issues": [], "total_count": 0 } }), + }, + ); + let contract = bare_contract("PROBETEST_LIST_REPOSITORY_ISSUES"); + assert_eq!(contract.primary_array_path, None); + let overridden = apply_probe_override(contract); + assert_eq!( + overridden.primary_array_path, + Some("data.issues".to_string()) + ); + assert_eq!( + overridden.output_fields, + vec!["issues".to_string(), "total_count".to_string()] + ); + } + + #[test] + fn apply_probe_override_passes_through_unchanged_without_a_cached_probe() { + let contract = bare_contract("PROBETEST_SOME_UNPROBED_ACTION"); + let overridden = apply_probe_override(contract.clone()); + assert_eq!(overridden.primary_array_path, contract.primary_array_path); + assert_eq!(overridden.output_fields, contract.output_fields); + } + + /// CodeRabbit (PR #4702 review): a probe that OBSERVED the real response + /// and found no array anywhere must CLEAR a stale schema-derived + /// `primary_array_path`, not merely leave it in place because the probe's + /// own path happens to be `None`. A schema-derived path a real + /// observation just disproved is worse than no path at all — it would + /// otherwise keep suggesting a `split_out.path` the probe itself showed + /// is wrong. + #[test] + fn apply_probe_override_clears_a_stale_schema_path_when_the_probe_finds_no_array() { + seed_probe_cache( + "PROBETEST_CLEARS_STALE_PATH", + ProbedOutputSample { + primary_array_path: None, + output_fields: vec![], + sample: json!({ "data": { "id": "abc123" } }), + }, + ); + let mut contract = bare_contract("PROBETEST_CLEARS_STALE_PATH"); + contract.primary_array_path = Some("data.items".to_string()); + let overridden = apply_probe_override(contract); + assert_eq!(overridden.primary_array_path, None); + } + + /// PR #4702 review (security): the process-wide [`PROBE_CACHE`] must + /// never retain the raw observed payload — only derived metadata. A real + /// probe response can carry one user/connection/args' actual private + /// data (repo issues, messages, …), and nothing that reads the CACHE + /// (only [`apply_probe_override`], via [`probed_output_sample`]) ever + /// needs the raw payload. + #[test] + fn cache_probe_result_redacts_the_raw_sample_before_caching() { + cache_probe_result( + "PROBETEST_REDACTS_SAMPLE", + ProbedOutputSample { + primary_array_path: Some("data.issues".to_string()), + output_fields: vec!["issues".to_string()], + sample: json!({ "data": { "issues": [{"secret": "do-not-retain"}] } }), + }, + ); + let cached = + probed_output_sample("PROBETEST_REDACTS_SAMPLE").expect("just cached this slug"); + assert_eq!(cached.sample, Value::Null); + // The derived metadata is still cached faithfully — only the raw + // payload is redacted. + assert_eq!(cached.primary_array_path, Some("data.issues".to_string())); + } + + // ── resolve_composio_action_scope (B12: hard Read-only gate) ───────────── + + #[test] + fn resolve_composio_action_scope_uses_the_curated_catalog_when_available() { + use crate::openhuman::memory_sync::composio::providers::ToolScope; + // GITHUB_LIST_REPOSITORY_ISSUES is curated as Read (github/tools.rs). + assert_eq!( + resolve_composio_action_scope("GITHUB_LIST_REPOSITORY_ISSUES"), + Some(ToolScope::Read) + ); + // A curated Write action must classify as Write, not Read — the probe + // must refuse it regardless of the verb heuristic agreeing or not. + assert_eq!( + resolve_composio_action_scope("GMAIL_SEND_EMAIL"), + Some(ToolScope::Write) + ); + } + + /// PR #4702 review (P1): a toolkit with a static curated catalog (like + /// `github`) must NOT fall through to the `classify_unknown` verb + /// heuristic for a slug that isn't actually one of its curated actions — + /// `GITHUB_LIST_WORKFLOWS` is a REAL GitHub action name (reads as + /// Read-scope by its `LIST` verb) that was deliberately left uncurated + /// (see the commented-out entry in `github/tools.rs`), so this must + /// resolve to `None` (fail closed), not `Some(ToolScope::Read)` — the + /// heuristic agreeing with the "looks safe" name is exactly the + /// misclassification hole this guards against. + #[test] + fn resolve_composio_action_scope_rejects_an_uncurated_slug_on_a_cataloged_toolkit() { + assert_eq!(resolve_composio_action_scope("GITHUB_LIST_WORKFLOWS"), None); + } + + #[test] + fn resolve_composio_action_scope_falls_back_to_the_verb_heuristic_only_without_a_static_catalog( + ) { + use crate::openhuman::memory_sync::composio::providers::ToolScope; + assert_eq!( + resolve_composio_action_scope("MADEUPTOOLKIT_LIST_THINGS"), + Some(ToolScope::Read) + ); + assert_eq!( + resolve_composio_action_scope("MADEUPTOOLKIT_DELETE_THING"), + Some(ToolScope::Admin) + ); + } + + // ── probe_tool_output_sample (B12: gates) ──────────────────────────────── + + #[tokio::test] + async fn probe_tool_output_sample_refuses_a_non_read_action_before_any_client_call() { + let config = Config::default(); + let result = probe_tool_output_sample(&config, "GMAIL_SEND_EMAIL", json!({})).await; + let err = result.expect_err("a Write action must be refused"); + assert!(err.contains("READ-only"), "{err}"); + } + + /// PR #4702 review (P1): the probe entry point itself must refuse an + /// uncurated-but-read-sounding slug on a cataloged toolkit BEFORE any + /// client call — not just `resolve_composio_action_scope` in isolation. + #[tokio::test] + async fn probe_tool_output_sample_refuses_an_uncurated_slug_on_a_cataloged_toolkit_before_any_client_call( + ) { + let config = Config::default(); + let result = probe_tool_output_sample(&config, "GITHUB_LIST_WORKFLOWS", json!({})).await; + let err = result.expect_err("an uncurated slug on a cataloged toolkit must be refused"); + assert!(err.contains("could not confirm"), "{err}"); + } + // ── fetch_live_toolkit_catalog / composio_required_args / // composio_response_fields delegation ───────────────────────────────── diff --git a/src/openhuman/tools/ops.rs b/src/openhuman/tools/ops.rs index 5f335626a..c3c12e7cc 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -301,6 +301,15 @@ pub fn all_tools_with_runtime( // 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())), + // B12: ONE bounded, READ-ONLY, REAL Composio call to derive the real + // primary_array_path/output_fields when the live listing publishes no + // output schema at all (verified for every GitHub action) — overrides + // get_tool_contract's schema-derived hint for that slug from then on. + // Read-scope actions only (hard-refused otherwise), connected + // toolkits only — see builder_tools.rs's module doc for the carve-out + // this makes in the workflow-builder agent's "no composio_execute" + // invariant. + Box::new(GetToolOutputSampleTool::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.