diff --git a/Cargo.lock b/Cargo.lock index 39a0b01..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,12 +1396,13 @@ dependencies = [ [[package]] name = "tinyagents" -version = "1.7.1" +version = "2.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9fcab4dcd797f2ad03aac83c6525fe4d614af637441ae52018711a2119b74de8" +checksum = "0d9746cb9ff6f37646b2c91eebe3dee775a56083d21ed41446a54c930efaf056" dependencies = [ "async-trait", "bytes", + "chrono", "futures", "reqwest", "serde", @@ -1352,6 +1410,7 @@ dependencies = [ "sha2", "thiserror", "tokio", + "tracing", ] [[package]] @@ -1582,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" @@ -1718,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" 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/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(), + } + ); + } } diff --git a/tests/eng5_barrier_relief_e2e.rs b/tests/eng5_barrier_relief_e2e.rs new file mode 100644 index 0000000..5878805 --- /dev/null +++ b/tests/eng5_barrier_relief_e2e.rs @@ -0,0 +1,383 @@ +#![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}" + ); + } +} + +/// 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 +/// `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" + ); +}