fix(flows): stop condition nodes starving downstream =item + dry-run routing-divergence warning (B15) (#4698)

src/openhuman/flows/builder_tools.rs

@@ -964,6 +964,24 @@ impl Tool for ListAgentProfilesTool {
 /// (`{ node_id, error }`, the error text read back out of the run's `output`
 /// state — see [`tool_call_error_message`]) and fails the dry run the same as
 /// a null resolution.
+///
+/// **Routing-divergence warning (B15's dry-run blind spot):** none of the
+/// checks above see a node that never ran at all. An `agent`/`tool_call` node
+/// downstream of a `condition` can be silently unexercised because the
+/// sandbox's mock trigger payload has a different *shape* than a real
+/// trigger's (e.g. a webhook's real JSON body vs. the dry run's `{}`
+/// default), so the condition takes a different branch under mock data than
+/// it would at runtime — a graph can dry-run `ok: true` while its most
+/// data-dependent node was never actually checked. After the run settles,
+/// every `agent`/`tool_call` node with no [`ExecutionStep`] in the
+/// [`CapturingObserver`] is collected into `routing_divergence_warnings`
+/// (`{ node_id, condition_node_id, message }`, `condition_node_id` naming the
+/// nearest upstream `condition` node found by walking predecessors — see
+/// [`find_upstream_condition`] — or `null` if none is found). This is a
+/// **warning, not a hard reject**: it never flips `ok` to `false` by itself
+/// (an unexercised branch can be entirely intentional), and is surfaced on
+/// both the `ok: true` and `ok: false` result shapes so the caller can
+/// double-check that node's wiring by hand.
 pub struct DryRunWorkflowTool {
     security: Arc<SecurityPolicy>,
     config: Arc<Config>,
@@ -1254,6 +1272,57 @@ impl Tool for DryRunWorkflowTool {
             })
             .collect();
 
+        // Routing-divergence blind spot (B15): an `agent`/`tool_call` node that
+        // did NOT execute during the sandbox run at all — because an upstream
+        // `condition` routed the mock trigger payload onto its OTHER branch —
+        // is invisible to every check above (`null_resolutions` etc. only
+        // inspect steps that ran). But the mock input's *shape* need not match
+        // a real trigger's shape (a webhook's real JSON vs. the dry run's `{}`
+        // default, say), so a condition that took the `false` branch under mock
+        // data may well take `true` at runtime with real data — or vice versa.
+        // Either way, the dry run silently never exercised the very node whose
+        // wiring most needed checking. This is a WARNING, not a hard reject
+        // (an unexercised branch can be entirely intentional), surfaced
+        // alongside the other diagnostics so the caller can double-check the
+        // wiring by hand.
+        let executed_steps = observer.steps();
+        let executed_node_ids: std::collections::HashSet<&str> = executed_steps
+            .iter()
+            .map(|step| step.node_id.as_str())
+            .collect();
+        let routing_divergence_warnings: Vec<Value> = graph
+            .nodes
+            .iter()
+            .filter(|node| {
+                node.kind != tinyflows::model::NodeKind::Trigger
+                    && (agent_node_ids.contains(node.id.as_str())
+                        || tool_call_node_ids.contains(node.id.as_str()))
+                    && !executed_node_ids.contains(node.id.as_str())
+            })
+            .map(|node| {
+                let condition_node_id = find_upstream_condition(&graph, &node.id);
+                let message = match &condition_node_id {
+                    Some(cid) => format!(
+                        "Node '{}' did not execute in the dry run (condition '{}' routed to \
+                         the other branch under mock data); verify the wiring — at runtime \
+                         with real data it may route differently.",
+                        node.id, cid
+                    ),
+                    None => format!(
+                        "Node '{}' did not execute in the dry run (an upstream branch routed \
+                         the mock data away from it); verify the wiring — at runtime with real \
+                         data it may route differently.",
+                        node.id
+                    ),
+                };
+                json!({
+                    "node_id": node.id,
+                    "condition_node_id": condition_node_id,
+                    "message": message,
+                })
+            })
+            .collect();
+
         tracing::info!(
             target: "flows",
             node_count = graph.nodes.len(),
@@ -1262,6 +1331,7 @@ impl Tool for DryRunWorkflowTool {
             agent_prompt_null_count = agent_prompt_nulls.len(),
             agent_input_context_null_count = agent_input_context_nulls.len(),
             node_error_count = node_errors.len(),
+            routing_divergence_warning_count = routing_divergence_warnings.len(),
             "[flows] dry_run_workflow: sandbox run finished"
         );
 
@@ -1286,6 +1356,7 @@ impl Tool for DryRunWorkflowTool {
                 "agent_prompt_nulls": agent_prompt_nulls,
                 "agent_input_context_nulls": agent_input_context_nulls,
                 "node_errors": node_errors,
+                "routing_divergence_warnings": routing_divergence_warnings,
                 "message": "These tool_call args resolved to null, an agent node's prompt or \
                     input_context resolved to null (an EMPTY prompt — see agent_prompt_nulls — \
                     or no upstream data at all — see agent_input_context_nulls), or a tool_call \
@@ -1294,7 +1365,10 @@ impl Tool for DryRunWorkflowTool {
                     output (give any agent node an output_parser.schema so its fields are \
                     addressable), feed upstream data into a null-resolved agent prompt/ \
                     input_context from a real upstream field instead of a jq expression inside \
-                    the prompt text, and fix or rewire whatever tool_call node_errors names.",
+                    the prompt text, and fix or rewire whatever tool_call node_errors names. Also \
+                    check routing_divergence_warnings: any agent/tool_call node listed there \
+                    never ran in this sandbox at all because an upstream condition routed the \
+                    mock data past it — verify that wiring by hand too.",
             }))?));
         }
 
