fix(engine,condition): stop starving a downstream node after a single-branch condition

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.<field>` 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.<field>` 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

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<NodeOutput> {
-        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"
+        );
+    }
 }