From 3796e30af30022cb4b621c053c3881f0941ec5ba Mon Sep 17 00:00:00 2001 From: Cyrus Gray <144336577+graycyrus@users.noreply.github.com> Date: Mon, 13 Jul 2026 21:04:01 +0530 Subject: [PATCH] fix(flows): condition-routing hard gate, engine validation, and null-arg reject (#4826) --- .../flows/agents/workflow_builder/prompt.md | 23 ++ src/openhuman/flows/builder_tools.rs | 43 ++- src/openhuman/flows/ops.rs | 268 ++++++++++++++++++ src/openhuman/flows/ops_tests.rs | 202 +++++++++++++ src/openhuman/flows/tools.rs | 33 ++- vendor/tinyflows | 2 +- 6 files changed, 564 insertions(+), 7 deletions(-) diff --git a/src/openhuman/flows/agents/workflow_builder/prompt.md b/src/openhuman/flows/agents/workflow_builder/prompt.md index a9805b4f0..ed61c6335 100644 --- a/src/openhuman/flows/agents/workflow_builder/prompt.md +++ b/src/openhuman/flows/agents/workflow_builder/prompt.md @@ -160,6 +160,13 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. - **Node:** `{ id, kind, name, config }`. `id` is unique within the graph. - **Edge:** `{ from_node, to_node, from_port?, to_port? }`. Ports default to `"main"`. Branch nodes emit on named ports (below) — wire those explicitly. + **The branch label ALWAYS goes on `from_port` — never on `to_port`.** + Routing is keyed exclusively on the SOURCE node's `from_port`; `to_port` + is not consulted to pick a successor, so a branch label put on `to_port` + instead (a common mistake) is silently wrong: `save_workflow`/ + `propose_workflow`/`revise_workflow` now HARD-REJECT it (a `condition` + node's outgoing edges must have `from_port` in `"true"`/`"false"`), so + fix the graph and call the tool again if you see that error. - **Exactly ONE `trigger` node is required.** Every other node should be reachable from it; a dry-run helps catch orphans. @@ -288,6 +295,22 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. `output_parser.schema` property MUST be declared `"type": "boolean"` (see the `agent` node kind above) — an untyped/string field can carry the truthy string `"false"` and route to the wrong port. + + **The branch label is the edge's `from_port`, not `to_port`** — `to_port` + on an edge leaving a `condition` node just stays `"main"` (or is omitted). + Given a condition node `"gate"` with a `"true"` successor `"send_summary"` + and a `"false"` successor `"done"`, the two outgoing edges are: + ```json + { "from_node": "gate", "from_port": "true", "to_node": "send_summary", "to_port": "main" }, + { "from_node": "gate", "from_port": "false", "to_node": "done", "to_port": "main" } + ``` + NOT `{ "from_node": "gate", "from_port": "main", "to_node": "send_summary", "to_port": "true" }` + — that shape puts both edges in the same `from_port` group, which the + engine treats as an unconditional parallel fan-out (BOTH branches run + every time, regardless of the condition's actual result). This is + enforced: `propose_workflow`/`revise_workflow`/`save_workflow` reject a + `condition` node whose outgoing edges don't emit `"true"`/`"false"` on + `from_port`. 7. **`switch`** — multi-way on `config.expression` or `config.field`; routes to the matching **case** port, else **`default`**. 8. **`merge`** — fan-in barrier; passes inputs through. No config. diff --git a/src/openhuman/flows/builder_tools.rs b/src/openhuman/flows/builder_tools.rs index 63c6cb222..7eeb2c6cb 100644 --- a/src/openhuman/flows/builder_tools.rs +++ b/src/openhuman/flows/builder_tools.rs @@ -215,6 +215,24 @@ impl Tool for ReviseWorkflowTool { ))); } + // Required-arg resolvability gate (issue B18): reject outright — not + // just warn — a REQUIRED outbound arg that LOOKS wired but resolves + // to `null` in a sandboxed test run. See + // `ops::validate_required_arg_resolvability`. + let null_arg_errors = ops::validate_required_arg_resolvability(&graph).await; + if !null_arg_errors.is_empty() { + tracing::debug!( + target: "flows", + %name, + error_count = null_arg_errors.len(), + "[flows] revise_workflow: required-arg resolvability check rejected the revised graph" + ); + return Ok(ToolResult::error(format!( + "{}\n\nFix these bindings and call revise_workflow again.", + null_arg_errors.join("\n\n") + ))); + } + let summary = super::tools::build_summary(&graph); let mut warnings = ops::graph_trigger_warnings(&graph); // Author-time wiring check: unwired REQUIRED Composio args come back @@ -1604,8 +1622,12 @@ fn tool_call_error_message(output: &Value, node_id: &str) -> Option { /// traced during that node's config resolution) — so [`DryRunWorkflowTool`] /// can inspect them once the sandbox run settles. See the struct's "Null- /// resolution check" doc for why this exists. +/// `pub(crate)` (not private) so [`crate::openhuman::flows::ops::validate_required_arg_resolvability`] +/// (issue B18 — escalating a null-resolved REQUIRED outbound arg to a hard +/// authoring-time reject) can run the identical sandbox-capture shape without +/// duplicating this struct. #[derive(Default)] -struct CapturingObserver { +pub(crate) struct CapturingObserver { steps: std::sync::Mutex>, } @@ -1622,7 +1644,7 @@ impl CapturingObserver { /// A snapshot of every step recorded so far (steps are pushed /// synchronously from `on_step_finish`, so once the run's future resolves /// every step it will ever record is already present). - fn steps(&self) -> Vec { + pub(crate) fn steps(&self) -> Vec { self.steps .lock() .expect("CapturingObserver steps mutex poisoned") @@ -1784,6 +1806,23 @@ impl Tool for SaveWorkflowTool { contract_errors.join("\n\n") ))); } + // Required-arg resolvability gate (issue B18): reject outright — not + // just warn — a REQUIRED outbound arg that LOOKS wired but resolves + // to `null` in a sandboxed test run, before the graph is ever + // persisted. See `ops::validate_required_arg_resolvability`. + let null_arg_errors = ops::validate_required_arg_resolvability(&graph).await; + if !null_arg_errors.is_empty() { + tracing::debug!( + target: "flows", + %flow_id, + error_count = null_arg_errors.len(), + "[flows] save_workflow: required-arg resolvability check rejected the graph" + ); + return Ok(ToolResult::error(format!( + "{}\n\nFix these bindings and call save_workflow again.", + null_arg_errors.join("\n\n") + ))); + } // Author-time warnings (unfired trigger kinds + unwired REQUIRED // Composio args) were previously computed by propose/revise but never // surfaced again at save time — add them here so the agent sees any diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index b487efd91..5b93bf7e6 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -1069,6 +1069,274 @@ pub(crate) async fn validate_tool_contracts(config: &Config, graph: &WorkflowGra errors } +// ───────────────────────────────────────────────────────────────────────────── +// Required-arg resolvability gate (issue B18) +// ───────────────────────────────────────────────────────────────────────────── +// +// `validate_tool_contracts` (above) proves a required arg is PRESENT +// (`missing_required_args`: absent or literal `null`) — it has no opinion on +// whether an arg wired to a real-looking `=`-expression actually RESOLVES to +// something at runtime, and it says nothing at all about an arg the live +// schema doesn't individually mark `required` even though the PROVIDER +// enforces it as a business rule — e.g. `GMAIL_SEND_EMAIL.subject`/`.body` +// are each individually optional in the schema, but Gmail rejects a send +// where BOTH are empty ("At least one of 'subject' or 'body' must be +// provided with non-empty content"). A builder can wire either to an +// upstream path that looks fully wired but resolves `null`, and neither +// static check above has anything to say about it. +// +// `crate::openhuman::flows::builder_tools::DryRunWorkflowTool` already +// detects exactly this class of null resolution (`null_resolutions`) by +// running the graph through the same MOCK sandbox — but only as information +// the agent is *instructed* (by prompt, not enforced in code) to act on +// before calling `propose_workflow`/`save_workflow`. Nothing previously +// stopped those tools from persisting the graph anyway. +// [`validate_required_arg_resolvability`] closes that gap: it re-runs the +// identical sandbox check and escalates ANY arg of a real (non-`=`-derived, +// non-native) `tool_call` node that resolved `null` to a hard reject, wired +// into `propose_workflow` / `revise_workflow` / `save_workflow` alongside +// [`validate_binding_resolvability`] and [`validate_tool_contracts`]. + +/// Wall-clock bound on the sandbox run this gate performs. Mirrors +/// `builder_tools::DRY_RUN_TIMEOUT_SECS`'s purpose but kept short: unlike the +/// opt-in `dry_run_workflow` tool, this check runs on EVERY +/// propose/revise/save call, so a slow or pathological draft must not stall +/// authoring. +const REQUIRED_ARG_NULL_CHECK_TIMEOUT_SECS: u64 = 15; + +/// Sandbox-executes `graph` against `tinyflows`' deterministic MOCK +/// capabilities (the same shape `DryRunWorkflowTool` uses — see this +/// section's module doc) and returns one human-readable error per arg of a +/// real (non-`=`-derived, non-native) `tool_call` node whose `=`-expression +/// resolved to `null` during that run **and** whose expression is wired to a +/// specific upstream node's output (directly, via the implicit +/// `item`/`items` scope, or explicitly via `nodes....`) rather than to +/// the trigger. +/// +/// This run always sandboxes against `json!({})` as the trigger payload (see +/// below), so any arg wired to trigger-scoped data — `=item.` / +/// `=items...` fed directly from the trigger node, or `=run.` (the +/// trigger metadata itself) — legitimately resolves `null` here even though a +/// real webhook/app-event/manual trigger WILL populate it at runtime. Hard +/// gate that on an empty mock run would reject every ordinary trigger-bound +/// workflow (Codex feedback on PR #4826). Only a `null` resolved from a +/// genuine upstream **node** reference is escalated — that's the real B18 +/// bug this gate exists to catch: an arg wired to a node output path that can +/// never resolve (e.g. `GMAIL_SEND_EMAIL.subject = +/// "=nodes.build_body.item.subject"` where `build_body` never produces +/// `subject`), which stays broken no matter what the trigger payload is. +/// +/// Deliberately does **not** wrap the mock `ToolInvoker` in +/// [`crate::openhuman::tinyflows::caps::PreflightToolInvoker`] the way +/// `DryRunWorkflowTool` does: that wrapper aborts the WHOLE sandbox run the +/// instant a node with a `stop` `on_error` policy (the default) hits a +/// schema-required null arg, which would lose the per-field diagnostic this +/// gate exists to report for every OTHER node — and this check cares about +/// EVERY arg, not just ones the schema happens to mark `required`. The plain +/// mock tool invoker always "succeeds" (a deterministic echo), so the run +/// settles and every node's config-resolution diagnostics get captured +/// regardless of on_error policy or schema required-ness. +/// +/// Best-effort, same posture as [`validate_tool_contracts`]: a compile +/// failure (structural errors are already caught by +/// [`validate_and_migrate_graph`] before this gate ever runs) or a sandbox +/// error/timeout is SKIPPED — never turned into a false rejection. This +/// check only ever adds a diagnostic the sandbox actually observed. +pub(crate) async fn validate_required_arg_resolvability(graph: &WorkflowGraph) -> Vec { + use crate::openhuman::flows::builder_tools::CapturingObserver; + use crate::openhuman::tinyflows::caps::SchemaAwareMockAgentRunner; + + let Ok(compiled) = tinyflows::compiler::compile(graph) else { + return Vec::new(); + }; + + let caps = tinyflows::caps::mock::mock_capabilities_with_agent(SchemaAwareMockAgentRunner); + + let observer = Arc::new(CapturingObserver::default()); + let observer_dyn: Arc = observer.clone(); + let run = tinyflows::engine::run_with_observer(&compiled, json!({}), &caps, &observer_dyn); + if tokio::time::timeout( + std::time::Duration::from_secs(REQUIRED_ARG_NULL_CHECK_TIMEOUT_SECS), + run, + ) + .await + .is_err() + { + // Timed out — a different class of problem than this gate exists to + // catch; never block authoring on it here. + return Vec::new(); + } + // A sandbox `Err` outcome here is a compile/capability issue unrelated + // to null args (the plain mock invoker never itself fails) — surfaced by + // the other gates / `dry_run_workflow` instead; this gate only adds + // diagnostics from a run that actually settled, so an error is silently + // skipped rather than turned into a (misleading) empty-errors success. + + let tool_call_slugs: std::collections::HashMap<&str, &str> = graph + .nodes + .iter() + .filter(|n| n.kind == NodeKind::ToolCall) + .filter_map(|n| { + let slug = n.config.get("slug").and_then(Value::as_str)?; + Some((n.id.as_str(), slug)) + }) + .collect(); + + // The trigger node's id, if any — used below to tell a trigger-scoped + // `item`/`items` reference (the direct predecessor IS the trigger) apart + // from a real upstream-node reference. Graphs are expected to have + // exactly one trigger; `flows_validate` rejects zero/multiple before this + // gate ever runs, so `first()` here doesn't hide ambiguity. + let trigger_id: Option<&str> = graph + .nodes + .iter() + .find(|n| n.kind == NodeKind::Trigger) + .map(|n| n.id.as_str()); + + let mut errors = Vec::new(); + for step in observer.steps() { + let Some(&slug) = tool_call_slugs.get(step.node_id.as_str()) else { + continue; + }; + // `=`-derived slugs resolve from upstream/trigger data at runtime; + // native `oh:` tools have no external-provider rejection mode. + if slug.starts_with('=') || slug.starts_with("oh:") { + continue; + } + for diag in &step.diagnostics { + let Some(field) = diag.location.strip_prefix("args.") else { + continue; + }; + if is_trigger_scoped_expression(&diag.expression, graph, &step.node_id, trigger_id) { + // Legitimately empty in this gate's `{}` mock run — the real + // trigger (webhook/app-event/manual) will populate it. Not + // the B18 broken-wiring case this gate exists to catch. + tracing::debug!( + target: "flows", + node = %step.node_id, + %slug, + %field, + expression = %diag.expression, + "[flows] required-arg resolvability check: trigger-scoped null in empty \ + mock run — not rejecting" + ); + continue; + } + tracing::warn!( + target: "flows", + node = %step.node_id, + %slug, + %field, + expression = %diag.expression, + "[flows] required-arg resolvability check: arg resolved null in sandbox — \ + rejecting" + ); + errors.push(format!( + "Node '{}': arg `{field}` of `{slug}` (`{}`) resolved to `null` during a \ + sandboxed test run — an empty/missing `{field}` can be rejected by the real \ + provider at runtime (e.g. Gmail rejects a send with no subject or body). \ + Rewire it from an upstream node's output that actually has a value — call \ + dry_run_workflow to see exactly which upstream field is null — or drop the \ + field from args if it isn't really needed.", + step.node_id, diag.expression + )); + } + } + errors +} + +/// Returns the node id an explicit `nodes....` expression addresses — +/// either the legacy dotted shorthand (`=nodes.build_body.item.subject`) or +/// the jq bracket form (`=.nodes["build_body"].item.subject`) — or `None` if +/// the expression's root isn't the `nodes` scope key at all. The expression +/// scope's shape (`item` / `items` / `run` / `nodes`) is documented on +/// `tinyflows`'s `expr` module and `nodes::expr_scope`. +fn explicit_nodes_ref(expr: &str) -> Option<&str> { + let body = expr.strip_prefix('=')?.trim(); + let body = body.strip_prefix('.').unwrap_or(body); + let rest = body.strip_prefix("nodes")?; + if let Some(after_dot) = rest.strip_prefix('.') { + // Dotted shorthand: `nodes..item.` — the id ends at the + // next `.` or `[`. + let id = after_dot.split(['.', '[']).next()?; + (!id.is_empty()).then_some(id) + } else if let Some(after_bracket) = rest.strip_prefix('[') { + // jq bracket form: `nodes[""]` / `nodes['']`. + let after_bracket = after_bracket.trim_start(); + let after_bracket = after_bracket + .strip_prefix('"') + .or_else(|| after_bracket.strip_prefix('\'')) + .unwrap_or(after_bracket); + let id = after_bracket.split(['"', '\'', ']']).next()?; + (!id.is_empty()).then_some(id) + } else { + // `rest` is empty (bare `nodes`) or continues some other identifier + // (e.g. a hypothetical `nodesomething` — not this scope key at all). + None + } +} + +/// Whether a null-resolved config expression on `node_id` is scoped to the +/// TRIGGER's data rather than a specific upstream node's output — and +/// therefore legitimately empty in [`validate_required_arg_resolvability`]'s +/// `{}` mock run rather than evidence of broken wiring (see that function's +/// doc comment and the Codex feedback it links). +/// +/// - `=run...` always addresses the trigger payload/metadata directly +/// (`crate::openhuman::tinyflows`'s `expr_scope` docs) — always +/// trigger-scoped. +/// - `=nodes....` / `=.nodes[""]...` explicitly names an upstream +/// node. Trigger-scoped only if `` IS the trigger node; naming any +/// other node is exactly the B18 broken-wiring case this gate exists to +/// catch, so it is never treated as trigger-scoped. +/// - `=item...` / `=items...` implicitly addresses `node_id`'s direct +/// predecessor(s) output. Trigger-scoped only when EVERY incoming edge to +/// `node_id` comes from the trigger node — a fan-in that mixes the trigger +/// with a real upstream node, or an `item`/`items` reference fed entirely +/// by real upstream nodes, keeps the existing (reject) behavior, since a +/// node that already ran in the sandbox is expected to have produced its +/// real, deterministic output. +/// - Anything else (a jq expression not rooted at one of the above, or a +/// malformed one) is conservatively treated as NOT trigger-scoped, matching +/// this gate's pre-existing behavior. +fn is_trigger_scoped_expression( + expr: &str, + graph: &WorkflowGraph, + node_id: &str, + trigger_id: Option<&str>, +) -> bool { + let body = expr.strip_prefix('=').unwrap_or(expr).trim(); + let body = body.strip_prefix('.').unwrap_or(body); + + if body == "run" || body.starts_with("run.") || body.starts_with("run[") { + return true; + } + + if let Some(referenced_id) = explicit_nodes_ref(expr) { + return trigger_id == Some(referenced_id); + } + + let is_item_scoped = body == "item" + || body.starts_with("item.") + || body.starts_with("item[") + || body == "items" + || body.starts_with("items.") + || body.starts_with("items["); + if !is_item_scoped { + return false; + } + + let Some(trigger_id) = trigger_id else { + return false; + }; + let mut predecessors = graph + .edges + .iter() + .filter(|e| e.to_node == node_id) + .peekable(); + predecessors.peek().is_some() && predecessors.all(|e| e.from_node == trigger_id) +} + /// Validates a candidate graph without persisting it — the same /// migrate/validate path `flows_create` and `ProposeWorkflowTool` use — and /// reports structural errors alongside non-fatal trigger warnings diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index 9b1b93e37..b3dd1234b 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -1916,6 +1916,137 @@ async fn validate_tool_contracts_passes_a_fully_wired_real_slug() { assert!(errors.is_empty(), "{errors:?}"); } +// ── validate_required_arg_resolvability (issue B18) ───────────────────────── +// +// `validate_tool_contracts`'s `missing_required_args` only proves an arg is +// PRESENT (absent/literal-null) — it says nothing about whether an arg wired +// to a real-looking `=`-expression actually RESOLVES to a value at runtime, +// nor about an arg the schema doesn't individually mark `required` even +// though the provider enforces it as a business rule (the real B18 bug: +// `GMAIL_SEND_EMAIL.subject`/`.body` are each optional in the schema, but +// Gmail rejects a send where both are empty). These tests sandbox-run the +// graph the same way `dry_run_workflow` does and prove ANY tool_call arg +// that resolves `null` (because it's bound to a field that doesn't exist +// upstream) is a hard reject, while a fully-resolved graph passes clean. No +// live-catalog seeding needed — this check doesn't consult the Composio +// schema at all, only the sandbox's own traced diagnostics. + +#[tokio::test] +async fn validate_required_arg_resolvability_rejects_a_null_resolved_arg() { + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "prep", "kind": "code", "name": "Prep", + "config": { "language": "javascript", "source": "return {};" } }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "GMAIL_SEND_EMAIL", + "args": { "recipient_email": "a@b.com", "subject": "=item.nonexistent_field" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "prep" }, + { "from_node": "prep", "to_node": "post" } + ] + })); + let errors = validate_required_arg_resolvability(&g).await; + assert_eq!(errors.len(), 1, "{errors:?}"); + assert!(errors[0].contains("post"), "{}", errors[0]); + assert!(errors[0].contains("`subject`"), "{}", errors[0]); + assert!(errors[0].contains("GMAIL_SEND_EMAIL"), "{}", errors[0]); +} + +#[tokio::test] +async fn validate_required_arg_resolvability_accepts_a_fully_resolved_graph() { + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "GMAIL_SEND_EMAIL", + "args": { "recipient_email": "a@b.com", "subject": "hello", "body": "hi there" } } } + ], + "edges": [ { "from_node": "t", "to_node": "post" } ] + })); + let errors = validate_required_arg_resolvability(&g).await; + assert!(errors.is_empty(), "{errors:?}"); +} + +#[tokio::test] +async fn validate_required_arg_resolvability_ignores_native_and_dynamic_slugs() { + // `oh:` native tools and `=`-derived slugs have no external-provider + // rejection mode this gate should be checking. + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "prep", "kind": "code", "name": "Prep", + "config": { "language": "javascript", "source": "return {};" } }, + { "id": "native", "kind": "tool_call", "name": "Native", + "config": { "slug": "oh:web_search", + "args": { "query": "=item.nonexistent_field" } } }, + { "id": "dynamic", "kind": "tool_call", "name": "Dynamic", + "config": { "slug": "=item.nonexistent_field", + "args": { "x": "=item.nonexistent_field" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "prep" }, + { "from_node": "prep", "to_node": "native" }, + { "from_node": "native", "to_node": "dynamic" } + ] + })); + let errors = validate_required_arg_resolvability(&g).await; + assert!(errors.is_empty(), "{errors:?}"); +} + +/// (Codex feedback on PR #4826) This gate sandbox-runs every graph against +/// `json!({})` as the trigger payload, so a `tool_call` arg wired straight to +/// the trigger's own data — `"to": "=item.email"` on a node whose only +/// predecessor is the trigger — always resolves `null` here, even though a +/// real webhook/app-event/manual trigger fires with a real payload. Hard- +/// rejecting that blocked every ordinary trigger-bound workflow. Contrast +/// with `validate_required_arg_resolvability_rejects_a_null_resolved_arg` +/// above, where the same `=item.` shorthand addresses a real +/// (non-trigger) upstream node and stays a hard reject. +#[tokio::test] +async fn validate_required_arg_resolvability_allows_a_trigger_scoped_null_arg() { + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Webhook" }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "GMAIL_SEND_EMAIL", + "args": { "recipient_email": "a@b.com", "subject": "hi", "body": "=item.email" } } } + ], + "edges": [ { "from_node": "t", "to_node": "post" } ] + })); + let errors = validate_required_arg_resolvability(&g).await; + assert!(errors.is_empty(), "{errors:?}"); +} + +/// The `nodes....` explicit-addressing form of the real B18 bug: an arg +/// wired to a specific upstream (non-trigger) node's output path that never +/// exists there. Unlike the trigger-scoped case above, this stays broken +/// regardless of what the trigger payload looks like at runtime, so it must +/// still hard-reject. +#[tokio::test] +async fn validate_required_arg_resolvability_rejects_an_explicit_nodes_reference() { + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "build_body", "kind": "code", "name": "Build Body", + "config": { "language": "javascript", "source": "return {};" } }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "GMAIL_SEND_EMAIL", + "args": { "recipient_email": "a@b.com", + "subject": "=nodes.build_body.item.subject" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "build_body" }, + { "from_node": "build_body", "to_node": "post" } + ] + })); + let errors = validate_required_arg_resolvability(&g).await; + assert_eq!(errors.len(), 1, "{errors:?}"); + assert!(errors[0].contains("`subject`"), "{}", errors[0]); + assert!(errors[0].contains("nodes.build_body"), "{}", errors[0]); +} + /// (Codex feedback on this PR) `notion` ships a static curated catalog /// (`catalog_for_toolkit`), so at RUNTIME `flow_tool_allowed`'s Path A /// hard-rejects any slug `find_curated` doesn't recognize — even a real, @@ -2993,3 +3124,74 @@ async fn flows_build_hides_the_live_run_tool_from_the_builder_belt() { ); } } + +// ───────────────────────────────────────────────────────────────────────────── +// B23/B24 — condition node branch label must be on `from_port`, not `to_port` +// ───────────────────────────────────────────────────────────────────────────── + +fn condition_graph( + true_from_port: &str, + true_to_port: &str, + false_from_port: &str, + false_to_port: &str, +) -> Value { + json!({ + "name": "condition-routing", + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Trigger" }, + { "id": "gate", "kind": "condition", "name": "Gate", "config": { "field": "has_important" } }, + { "id": "send_summary", "kind": "output_parser", "name": "Send" }, + { "id": "done", "kind": "output_parser", "name": "Done" } + ], + "edges": [ + { "from_node": "t", "from_port": "main", "to_node": "gate", "to_port": "main" }, + { "from_node": "gate", "from_port": true_from_port, "to_node": "send_summary", "to_port": true_to_port }, + { "from_node": "gate", "from_port": false_from_port, "to_node": "done", "to_port": false_to_port } + ] + }) +} + +#[test] +fn validate_and_migrate_graph_rejects_condition_edges_with_branch_label_on_to_port() { + // The exact malformed shape the workflow_builder agent produced live + // (see issue B23): both edges share `from_port: "main"` with the branch + // label on `to_port` instead. The engine routes exclusively on + // `from_port` (B24, `tinyflows::validate`), so this must be a hard + // reject here — never persisted as a silently-broken no-op condition. + let bad_graph = condition_graph("main", "true", "main", "false"); + + let err = validate_and_migrate_graph(bad_graph) + .expect_err("condition edges with the branch label on to_port must be rejected"); + assert!( + err.contains("condition") && err.contains("from_port"), + "expected an InvalidConditionRouting-style error naming from_port, got: {err}" + ); +} + +#[test] +fn validate_and_migrate_graph_accepts_condition_edges_with_branch_label_on_from_port() { + // The correct shape: `from_port` carries "true"/"false", `to_port` stays + // "main". + let good_graph = condition_graph("true", "main", "false", "main"); + + validate_and_migrate_graph(good_graph) + .expect("correctly-routed condition graph (branch label on from_port) must validate"); +} + +#[tokio::test] +async fn flows_create_rejects_condition_edges_with_branch_label_on_to_port() { + // The same hard gate applies at the actual persistence path + // (`flows_create`), not just the standalone validate helper — a graph + // with this shape must never reach the store. + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + let bad_graph = condition_graph("main", "true", "main", "false"); + let err = flows_create(&config, "bad-condition".to_string(), bad_graph, false) + .await + .expect_err("flows_create must reject a condition graph routed on to_port"); + assert!( + err.contains("condition") && err.contains("from_port"), + "expected an InvalidConditionRouting-style error, got: {err}" + ); +} diff --git a/src/openhuman/flows/tools.rs b/src/openhuman/flows/tools.rs index 2272dc433..8928a0d3c 100644 --- a/src/openhuman/flows/tools.rs +++ b/src/openhuman/flows/tools.rs @@ -54,7 +54,11 @@ impl Tool for ProposeWorkflowTool { ONLY VALIDATES the graph and returns a summary — it NEVER creates or enables the flow; \ the user must click \"Save & enable\" in the UI before anything is persisted or can \ run. Build a tinyflows WorkflowGraph: nodes[] ({id, kind, name, config}) + edges[] \ - ({from_node, to_node, from_port?, to_port?}; ports default \"main\"). Exactly ONE \ + ({from_node, to_node, from_port?, to_port?}; ports default \"main\"). For a branching \ + node (condition/switch), the branch label goes on from_port — NEVER on to_port \ + (to_port just stays \"main\"); routing is keyed exclusively on from_port, so a label \ + on to_port instead silently turns the branch into an unconditional fan-out and is a \ + hard reject. Exactly ONE \ trigger node is required. The 12 node kinds: trigger (config.trigger_kind: manual | \ schedule | webhook | app_event | form | chat_message | evaluation | system | \ execute_by_workflow; schedule needs config.schedule = {kind:\"cron\",expr,tz?} | \ @@ -62,7 +66,8 @@ impl Tool for ProposeWorkflowTool { config.trigger_slug), agent (config.prompt), tool_call (config.slug REQUIRED + \ config.args), http_request (config.method/url, optional headers/body), code \ (config.language: \"javascript\"|\"python\" + config.source), condition (config.field; \ - routes ports \"true\"/\"false\"), switch (config.expression or config.field; routes to \ + routes on from_port \"true\"/\"false\", e.g. {from_node:\"gate\",from_port:\"true\",\ + to_node:\"x\",to_port:\"main\"}), switch (config.expression or config.field; routes to \ the matching case port, or \"default\"), transform (config.set: {key: \"=expr\"} \ merged onto each item), split_out (config.path to an array field; fans out one item per \ element), merge (fan-in passthrough, no config), output_parser (passthrough today; no \ @@ -109,8 +114,8 @@ impl Tool for ProposeWorkflowTool { "properties": { "from_node": { "type": "string" }, "to_node": { "type": "string" }, - "from_port": { "type": "string", "description": "Defaults to \"main\"." }, - "to_port": { "type": "string", "description": "Defaults to \"main\"." } + "from_port": { "type": "string", "description": "Defaults to \"main\". For a condition/switch branch, this is where the branch label (e.g. \"true\"/\"false\") goes." }, + "to_port": { "type": "string", "description": "Defaults to \"main\". Almost always stays \"main\" — branch labels go on from_port, not here." } }, "required": ["from_node", "to_node"] } @@ -215,6 +220,26 @@ impl Tool for ProposeWorkflowTool { ))); } + // Required-arg resolvability gate (issue B18): reject outright — + // rather than merely warn — a REQUIRED outbound arg (e.g. + // `GMAIL_SEND_EMAIL.subject`/`.body`) that LOOKS wired but resolves + // to `null` in a sandboxed test run, before the user ever reviews + // the proposal. See `ops::validate_required_arg_resolvability`. + let null_arg_errors = + crate::openhuman::flows::ops::validate_required_arg_resolvability(&graph).await; + if !null_arg_errors.is_empty() { + tracing::debug!( + target: "flows", + %name, + error_count = null_arg_errors.len(), + "[flows] propose_workflow: required-arg resolvability check rejected the graph" + ); + return Ok(ToolResult::error(format!( + "{}\n\nFix these bindings and call propose_workflow again.", + null_arg_errors.join("\n\n") + ))); + } + let summary = build_summary(&graph); // Author-time warnings: unfired trigger kinds + unwired REQUIRED // Composio args (see `ops::graph_wiring_warnings`) — surfaced on the diff --git a/vendor/tinyflows b/vendor/tinyflows index db4165fee..be8ae6571 160000 --- a/vendor/tinyflows +++ b/vendor/tinyflows @@ -1 +1 @@ -Subproject commit db4165feec52aea101432af300878e4011a4d029 +Subproject commit be8ae6571883d9ce2e8f23c11779b2ee04e00ce6