@@ -1307,11 +1381,46 @@ impl Tool for DryRunWorkflowTool {
             "agent_prompt_nulls": agent_prompt_nulls,
             "agent_input_context_nulls": agent_input_context_nulls,
             "node_errors": node_errors,
-            "note": "SANDBOX (mock) output — LLM/tool/HTTP/code nodes returned deterministic echoes; NO real side effects occurred. This checks wiring/routing only, not whether real integrations work.",
+            "routing_divergence_warnings": routing_divergence_warnings,
+            "note": "SANDBOX (mock) output — LLM/tool/HTTP/code nodes returned deterministic echoes; NO real side effects occurred. This checks wiring/routing only, not whether real integrations work. \
+                If routing_divergence_warnings is non-empty, an agent/tool_call node never ran in \
+                this sandbox because an upstream condition routed the mock data past it — that \
+                node's wiring is unverified; check it by hand.",
         }))?))
     }
 }
 
+/// Walks a graph backward from `node_id`'s predecessors (any number of hops)
+/// to find the nearest ancestor that is a `condition` node — used to name the
+/// branch responsible for a routing-divergence warning (see
+/// [`DryRunWorkflowTool::execute`]'s routing-divergence check, just above).
+/// Returns `None` if no predecessor chain reaches a `condition` node (e.g. the
+/// node simply has no predecessors, or none of them is a condition) — the
+/// warning is still emitted, just without a named culprit node.
+fn find_upstream_condition(graph: &WorkflowGraph, node_id: &str) -> Option<String> {
+    let mut visited: std::collections::HashSet<&str> = std::collections::HashSet::new();
+    let mut queue: std::collections::VecDeque<&str> = graph
+        .edges
+        .iter()
+        .filter(|edge| edge.to_node == node_id)
+        .map(|edge| edge.from_node.as_str())
+        .collect();
+    while let Some(current) = queue.pop_front() {
+        if !visited.insert(current) {
+            continue;
+        }
+        if let Some(node) = graph.nodes.iter().find(|n| n.id == current) {
+            if node.kind == tinyflows::model::NodeKind::Condition {
+                return Some(node.id.clone());
+            }
+        }
+        for edge in graph.edges.iter().filter(|edge| edge.to_node == current) {
+            queue.push_back(edge.from_node.as_str());
+        }
+    }
+    None
+}
+
 /// Best-effort extraction of the human-readable error message the engine
 /// recorded for a `tool_call` node whose `on_error` policy is `"continue"` or
 /// `"route"`. Such a node's failure is converted into an error ITEM on its

