diff --git a/src/openhuman/flows/builder_tools.rs b/src/openhuman/flows/builder_tools.rs index 108c8dd097..22c6ba861e 100644 --- a/src/openhuman/flows/builder_tools.rs +++ b/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, config: Arc, @@ -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 = 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 { + 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 diff --git a/src/openhuman/flows/builder_tools_tests.rs b/src/openhuman/flows/builder_tools_tests.rs index 1be8b6d9ba..6751607715 100644 --- a/src/openhuman/flows/builder_tools_tests.rs +++ b/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 diff --git a/vendor/tinyflows b/vendor/tinyflows index eb4865d1a1..56d81dfdad 160000 --- a/vendor/tinyflows +++ b/vendor/tinyflows @@ -1 +1 @@ -Subproject commit eb4865d1a16679b03fe79858c7b656adc34baa98 +Subproject commit 56d81dfdadc1454189322d744e050ae4f7e525f9