From 48e51f48112266a4a52ff1117e75caf0a87b8a92 Mon Sep 17 00:00:00 2001 From: cyrus Date: Wed, 8 Jul 2026 15:15:25 +0530 Subject: [PATCH] fix(engine,condition): stop starving items after a single-branch condition MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit B15: a condition node wired with only a `true` edge (no `false` edge — the common gate/filter shape) was lowered as an UNCONDITIONAL plain edge in build_graph, so the successor ran on every execution regardless of which port the condition actually emitted. collect_input then port-matched the edge's `true` against the emitted `false` and silently dropped the items, so the successor ran with an EMPTY input and downstream `=item`/`=item.` expressions resolved to null instead of the run simply ending. Fix: when a node's single outgoing edge has from_port != "main", lower it as a conditional edge (mirroring the existing multi-edge branch) whose router returns from_port when the emitted port matches, and END otherwise. A `true`-only condition now behaves as a FILTER: it runs the successor with items on `true`, and terminates the run to END on `false`. Also fixes condition.rs: `field` was read as a raw literal key, so an `=`-expression field (e.g. `field: "=item.assignee"`) was looked up as a literal key named "=item.assignee" and always missed, always routing `false`. Now an expression field is resolved against the node's expression scope first and the resolved value's truthiness is checked directly; non-expression fields keep the existing literal-key-lookup behavior. Tests: condition_single_true_edge_filters_on_false, condition_single_true_edge_item_flows_through (engine.rs), expression_in_field_resolves_and_checks_truthiness (condition.rs). All existing condition/engine tests remain green. --- Cargo.lock | 2 +- src/engine.rs | 158 +++++++++++++++++++++++++--- src/nodes/control_flow/condition.rs | 62 +++++++++-- 3 files changed, 198 insertions(+), 24 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 086f31d..c6b8874 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1355,7 +1355,7 @@ dependencies = [ [[package]] name = "tinyflows" -version = "0.5.0" +version = "0.5.1" dependencies = [ "async-trait", "futures-timer", diff --git a/src/engine.rs b/src/engine.rs index 4677044..6c4d895 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -1066,24 +1066,69 @@ fn build_graph( } HandlerRouting::Plain => { if let [edge] = outgoing.as_slice() { - // Single successor. If the target is a fan-in point (more than - // one predecessor, e.g. a `merge`) it normally gets a waiting - // edge so it runs only once every predecessor completed — the - // merge barrier. But a fan-in whose predecessors are mutually - // exclusive conditional branches (a *conditional join*) must - // not hard-wait on the untaken branch, which never arrives and - // would deadlock the barrier; wire those with plain edges so the - // taken branch fires the join. Everything else is a plain edge. let target = edge.to_node.clone(); - let is_fan_in = incoming_counts - .get(edge.to_node.as_str()) - .copied() - .unwrap_or(0) - > 1; - if is_fan_in && !is_conditional_join(graph, &target) { - builder = builder.add_waiting_edge(node.id.clone(), target); + if edge.from_port != "main" { + // A single outgoing edge whose port is NOT the default + // `main` (e.g. a `condition` wired with only a `true` + // edge, no `false` edge) is not a plain pass-through: the + // node still records which port it emitted on (see + // `items_update`), and `collect_input` port-matches an + // edge's `from_port` against that emitted port before + // handing items to the successor. Wiring this as an + // unconditional `add_edge` (the old behavior) made the + // successor run on *every* execution — including when the + // node emitted the other port — but with an EMPTY input, + // since `collect_input` silently drops the mismatched + // items. Downstream `=item`/`=item.` expressions + // then resolve to `null` instead of the run simply ending. + // + // Fix: lower it as a conditional edge (mirroring the + // multi-edge branch below) that only follows `target` when + // the emitted port matches `edge.from_port`, and falls + // back to `END` otherwise. A `true`-only condition thus + // behaves as a FILTER — it runs the successor (with + // items) on `true`, and terminates the run to `END` on + // `false` — instead of a leaky always-on edge. + let from = node.id.clone(); + let from_port = edge.from_port.clone(); + builder = builder.add_conditional_edges( + node.id.clone(), + move |state: &Value| -> String { + let emitted = state + .get("nodes") + .and_then(|nodes| nodes.get(&from)) + .and_then(|slot| slot.get("port")) + .and_then(Value::as_str) + .unwrap_or("main"); + if emitted == from_port.as_str() { + from_port.clone() + } else { + END.to_string() + } + }, + [(edge.from_port.clone(), target), (END.to_string(), END.to_string())], + ); } else { - builder = builder.add_edge(node.id.clone(), target); + // Single successor on the default `main` port. If the + // target is a fan-in point (more than one predecessor, + // e.g. a `merge`) it normally gets a waiting edge so it + // runs only once every predecessor completed — the merge + // barrier. But a fan-in whose predecessors are mutually + // exclusive conditional branches (a *conditional join*) + // must not hard-wait on the untaken branch, which never + // arrives and would deadlock the barrier; wire those with + // plain edges so the taken branch fires the join. + // Everything else is a plain edge. + let is_fan_in = incoming_counts + .get(edge.to_node.as_str()) + .copied() + .unwrap_or(0) + > 1; + if is_fan_in && !is_conditional_join(graph, &target) { + builder = builder.add_waiting_edge(node.id.clone(), target); + } else { + builder = builder.add_edge(node.id.clone(), target); + } } } else { // Branching: distinct ports (one target each) lower to @@ -2782,6 +2827,87 @@ mod tests { ); } + #[tokio::test] + async fn condition_single_true_edge_filters_on_false() { + // Regression for B15: a `condition` wired with only a `true` edge (no + // `false` edge — the common "gate/filter" shape) used to be lowered as + // an UNCONDITIONAL plain edge, so `sink` ran on *every* execution — + // including when the condition emitted `false` — but with an EMPTY + // input, since `collect_input` port-matches the edge's `true` against + // the emitted `false` and drops the items. It must now act as a FILTER: + // run `sink` (with items) when the branch is taken, and terminate the + // run to END — never running `sink` at all — when it isn't. + let mut condition = node("c", NodeKind::Condition); + condition.config = json!({ "field": "active" }); + let graph = WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger), + condition, + node("sink", NodeKind::OutputParser), + ], + edges: vec![edge("t", "c"), port_edge("c", "true", "sink")], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + let caps = mock_capabilities(); + + let truthy = run(&compiled, json!({ "active": true }), &caps) + .await + .expect("run"); + assert_eq!( + truthy.output["nodes"]["sink"]["items"][0]["json"], + json!({ "active": true }), + "true branch must run sink WITH the item, not an empty input" + ); + + let falsey = run(&compiled, json!({ "active": false }), &caps) + .await + .expect("run"); + assert!( + falsey.output["nodes"]["sink"].is_null(), + "false branch must terminate the run to END without running sink" + ); + } + + #[tokio::test] + async fn condition_single_true_edge_item_flows_through() { + // trigger -> split_out (per-item fan-out) -> condition(field="assignee", + // true-only edge) -> sink. Proves the B15 fix end-to-end through a + // realistic shape: a downstream node's `=item.` must see the real + // item, not null, when the guarding condition took the `true` branch. + let mut condition = node("c", NodeKind::Condition); + condition.config = json!({ "field": "assignee" }); + let graph = WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger), + node("s", NodeKind::SplitOut), + condition, + node("sink", NodeKind::OutputParser), + ], + edges: vec![edge("t", "s"), edge("s", "c"), port_edge("c", "true", "sink")], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + let caps = mock_capabilities(); + + // Truthy assignee: the item must flow all the way through to `sink`. + let assigned = run(&compiled, json!({ "assignee": "alice" }), &caps) + .await + .expect("run"); + assert_eq!( + assigned.output["nodes"]["sink"]["items"][0]["json"]["assignee"], + json!("alice"), + "true branch must carry the real item through — not starve it to null" + ); + + // Missing assignee (falsey): `sink` must not run at all. + let unassigned = run(&compiled, json!({}), &caps).await.expect("run"); + assert!( + unassigned.output["nodes"]["sink"].is_null(), + "false branch must not execute the guarded successor" + ); + } + #[tokio::test] async fn switch_field_matching_case_routes_there() { // switch(field=kind) with input kind="a" routes only to the `a` case; the diff --git a/src/nodes/control_flow/condition.rs b/src/nodes/control_flow/condition.rs index 8ee2fff..417c232 100644 --- a/src/nodes/control_flow/condition.rs +++ b/src/nodes/control_flow/condition.rs @@ -1,9 +1,11 @@ //! The `condition` node: a two-way IF branch. use async_trait::async_trait; +use serde_json::Value; use crate::error::Result; -use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; +use crate::expr; +use crate::nodes::{NodeContext, NodeExecutor, NodeOutput, expr_scope}; /// Two-way conditional branch, emitting on the `true` or `false` port. #[derive(Debug, Default, Clone)] @@ -12,14 +14,26 @@ pub struct ConditionNode; #[async_trait] impl NodeExecutor for ConditionNode { async fn execute(&self, ctx: NodeContext<'_>) -> Result { - let field = ctx - .node - .config - .get("field") - .and_then(serde_json::Value::as_str); + let field = ctx.node.config.get("field").and_then(Value::as_str); + // `field` supports two shapes: + // - a plain string (e.g. `"assignee"`) — a literal key looked up on the + // item's JSON, as before (backward-compatible); + // - an `=`-expression (e.g. `"=item.assignee"`) — resolved against the + // node's expression scope (`item`/`items`/`run`/`nodes`) first, and + // the *resolved value itself* is what gets truthiness-checked. Without + // this, an expression field was read as a raw literal key + // (`item.json.get("=item.assignee")`), which never matches any real + // key and always routes `false`. + let resolved_expr_value = field.filter(|f| expr::is_expression(f)).map(|f| { + let scope = expr_scope(&ctx); + expr::evaluate(&Value::String(f.to_string()), &scope) + }); let truthy = ctx.input.first().is_some_and(|item| { + if let Some(value) = &resolved_expr_value { + return is_truthy(value); + } let value = match field { - Some(f) => item.json.get(f).unwrap_or(&serde_json::Value::Null), + Some(f) => item.json.get(f).unwrap_or(&Value::Null), None => &item.json, }; is_truthy(value) @@ -186,4 +200,38 @@ mod tests { assert_eq!(port, "true", "first item decides the branch"); assert_eq!(items, input, "all input items are routed through unchanged"); } + + #[tokio::test] + async fn expression_in_field_resolves_and_checks_truthiness() { + // `field: "=item.assignee"` must be resolved against the expression + // scope and the RESOLVED VALUE's truthiness checked — not a literal key + // lookup for a key named `"=item.assignee"` (which would always be + // absent and always route `false`, regardless of the real assignee). + let (truthy_port, _) = route( + json!({ "field": "=item.assignee" }), + vec![Item::new(json!({ "assignee": "alice" }))], + ) + .await; + assert_eq!( + truthy_port, "true", + "a non-empty assignee value must route true" + ); + + let (empty_port, _) = route( + json!({ "field": "=item.assignee" }), + vec![Item::new(json!({ "assignee": "" }))], + ) + .await; + assert_eq!(empty_port, "false", "an empty assignee value must route false"); + + let (missing_port, _) = route( + json!({ "field": "=item.assignee" }), + vec![Item::new(json!({}))], + ) + .await; + assert_eq!( + missing_port, "false", + "a missing assignee resolves to null via the expression and routes false" + ); + } }