src/openhuman/flows/builder_tools_tests.rs

@@ -951,6 +951,92 @@ async fn dry_run_passes_when_agent_uses_input_context_instead_of_prompt_expressi
     );
 }
 
+#[tokio::test]
+async fn dry_run_warns_on_unexercised_agent_after_condition() {
+    // B15's dry-run blind spot: `gate` is a `condition` wired with only a
+    // `true` edge to `classify`. The dry run's default trigger input is `{}`
+    // (no `input` param passed), so `gate`'s configured field ("active") is
+    // absent — falsey — and the condition emits `false`. Since `false` has no
+    // outgoing edge, `classify` never executes at all: not a null resolution,
+    // not a node error, just silently unexercised. A real trigger's payload
+    // could easily carry `active: true` and take the other branch, so the
+    // dry run must still surface this as a warning even though `ok` stays
+    // `true` — there's nothing here that flips it to a hard reject.
+    let tool = DryRunWorkflowTool::new(
+        policy(AutonomyLevel::Supervised),
+        test_config(&TempDir::new().unwrap()),
+    );
+    let graph = json!({
+        "nodes": [
+            { "id": "t", "kind": "trigger", "name": "Manual" },
+            { "id": "gate", "kind": "condition", "name": "Gate",
+              "config": { "field": "active" } },
+            { "id": "classify", "kind": "agent", "name": "Classify",
+              "config": { "prompt": "Classify the item.", "input_context": "=item" } }
+        ],
+        "edges": [
+            { "from_node": "t", "to_node": "gate" },
+            { "from_node": "gate", "from_port": "true", "to_node": "classify" }
+        ]
+    });
+
+    let result = tool.execute(json!({ "graph": graph })).await.unwrap();
+    assert!(!result.is_error, "{}", result.output());
+    let parsed: Value = serde_json::from_str(&result.output()).unwrap();
+    assert_eq!(
+        parsed["ok"], true,
+        "an unexercised branch is a warning, not a hard reject: {parsed}"
+    );
+    let warnings = parsed["routing_divergence_warnings"]
+        .as_array()
+        .expect("routing_divergence_warnings array");
+    assert_eq!(warnings.len(), 1, "{parsed}");
+    assert_eq!(warnings[0]["node_id"], "classify");
+    assert_eq!(warnings[0]["condition_node_id"], "gate");
+    assert!(
+        warnings[0]["message"]
+            .as_str()
+            .unwrap()
+            .contains("classify"),
+        "{parsed}"
+    );
+}
+
+#[tokio::test]
+async fn dry_run_no_routing_divergence_warning_when_every_node_executes() {
+    // FALSE-POSITIVE-PREVENTION: a condition whose taken branch under the
+    // default mock input DOES reach the downstream agent must not warn.
+    let tool = DryRunWorkflowTool::new(
+        policy(AutonomyLevel::Supervised),
+        test_config(&TempDir::new().unwrap()),
+    );
+    let graph = json!({
+        "nodes": [
+            { "id": "t", "kind": "trigger", "name": "Manual" },
+            { "id": "gate", "kind": "condition", "name": "Gate",
+              "config": { "field": "active" } },
+            { "id": "classify", "kind": "agent", "name": "Classify",
+              "config": { "prompt": "Classify the item.", "input_context": "=item" } }
+        ],
+        "edges": [
+            { "from_node": "t", "to_node": "gate" },
+            { "from_node": "gate", "from_port": "false", "to_node": "classify" }
+        ]
+    });
+
+    let result = tool.execute(json!({ "graph": graph })).await.unwrap();
+    assert!(!result.is_error, "{}", result.output());
+    let parsed: Value = serde_json::from_str(&result.output()).unwrap();
+    assert_eq!(parsed["ok"], true, "{parsed}");
+    assert!(
+        parsed["routing_divergence_warnings"]
+            .as_array()
+            .unwrap()
+            .is_empty(),
+        "{parsed}"
+    );
+}
+
 /// (systemic tool-contract fix, Part 2b) A missing required Composio arg is
 /// now a HARD REJECT at `revise_workflow` — `validate_tool_contracts` runs
 /// ahead of the older advisory `graph_wiring_warnings` check and catches the