From e5327de9f602c1fbbf72d45150729f45b91fa3a8 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 15 Jul 2026 09:48:31 +0300 Subject: [PATCH 1/4] =?UTF-8?q?feat(flows):=20agent-friendly=20edit=5Fwork?= =?UTF-8?q?flow=20op=20errors=20=E2=80=94=20aliases,=20per-op=20diagnostic?= =?UTF-8?q?s,=20ordering=20hint?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/graph_ops.rs | 98 +++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 92 insertions(+), 6 deletions(-) diff --git a/src/graph_ops.rs b/src/graph_ops.rs index 8c31af8..8f1fe32 100644 --- a/src/graph_ops.rs +++ b/src/graph_ops.rs @@ -40,14 +40,16 @@ pub enum GraphOp { /// `config` are recursively merged onto the node's existing config, and a /// `null` value deletes that key. Fails if no node has `id`. UpdateNodeConfig { - /// The target node id. + /// The target node id. Accepts the alias `node_id`. + #[serde(alias = "node_id")] id: NodeId, /// The partial config to merge (a `null` leaf deletes the key). config: Value, }, /// Replace a node's human-readable `name`. Fails if no node has `id`. SetNodeName { - /// The target node id. + /// The target node id. Accepts the alias `node_id`. + #[serde(alias = "node_id")] id: NodeId, /// The new display name. name: String, @@ -59,14 +61,17 @@ pub enum GraphOp { /// node that others bind to should re-point those bindings itself. Fails if /// `new_id` is empty or already in use, or if no node has `id`. RenameNode { - /// The current node id. + /// The current node id. Accepts the alias `node_id`. + #[serde(alias = "node_id")] id: NodeId, - /// The new node id. + /// The new node id. Accepts the alias `new_node_id`. + #[serde(alias = "new_node_id")] new_id: NodeId, }, /// Remove a node and every edge incident on it. Fails if no node has `id`. RemoveNode { - /// The node id to remove. + /// The node id to remove. Accepts the alias `node_id`. + #[serde(alias = "node_id")] id: NodeId, }, /// Add a directed edge. Fails if either endpoint node is missing or the @@ -91,7 +96,8 @@ pub enum GraphOp { }, /// Set (or move) a node's canvas position. Fails if no node has `id`. SetNodePosition { - /// The target node id. + /// The target node id. Accepts the alias `node_id`. + #[serde(alias = "node_id")] id: NodeId, /// The new canvas position. position: Position, @@ -649,4 +655,84 @@ mod tests { _ => panic!("wrong variant"), } } + + #[test] + fn id_fields_accept_the_node_id_alias() { + // remove_node { node_id } is a natural guess and must round-trip to `id`. + let op: GraphOp = serde_json::from_value(json!({ + "op": "remove_node", "node_id": "trigger" + })) + .unwrap(); + assert_eq!( + op, + GraphOp::RemoveNode { + id: "trigger".into() + } + ); + + // update_node_config { node_id, config } + let op: GraphOp = serde_json::from_value(json!({ + "op": "update_node_config", "node_id": "a", "config": { "prompt": "hi" } + })) + .unwrap(); + assert_eq!( + op, + GraphOp::UpdateNodeConfig { + id: "a".into(), + config: json!({ "prompt": "hi" }), + } + ); + + // set_node_name { node_id, name } + let op: GraphOp = serde_json::from_value(json!({ + "op": "set_node_name", "node_id": "a", "name": "Renamed" + })) + .unwrap(); + assert_eq!( + op, + GraphOp::SetNodeName { + id: "a".into(), + name: "Renamed".into(), + } + ); + + // set_node_position { node_id, position } + let op: GraphOp = serde_json::from_value(json!({ + "op": "set_node_position", "node_id": "a", "position": { "x": 1.0, "y": 2.0 } + })) + .unwrap(); + assert_eq!( + op, + GraphOp::SetNodePosition { + id: "a".into(), + position: Position { x: 1.0, y: 2.0 }, + } + ); + + // rename_node accepts node_id + new_node_id aliases together. + let op: GraphOp = serde_json::from_value(json!({ + "op": "rename_node", "node_id": "a", "new_node_id": "agent1" + })) + .unwrap(); + assert_eq!( + op, + GraphOp::RenameNode { + id: "a".into(), + new_id: "agent1".into(), + } + ); + + // Canonical `id` / `new_id` still work unchanged. + let op: GraphOp = serde_json::from_value(json!({ + "op": "rename_node", "id": "a", "new_id": "agent1" + })) + .unwrap(); + assert_eq!( + op, + GraphOp::RenameNode { + id: "a".into(), + new_id: "agent1".into(), + } + ); + } } From 93dbebb605cf2c2d103bdac51f56e43572a0c2a0 Mon Sep 17 00:00:00 2001 From: cyrus Date: Thu, 16 Jul 2026 22:15:17 +0530 Subject: [PATCH 2/4] =?UTF-8?q?fix(engine):=20fix=20mixed=20fan-in=20barri?= =?UTF-8?q?er=20deadlock=20via=20BarrierRelief=20(Eng=20=C2=A75)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit is_conditional_join made a binary all-plain-or-all-waiting decision for fan-in nodes: a merge was only treated as a "conditional join" (plain edges) when EVERY predecessor was reachable from a single common brancher's distinct ports. A MIXED fan-in — one unconditionally-reachable predecessor plus one reachable only via a conditional branch — fell through to the all-waiting barrier, so the untaken branch's predecessor never arrived and the merge deadlocked forever. The naive fix (lower the conditional predecessor's edge as plain) causes silent data loss instead: when that branch IS taken, the merge fires off the superstep snapshot before the conditional predecessor's items commit. Fix: every fan-in edge is now lowered as a waiting edge unconditionally (removes is_conditional_join and the plain/waiting branch). Two new helpers classify predecessors instead of the join as a whole: - has_unconditional_path: reachable from the trigger via "main"-port edges only (so it always runs, whether that's a plain pass-through or a parallel fan-out). - conditional_predecessors: a fan-in node's direct predecessors that are NOT unconditionally reachable — i.e. sit behind a branch port. For each conditional predecessor, find_conditional_brancher locates the brancher whose port leads to it, and the lowering registers tinyagents::GraphBuilder::add_barrier_relief(brancher, predecessor, merge) so the barrier can clear on its remaining predecessors when that branch goes untaken, without ever weakening it into a plain edge. Adds tests/eng5_barrier_relief_e2e.rs: the mixed fan-in repro (both untaken- and taken-branch directions, the latter guarding against data loss), plus pure-conditional-join and pure-unconditional-fan-in regressions. Depends on tinyagents' new add_barrier_relief primitive (companion change in the tinyagents submodule/crate, not yet published — this change is not yet buildable against a released tinyagents version). --- src/engine.rs | 182 +++++++++++++++------ tests/eng5_barrier_relief_e2e.rs | 272 +++++++++++++++++++++++++++++++ 2 files changed, 405 insertions(+), 49 deletions(-) create mode 100644 tests/eng5_barrier_relief_e2e.rs diff --git a/src/engine.rs b/src/engine.rs index 6cdd5ef..e494329 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -268,57 +268,113 @@ fn handler_routing(graph: &crate::model::WorkflowGraph, node_id: &str) -> Handle } } -/// Whether the fan-in node `merge_id` is a **conditional join**: every one of its -/// predecessors sits on a distinct port of a common upstream brancher, so at most -/// one predecessor ever runs. Such a join must not hard-wait on all predecessors -/// (a waiting-edge barrier would deadlock on the untaken branch) — it fires when -/// the taken branch arrives. +/// Whether `to` is reachable from `from` by following only edges on the +/// default `"main"` port, without ever expanding through `stop`. /// -/// Detected conservatively: there must be a brancher `B` (a node whose outgoing -/// edges use >=2 distinct ports) such that each predecessor is reachable from -/// **exactly one** of `B`'s ports, and the predecessor→port mapping is injective -/// (no two predecessors share a port). When detection is unsure it returns -/// `false`, preserving the safe waiting-edge barrier. Reachability is measured -/// forward from each of `B`'s ports without passing through `merge_id`, so the -/// join point itself never counts as a shared reconvergence. -fn is_conditional_join(graph: &crate::model::WorkflowGraph, merge_id: &str) -> bool { - let preds: Vec<&str> = graph +/// This is the "always runs" reachability test: a node reached this way runs +/// on *every* execution (plain pass-throughs and parallel fan-outs alike, +/// since a `FanOut`/`PortCommand` node's declared successors all still sit on +/// a single shared port in `graph.edges`), whereas a node only reachable by +/// stepping onto a non-`main` port (a `true`/`false`/custom branch port) may +/// legitimately never run in a given execution — its brancher chose a +/// different port. `stop` mirrors [`reaches_via_port`]'s parameter: it keeps +/// the search from walking past a fan-in join point. +fn has_unconditional_path( + graph: &crate::model::WorkflowGraph, + from: &str, + to: &str, + stop: &str, +) -> bool { + if from == to { + return true; + } + let mut stack: Vec<&str> = graph .edges .iter() - .filter(|e| e.to_node == merge_id) - .map(|e| e.from_node.as_str()) + .filter(|e| e.from_node == from && e.from_port == "main") + .map(|e| e.to_node.as_str()) .collect(); - if preds.len() < 2 { - return false; + let mut seen: std::collections::HashSet<&str> = std::collections::HashSet::new(); + while let Some(node) = stack.pop() { + if node == to { + return true; + } + if node == stop || !seen.insert(node) { + continue; + } + stack.extend( + graph + .edges + .iter() + .filter(|e| e.from_node == node && e.from_port == "main") + .map(|e| e.to_node.as_str()), + ); } + false +} + +/// The direct predecessors of the fan-in node `merge_id` that are **not** +/// reachable from the workflow's trigger via an unconditional (`"main"`-only) +/// path — i.e. predecessors that sit behind a conditional/branch port and so +/// may legitimately never run in a given execution. +/// +/// Every incoming edge of a fan-in node is lowered as a waiting edge (see the +/// edge-lowering loop below), so a fan-in with any conditional predecessor +/// needs a [`GraphBuilder::add_barrier_relief`] registration for it or its +/// barrier would wait forever on the branch that was never taken. A fan-in +/// whose predecessors are all unconditionally reachable (a plain diamond or a +/// parallel fan-out join) returns an empty set — those still hard-wait on +/// every predecessor, which is correct because every predecessor always runs. +/// +/// Returns an empty set when the graph has no unique trigger (defensive; a +/// validated workflow always has exactly one). +fn conditional_predecessors( + graph: &crate::model::WorkflowGraph, + merge_id: &str, +) -> std::collections::HashSet { + let Some(trigger) = graph.trigger() else { + return std::collections::HashSet::new(); + }; + graph + .edges + .iter() + .filter(|e| e.to_node == merge_id) + .map(|e| e.from_node.clone()) + .filter(|pred| !has_unconditional_path(graph, &trigger.id, pred, merge_id)) + .collect() +} + +/// Finds a brancher node `B` and one of its (>=2) outgoing ports whose +/// routing decides whether `predecessor` runs, so a +/// [`GraphBuilder::add_barrier_relief`] can be registered against `B`'s own +/// completion — the exact activation that decides `predecessor`'s fate. +/// +/// A node counts as a brancher only when it has two or more distinct +/// outgoing ports (mirrors the old `is_conditional_join`'s brancher +/// detection): a same-port `FanOut`/`PortCommand` node's successors all run +/// together and are never conditional. Returns the first match in graph node +/// order; when no brancher is found (defensive — should not happen for a +/// `conditional_predecessors` result on a validated graph) the caller skips +/// registering a relief, preserving the safe (if potentially +/// over-cautious/deadlocking) waiting-edge barrier rather than guessing. +fn find_conditional_brancher( + graph: &crate::model::WorkflowGraph, + predecessor: &str, + stop: &str, +) -> Option { for brancher in &graph.nodes { - let ports: Vec = outgoing_by_port(graph, &brancher.id) - .into_iter() - .map(|(port, _)| port) - .collect(); + let ports = outgoing_by_port(graph, &brancher.id); if ports.len() < 2 { continue; } - let mut used_ports: std::collections::HashSet<&str> = std::collections::HashSet::new(); - let mut ok = true; - for pred in &preds { - let reaching: Vec<&str> = ports - .iter() - .filter(|port| reaches_via_port(graph, &brancher.id, port, pred, merge_id)) - .map(String::as_str) - .collect(); - // A predecessor must be reachable from exactly one of the brancher's - // ports, and that port must be unique to it (injective mapping). - if reaching.len() != 1 || !used_ports.insert(reaching[0]) { - ok = false; - break; - } - } - if ok && used_ports.len() >= 2 { - return true; + if ports + .iter() + .any(|(port, _)| reaches_via_port(graph, &brancher.id, port, predecessor, stop)) + { + return Some(brancher.id.clone()); } } - false + None } /// Whether `target` is reachable from `brancher`'s `port` successors, walking @@ -1118,20 +1174,27 @@ fn build_graph( } else { // 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. + // e.g. a `merge`) it gets a waiting edge so it runs only + // once every predecessor completed — the merge barrier. + // + // Every fan-in edge is wired as waiting unconditionally, + // even when some predecessors are mutually exclusive + // conditional branches (a *conditional join*): lowering a + // conditional predecessor's edge as *plain* instead would + // let a *taken* branch's downstream node race past the + // barrier — the join can fire off the superstep snapshot + // *before* the conditional branch's items are committed, + // silently dropping them. Any fan-in with a conditional + // predecessor instead gets a barrier-relief registration + // (see below) so the all-waiting barrier still clears when + // that predecessor's branch is never taken, without ever + // weakening the barrier itself. 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) { + if is_fan_in { builder = builder.add_waiting_edge(node.id.clone(), target); } else { builder = builder.add_edge(node.id.clone(), target); @@ -1164,6 +1227,27 @@ fn build_graph( } } + // Mixed fan-in barrier relief. Every fan-in node above was wired with + // all-waiting edges, even when one or more of its predecessors sit behind + // a conditional branch that may never be taken — hard-waiting on all of + // them is the only way to avoid the taken-branch data race described + // above, but it means a *mixed* fan-in (one unconditionally-reachable + // predecessor plus one reachable only via a conditional branch) would + // deadlock forever on the branch that was never taken. For each such + // conditional predecessor, register a relief against the brancher whose + // port decides its fate, so the barrier can still clear on its remaining + // predecessors when that branch goes untaken. + for merge in &graph.nodes { + if incoming_counts.get(merge.id.as_str()).copied().unwrap_or(0) <= 1 { + continue; + } + for predecessor in conditional_predecessors(graph, &merge.id) { + if let Some(brancher) = find_conditional_brancher(graph, &predecessor, &merge.id) { + builder = builder.add_barrier_relief(brancher, predecessor, merge.id.clone()); + } + } + } + // A checkpointer (plus a thread id on the run) is required for tinyagents to // persist the interrupt boundary and hand pending approvals back to us. The // checkpointer is host-injected: the default entry points supply an diff --git a/tests/eng5_barrier_relief_e2e.rs b/tests/eng5_barrier_relief_e2e.rs new file mode 100644 index 0000000..ca59b9e --- /dev/null +++ b/tests/eng5_barrier_relief_e2e.rs @@ -0,0 +1,272 @@ +#![cfg(feature = "mock")] +//! Eng §5 regression: the mixed fan-in barrier deadlock and its +//! `BarrierRelief`-based fix. +//! +//! A fan-in (`merge`) node with more than one predecessor is lowered as an +//! all-waiting barrier (see `src/engine.rs`'s edge-lowering loop) so a +//! downstream merge never fires off a stale superstep snapshot before a +//! *taken* conditional branch's items have committed (the data-loss failure +//! mode a plain edge would reintroduce). But when one predecessor is only +//! reachable via a conditional branch that this run does *not* take, that +//! predecessor never arrives on its own — without a relief, the barrier +//! would wait for it forever. +//! +//! These tests build the mixed fan-in repro graph directly (one +//! unconditionally-reachable predecessor `c`, one conditional-only +//! predecessor `a`) and assert both directions: the untaken-branch case must +//! not deadlock (guarded by a `tokio::time::timeout` so a regression fails +//! fast instead of hanging the suite), and the taken-branch case must not +//! lose `a`'s item. Two more tests pin the pure-conditional-join and +//! pure-unconditional-fan-in cases the old `is_conditional_join` special-cased, +//! now handled uniformly by `conditional_predecessors` + `BarrierRelief`. +//! +//! Gated behind the `mock` cargo feature, so plain `cargo test` skips it while +//! `cargo test --all-features` runs it (mirrors `branching_e2e.rs` / +//! `hardening_branching_e2e.rs`). + +use std::collections::HashSet; +use std::time::Duration; + +use serde_json::{Value, json}; +use tinyflows::caps::mock::mock_capabilities; +use tinyflows::compiler::compile; +use tinyflows::engine::run; +use tinyflows::model::{Edge, Node, NodeKind, TriggerKind, WorkflowGraph}; + +fn node(id: &str, kind: NodeKind, config: Value) -> Node { + Node { + id: id.to_string(), + kind, + type_version: 1, + name: id.to_string(), + config, + ports: vec![], + position: None, + } +} + +fn trigger(id: &str) -> Node { + node( + id, + NodeKind::Trigger, + json!({ "kind": TriggerKind::Manual }), + ) +} + +/// Edge from `from_node`'s `from_port` into `to_node`'s `"main"` port. +fn edge(from_node: &str, from_port: &str, to_node: &str) -> Edge { + Edge { + from_node: from_node.to_string(), + from_port: from_port.to_string(), + to_node: to_node.to_string(), + to_port: "main".to_string(), + } +} + +/// A `transform` node that passes its input through, stamped with `tag` so a +/// downstream `merge`'s items can be traced back to which node produced them. +fn tagged(id: &str, tag: &str) -> Node { + node(id, NodeKind::Transform, json!({ "set": { "tag": tag } })) +} + +/// The Eng §5 repro graph: `start` fans out to `cond` and `c` (both +/// unconditionally reachable — a parallel fan-out, not a branch). `cond` +/// routes on `flag`: `true -> a`, `false -> b`. `a` and `c` both feed the +/// merge `m`; `b` is a dead end (never wired into `m`). +/// +/// `c` is `m`'s one unconditionally-reachable predecessor (reachable from +/// `start` via a plain `"main"`-port fan-out); `a` is reachable only through +/// `cond`'s `true` port, so it is `m`'s one conditional predecessor. +fn mixed_fan_in_graph() -> WorkflowGraph { + WorkflowGraph { + name: "eng5_mixed_fan_in".to_string(), + nodes: vec![ + trigger("start"), + node("cond", NodeKind::Condition, json!({ "field": "flag" })), + tagged("a", "a"), + tagged("b", "b"), + tagged("c", "c"), + node("m", NodeKind::Merge, Value::Null), + ], + edges: vec![ + edge("start", "main", "cond"), + edge("start", "main", "c"), + edge("cond", "true", "a"), + edge("cond", "false", "b"), + edge("a", "main", "m"), + edge("c", "main", "m"), + ], + ..Default::default() + } +} + +fn merge_tags(output: &Value, node_id: &str) -> HashSet { + output["nodes"][node_id]["items"] + .as_array() + .unwrap_or_else(|| panic!("{node_id} emitted no items array: {output}")) + .iter() + .map(|item| { + item["json"]["tag"] + .as_str() + .unwrap_or_else(|| panic!("merged item missing `tag`: {item}")) + .to_string() + }) + .collect() +} + +/// flag:false — `cond` takes the `false` branch (to `b`, a dead end never +/// wired into `m`), so `a` never runs. Pre-fix, `m`'s all-waiting barrier +/// (required = {a, c}) held forever on `a`; the `BarrierRelief` registered +/// for (`cond`, `a`, `m`) fires a phantom arrival for `a` when `cond` +/// completes without routing to it, so `m` fires off `c`'s real arrival +/// instead of deadlocking. +#[tokio::test] +async fn mixed_fan_in_conditional_not_taken_completes() { + let compiled = compile(&mixed_fan_in_graph()).expect("compile"); + let outcome = tokio::time::timeout( + Duration::from_secs(5), + run(&compiled, json!({ "flag": false }), &mock_capabilities()), + ) + .await + .expect("mixed fan-in must not deadlock when the conditional branch is untaken") + .expect("run"); + let out = &outcome.output; + + assert!( + out["nodes"]["a"].is_null(), + "the untaken `true` branch (`a`) must not have run" + ); + let tags = merge_tags(out, "m"); + assert_eq!( + tags, + HashSet::from(["c".to_string()]), + "m should fire with only c's item once its barrier is relieved of a" + ); +} + +/// flag:true — `cond` takes the `true` branch, so `a` DOES run this time. +/// This is the data-loss guard: the fix must not weaken `m`'s barrier into a +/// plain edge (which would let `m` fire off the superstep snapshot *before* +/// `a`'s item commits, silently dropping it) — `m` must still wait for BOTH +/// `a` and `c` and include both items. +#[tokio::test] +async fn mixed_fan_in_conditional_taken_includes_both_items() { + let compiled = compile(&mixed_fan_in_graph()).expect("compile"); + let outcome = tokio::time::timeout( + Duration::from_secs(5), + run(&compiled, json!({ "flag": true }), &mock_capabilities()), + ) + .await + .expect("mixed fan-in must not deadlock when the conditional branch is taken") + .expect("run"); + let out = &outcome.output; + + assert!( + out["nodes"]["b"].is_null(), + "the untaken `false` branch (`b`) must not have run" + ); + let tags = merge_tags(out, "m"); + assert_eq!( + tags, + HashSet::from(["a".to_string(), "c".to_string()]), + "no data loss: m must include a's item as well as c's" + ); +} + +/// Pure conditional join: both of `m`'s predecessors sit behind `cond`'s two +/// ports (no unconditionally-reachable predecessor at all). Only the taken +/// branch ever runs, so `m` must fire with exactly that branch's item — the +/// classic conditional-join case the old (now-removed) `is_conditional_join` +/// special-cased by lowering the edges as plain instead of waiting. The +/// uniform all-waiting-plus-relief design must reproduce the same observable +/// behavior: relief registrations for *both* `a` and `b` let whichever +/// branch's own real arrival satisfy the barrier once the other's relief has +/// already landed. +fn pure_conditional_join_graph() -> WorkflowGraph { + WorkflowGraph { + name: "eng5_pure_conditional_join".to_string(), + nodes: vec![ + trigger("start"), + node("cond", NodeKind::Condition, json!({ "field": "flag" })), + tagged("a", "a"), + tagged("b", "b"), + node("m", NodeKind::Merge, Value::Null), + ], + edges: vec![ + edge("start", "main", "cond"), + edge("cond", "true", "a"), + edge("cond", "false", "b"), + edge("a", "main", "m"), + edge("b", "main", "m"), + ], + ..Default::default() + } +} + +#[tokio::test] +async fn pure_conditional_join_regression() { + let compiled = compile(&pure_conditional_join_graph()).expect("compile"); + + for (flag, taken, untaken) in [(true, "a", "b"), (false, "b", "a")] { + let outcome = tokio::time::timeout( + Duration::from_secs(5), + run(&compiled, json!({ "flag": flag }), &mock_capabilities()), + ) + .await + .unwrap_or_else(|_| panic!("pure conditional join must not deadlock for flag:{flag}")) + .expect("run"); + let out = &outcome.output; + + assert!( + out["nodes"][untaken].is_null(), + "the untaken branch `{untaken}` must not have run for flag:{flag}" + ); + let tags = merge_tags(out, "m"); + assert_eq!( + tags, + HashSet::from([taken.to_string()]), + "m should fire with exactly the taken branch's item for flag:{flag}" + ); + } +} + +/// Pure unconditional fan-in: `start` fans out to `a` and `b` in parallel (no +/// branching involved) and both always run; `m` waits for both, exactly as +/// before — `conditional_predecessors` is empty for this graph, so no +/// `BarrierRelief` is registered and the barrier behaves exactly as it did +/// pre-fix. +#[tokio::test] +async fn pure_unconditional_fan_in_regression() { + let graph = WorkflowGraph { + name: "eng5_pure_unconditional_fan_in".to_string(), + nodes: vec![ + trigger("start"), + tagged("a", "a"), + tagged("b", "b"), + node("m", NodeKind::Merge, Value::Null), + ], + edges: vec![ + edge("start", "main", "a"), + edge("start", "main", "b"), + edge("a", "main", "m"), + edge("b", "main", "m"), + ], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + let outcome = tokio::time::timeout( + Duration::from_secs(5), + run(&compiled, json!({ "seed": 1 }), &mock_capabilities()), + ) + .await + .expect("pure unconditional fan-in must not deadlock") + .expect("run"); + let out = &outcome.output; + + let tags = merge_tags(out, "m"); + assert_eq!( + tags, + HashSet::from(["a".to_string(), "b".to_string()]), + "m must include both a's and b's items" + ); +} From ac6931d023ad150b75b2a09a45206a64cbea17ee Mon Sep 17 00:00:00 2001 From: cyrus Date: Thu, 16 Jul 2026 22:26:53 +0530 Subject: [PATCH 3/4] =?UTF-8?q?test(engine):=20add=20multi-hop=20condition?= =?UTF-8?q?al=20fan-in=20regression=20tests=20(Eng=20=C2=A75)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends the eng5_barrier_relief_e2e coverage with a graph where the conditional predecessor sits TWO hops behind the brancher (cond--true-->x--main-->a, x a plain pass-through) instead of one, per the flagged gap: the previous same-superstep relief check fired a premature phantom arrival for `a` on `cond`'s own completion — before `x` (let alone `a`) had run — even on the taken branch, silently dropping `a`'s item from the merge. Confirmed multi_hop_conditional_taken_no_data_loss fails against the prior tinyagents logic (m ends up with only c's item) and passes against the resolved-routing-target fix (tinyagents commit "fix(graph): key barrier relief on resolved routing target, not same-superstep scheduling"). multi_hop_conditional_not_taken_completes pins the untaken-branch direction stays deadlock-free too. --- tests/eng5_barrier_relief_e2e.rs | 111 +++++++++++++++++++++++++++++++ 1 file changed, 111 insertions(+) diff --git a/tests/eng5_barrier_relief_e2e.rs b/tests/eng5_barrier_relief_e2e.rs index ca59b9e..5878805 100644 --- a/tests/eng5_barrier_relief_e2e.rs +++ b/tests/eng5_barrier_relief_e2e.rs @@ -230,6 +230,117 @@ async fn pure_conditional_join_regression() { } } +/// Multi-hop conditional predecessor: `cond--true-->x--main-->a` (`x` a +/// plain pass-through node, not the merge's direct predecessor), `c` is +/// unconditionally reachable, and `a`/`c` feed `m`. `a` sits TWO hops behind +/// `cond`'s `true` port, not one — this is the coordinator-flagged gap. A +/// same-superstep "is `a` freshly scheduled into `next` this step" check +/// would fire a phantom arrival for `a` the instant `cond` completes, even +/// on the taken branch — `a` is not yet in that step's `next` (only `x` +/// is), it needs one more superstep to run. The fix (`reaches_deterministically` +/// in `tinyagents`) instead resolves `cond`'s actual routing target (`x`) +/// and walks it forward through the compiled graph's deterministic edges, +/// which is correct regardless of hop count. +fn multi_hop_conditional_graph() -> WorkflowGraph { + WorkflowGraph { + name: "eng5_multi_hop_conditional".to_string(), + nodes: vec![ + trigger("start"), + node("cond", NodeKind::Condition, json!({ "field": "flag" })), + node("x", NodeKind::Transform, Value::Null), + tagged("a", "a"), + tagged("b", "b"), + tagged("c", "c"), + node("m", NodeKind::Merge, Value::Null), + ], + edges: vec![ + edge("start", "main", "cond"), + edge("start", "main", "c"), + edge("cond", "true", "x"), + edge("cond", "false", "b"), + edge("x", "main", "a"), + edge("a", "main", "m"), + edge("c", "main", "m"), + ], + ..Default::default() + } +} + +/// flag:true — `cond` takes the `true` branch, so `x` (and, one superstep +/// later, `a`) DOES run. This is the data-loss guard for the multi-hop case: +/// under the coordinator-flagged bug (a same-superstep "freshly scheduled" +/// check), the barrier relief for (`cond`, `a`, `m`) would fire the instant +/// `cond` completes — `a` is not yet in that step's `next`, only `x` is — +/// registering a phantom arrival that lets `m` fire off `c`'s arrival alone, +/// *before* `a`'s real item ever commits. This test MUST fail against that +/// bug and pass once the relief is keyed on `cond`'s resolved routing target +/// reaching `a` (Option 1, implemented in `tinyagents`'s +/// `reaches_deterministically`) — `m` must include BOTH `a`'s and `c`'s +/// items. +#[tokio::test] +async fn multi_hop_conditional_taken_no_data_loss() { + let compiled = compile(&multi_hop_conditional_graph()).expect("compile"); + let outcome = tokio::time::timeout( + Duration::from_secs(5), + run(&compiled, json!({ "flag": true }), &mock_capabilities()), + ) + .await + .expect("multi-hop conditional fan-in must not deadlock when the branch is taken") + .expect("run"); + let out = &outcome.output; + + assert!( + !out["nodes"]["x"]["items"].is_null(), + "the taken `true` branch (`x`) must have run" + ); + assert!( + out["nodes"]["b"].is_null(), + "the untaken `false` branch (`b`) must not have run" + ); + let tags = merge_tags(out, "m"); + assert_eq!( + tags, + HashSet::from(["a".to_string(), "c".to_string()]), + "no data loss: m must include a's item (reached via the x pass-through) as well as c's" + ); +} + +/// flag:false — `cond` takes the `false` branch (to `b`, never wired into +/// `m`), so neither `x` nor `a` ever runs. Implemented Option 1 (the +/// general, correct fix keyed on the brancher's resolved routing target, not +/// same-superstep scheduling — see `tinyagents`'s `reaches_deterministically` +/// in `src/graph/compiled/routing.rs`), so the multi-hop untaken-branch case +/// is fully fixed, not merely a documented residual limitation: `m`'s +/// barrier still clears on `c` alone via the relief, exactly as the direct +/// (one-hop) case does. +#[tokio::test] +async fn multi_hop_conditional_not_taken_completes() { + let compiled = compile(&multi_hop_conditional_graph()).expect("compile"); + let outcome = tokio::time::timeout( + Duration::from_secs(5), + run(&compiled, json!({ "flag": false }), &mock_capabilities()), + ) + .await + .expect("multi-hop conditional fan-in must not deadlock when the branch is untaken") + .expect("run"); + let out = &outcome.output; + + assert!( + out["nodes"]["x"].is_null(), + "the untaken `true` branch (`x`) must not have run" + ); + assert!( + out["nodes"]["a"].is_null(), + "the untaken `true` branch (`a`, behind `x`) must not have run" + ); + let tags = merge_tags(out, "m"); + assert_eq!( + tags, + HashSet::from(["c".to_string()]), + "m should fire with only c's item once its barrier is relieved of a" + ); +} + /// Pure unconditional fan-in: `start` fans out to `a` and `b` in parallel (no /// branching involved) and both always run; `m` waits for both, exactly as /// before — `conditional_predecessors` is empty for this graph, so no From 33f345ba76467daa2094c08d3bd4ce39b7925eb3 Mon Sep 17 00:00:00 2001 From: cyrus Date: Fri, 17 Jul 2026 12:26:37 +0530 Subject: [PATCH 4/4] build(deps): bump tinyagents to 2.1 for GraphBuilder::add_barrier_relief MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mixed fan-in fix (previous commit) depends on tinyagents's new add_barrier_relief primitive, published as tinyagents 2.1.0 (crates.io) — tinyagents's own major-version line moved to 2.x independently of this work, so this also crosses that boundary from the "1.2" this crate previously required. Verified tinyflows compiles and its full test suite (340 lib tests, all integration files, 12 doctests) stays green against the real published 2.1.0 — no other breaking changes between the two lines affect this crate. --- Cargo.lock | 169 +++++++++++++++++++++++++++++++++++++++++++---------- Cargo.toml | 2 +- 2 files changed, 139 insertions(+), 32 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c6b8874..8fdf794 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11,6 +11,15 @@ dependencies = [ "memchr", ] +[[package]] +name = "android_system_properties" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311" +dependencies = [ + "libc", +] + [[package]] name = "async-trait" version = "0.1.89" @@ -69,11 +78,11 @@ checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" [[package]] name = "block-buffer" -version = "0.10.4" +version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" dependencies = [ - "generic-array", + "hybrid-array", ] [[package]] @@ -121,23 +130,47 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" +[[package]] +name = "chrono" +version = "0.4.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" +dependencies = [ + "iana-time-zone", + "js-sys", + "num-traits", + "wasm-bindgen", + "windows-link", +] + +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + [[package]] name = "cpufeatures" -version = "0.2.17" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" dependencies = [ "libc", ] [[package]] name = "crypto-common" -version = "0.1.7" +version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" dependencies = [ - "generic-array", - "typenum", + "hybrid-array", ] [[package]] @@ -174,11 +207,12 @@ dependencies = [ [[package]] name = "digest" -version = "0.10.7" +version = "0.11.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" dependencies = [ "block-buffer", + "const-oid", "crypto-common", ] @@ -342,16 +376,6 @@ dependencies = [ "slab", ] -[[package]] -name = "generic-array" -version = "0.14.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" -dependencies = [ - "typenum", - "version_check", -] - [[package]] name = "getrandom" version = "0.2.17" @@ -430,6 +454,15 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "hybrid-array" +version = "0.4.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c" +dependencies = [ + "typenum", +] + [[package]] name = "hyper" version = "1.10.1" @@ -489,6 +522,30 @@ dependencies = [ "tracing", ] +[[package]] +name = "iana-time-zone" +version = "0.1.65" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470" +dependencies = [ + "android_system_properties", + "core-foundation-sys", + "iana-time-zone-haiku", + "js-sys", + "log", + "wasm-bindgen", + "windows-core", +] + +[[package]] +name = "iana-time-zone-haiku" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" +dependencies = [ + "cc", +] + [[package]] name = "icu_collections" version = "2.1.1" @@ -1224,9 +1281,9 @@ dependencies = [ [[package]] name = "sha2" -version = "0.10.9" +version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" dependencies = [ "cfg-if", "cpufeatures", @@ -1339,11 +1396,13 @@ dependencies = [ [[package]] name = "tinyagents" -version = "1.3.0" +version = "2.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3ebc9790a96fe910c98c60758fad23bd33faeb77889bc2355e6dc3f5aa70e0c" +checksum = "0d9746cb9ff6f37646b2c91eebe3dee775a56083d21ed41446a54c930efaf056" dependencies = [ "async-trait", + "bytes", + "chrono", "futures", "reqwest", "serde", @@ -1351,6 +1410,7 @@ dependencies = [ "sha2", "thiserror", "tokio", + "tracing", ] [[package]] @@ -1581,12 +1641,6 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" -[[package]] -name = "version_check" -version = "0.9.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" - [[package]] name = "wait-timeout" version = "0.2.1" @@ -1717,12 +1771,65 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "windows-core" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-implement" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "windows-interface" +version = "0.59.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-result" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-strings" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" +dependencies = [ + "windows-link", +] + [[package]] name = "windows-sys" version = "0.52.0" diff --git a/Cargo.toml b/Cargo.toml index 4d0fa16..5512b58 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,7 +27,7 @@ serde_json = "1" async-trait = "0.1" thiserror = "2" futures-timer = "3" -tinyagents = "1.2" +tinyagents = "2.1" tracing = "0.1" jaq-core = "3.1.0" jaq-std = "3.0.1"