feat(graph): BarrierRelief primitive for mixed conditional/unconditional fan-in

src/graph/builder/mod.rs

@@ -29,6 +29,32 @@ use crate::graph::reducer::{OverwriteStateReducer, StateReducer};
 use crate::harness::ids::{GraphId, NodeId};
 use crate::{Result, TinyAgentsError};
 
+/// A relief registration for a mixed fan-in barrier.
+///
+/// A waiting/barrier node (see [`GraphBuilder::add_waiting_edge`]) normally
+/// activates only once *every* registered predecessor has arrived. When one
+/// of those predecessors (`relief_node`) is only reachable via a conditional
+/// branch out of `source`, and `source` routes elsewhere instead, that
+/// predecessor never runs and the barrier would wait forever.
+///
+/// A [`BarrierRelief`] fixes that without weakening the barrier into a plain
+/// edge (which would let a *taken* branch's downstream data race the merge
+/// and get silently dropped): when `source` completes a superstep without
+/// routing to `relief_node`, the executor registers a phantom arrival of
+/// `relief_node` at `barrier_node`, so the barrier can still clear on the
+/// predecessors that actually ran.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct BarrierRelief {
+    /// The brancher node whose conditional routing determines whether
+    /// `relief_node` runs.
+    pub source: NodeId,
+    /// The conditional-only predecessor of `barrier_node` that `source` may
+    /// or may not route to.
+    pub relief_node: NodeId,
+    /// The mixed fan-in (all-waiting) node gated on `relief_node`'s arrival.
+    pub barrier_node: NodeId,
+}
+
 impl<State, Update> Default for GraphBuilder<State, Update>
 where
     State: Clone + Send + Sync + 'static,
@@ -55,6 +81,7 @@ where
             branches: HashMap::new(),
             command_nodes: HashSet::new(),
             waiting: HashMap::new(),
+            barrier_reliefs: Vec::new(),
             reducer: None,
             recursion_limit: 50,
             parallel: false,
@@ -201,6 +228,35 @@ where
         self
     }
 
+    /// Registers a barrier relief for a mixed fan-in barrier.
+    ///
+    /// When `source` completes a superstep without routing to `relief_node`
+    /// — because `relief_node` sits behind a conditional branch `source` did
+    /// not take this run — this records a phantom arrival of `relief_node`
+    /// at `barrier_node`'s waiting-edge barrier (registered separately via
+    /// [`Self::add_waiting_edge`]), so an all-waiting merge downstream of a
+    /// mixed fan-in (one unconditionally reachable predecessor plus one
+    /// reachable only via a conditional branch) can still clear on its
+    /// remaining predecessors instead of deadlocking on a branch that will
+    /// never run.
+    ///
+    /// `relief_node` must be one of `barrier_node`'s registered waiting
+    /// predecessors; a relief for a barrier with no matching waiting
+    /// registration is a no-op at execution time.
+    pub fn add_barrier_relief(
+        mut self,
+        source: impl Into<NodeId>,
+        relief_node: impl Into<NodeId>,
+        barrier_node: impl Into<NodeId>,
+    ) -> Self {
+        self.barrier_reliefs.push(BarrierRelief {
+            source: source.into(),
+            relief_node: relief_node.into(),
+            barrier_node: barrier_node.into(),
+        });
+        self
+    }
+
     /// Sets the entry node, i.e. `add_edge(START, node)`.
     pub fn set_entry(self, node: impl Into<NodeId>) -> Self {
         self.add_edge(START, node)
@@ -406,6 +462,7 @@ where
             max_concurrency,
             node_timeout,
             node_meta,
+            barrier_reliefs,
         } = self;
 
         Ok(CompiledGraph::from_parts(
@@ -423,6 +480,7 @@ where
             max_concurrency,
             node_timeout,
             node_meta,
+            barrier_reliefs,
         ))
     }
 

src/graph/builder/types.rs

@@ -222,6 +222,9 @@ pub struct GraphBuilder<State, Update> {
     /// Barrier/waiting edges: target node -> set of predecessor nodes that must
     /// all have completed (across steps) before the target activates.
     pub(crate) waiting: HashMap<NodeId, HashSet<NodeId>>,
+    /// Mixed fan-in barrier relief registrations; see
+    /// [`super::BarrierRelief`].
+    pub(crate) barrier_reliefs: Vec<super::BarrierRelief>,
     pub(crate) reducer: Option<Arc<dyn StateReducer<State, Update>>>,
     pub(crate) recursion_limit: usize,
     /// When true, active nodes in a superstep run concurrently (fan-out).

src/graph/compiled/mod.rs

@@ -78,7 +78,8 @@ use std::sync::Arc;
 use std::time::{Duration, SystemTime};
 
 use crate::graph::builder::{
-    Branch, BuilderNode, END, ForkId, NodeContext, NodeFuture, NodeHandler, NodeMeta, START,
+    BarrierRelief, Branch, BuilderNode, END, ForkId, NodeContext, NodeFuture, NodeHandler,
+    NodeMeta, START,
 };
 use crate::graph::checkpoint::{
     BarrierArrivals, Checkpoint, CheckpointConfig, CheckpointTuple, Checkpointer, DurabilityMode,
@@ -253,6 +254,7 @@ impl<State, Update> CompiledGraph<State, Update> {
         max_concurrency: Option<usize>,
         node_timeout: Option<Duration>,
         node_meta: HashMap<NodeId, NodeMeta>,
+        barrier_reliefs: Vec<BarrierRelief>,
     ) -> Self {
         Self {
             graph_id,
@@ -262,6 +264,7 @@ impl<State, Update> CompiledGraph<State, Update> {
             branches: Arc::new(branches),
             command_nodes: Arc::new(command_nodes),
             waiting: Arc::new(waiting),
+            barrier_reliefs: Arc::new(barrier_reliefs),
             node_meta: Arc::new(node_meta),
             entry,
             reducer,

src/graph/compiled/routing.rs

@@ -21,9 +21,16 @@ where
     ) -> Result<Vec<Activation>> {
         let mut next: Vec<Activation> = Vec::new();
         let mut next_seen: HashSet<NodeId> = HashSet::new();
+        // Resolved targets per activation index, captured once here and
+        // reused by the barrier-relief pass below instead of calling
+        // `self.route` a second time — a router closure is only guaranteed
+        // pure/idempotent per the `route`/`add_conditional_edges` contract,
+        // not safe to invoke twice for the same activation.
+        let mut resolved: Vec<Vec<RouteTarget>> = Vec::with_capacity(completed.len());
         for (index, activation) in completed.iter().enumerate() {
             let node_id = &activation.node;
             let targets = self.route(node_id, goto_map.get(&index).map(Vec::as_slice), state)?;
+            resolved.push(targets.clone());
             for target in targets {
                 let tnode = target.node().clone();
                 if tnode.as_str() == END {
@@ -59,9 +66,112 @@ where
                 }
             }
         }
+
+        // Mixed fan-in barrier relief: a waiting/barrier node normally
+        // activates only once every registered predecessor has arrived. When
+        // one of those predecessors is reachable only via a conditional
+        // branch, and the *taken* branch does not lead toward it, that
+        // predecessor will never arrive on its own — register a phantom
+        // arrival on its behalf so the barrier can still clear on the
+        // predecessors that actually ran, instead of deadlocking forever.
+        //
+        // The check is keyed on `source`'s *resolved routing target* this
+        // step (via `reaches_deterministically`), not on whether
+        // `relief_node` is freshly scheduled into `next` this same
+        // superstep. A same-superstep check is wrong for a multi-hop
+        // conditional predecessor (`source --branch--> x --main-->
+        // relief_node`, `x` a plain pass-through): `relief_node` would not
+        // yet be in `next` on the step `source` itself completes — even on
+        // the *taken* branch, where `relief_node` WILL run once `x` does —
+        // so a same-superstep check would fire a premature phantom arrival
+        // and let the barrier clear before the real predecessor's data
+        // commits, reintroducing the exact data-loss bug this primitive
+        // exists to prevent. Resolving `source`'s actual target and walking
+        // it forward through deterministic (non-branching) edges is correct
+        // for both direct and multi-hop cases because it answers "was the
+        // branch leading to `relief_node` taken", not "did `relief_node`
+        // happen to run in lockstep with `source`".
+        for relief in self.barrier_reliefs.iter() {
+            let source_indices: Vec<usize> = completed
+                .iter()
+                .enumerate()
+                .filter(|(_, activation)| activation.node == relief.source)
+                .map(|(index, _)| index)
+                .collect();
+            if source_indices.is_empty() {
+                continue;
+            }
+            let branch_taken = source_indices.iter().any(|index| {
+                resolved[*index].iter().any(|target| {
+                    self.reaches_deterministically(
+                        target.node(),
+                        &relief.relief_node,
+                        &relief.barrier_node,
+                    )
+                })
+            });
+            // A relief_node freshly scheduled into `next` this step (a
+            // direct, single-hop predecessor) is also proof the branch was
+            // taken; kept as a defensive fallback alongside the resolved-
+            // target check above.
+            if branch_taken || next_seen.contains(&relief.relief_node) {
+                continue;
+            }
+            let Some(required) = self.waiting.get(&relief.barrier_node) else {
+                continue;
+            };
+            let arrived = barrier_arrivals
+                .entry(relief.barrier_node.clone())
+                .or_default();
+            arrived.insert(relief.relief_node.clone());
+            if !required.is_subset(arrived) {
+                continue;
+            }
+            barrier_arrivals.remove(&relief.barrier_node);
+            if next_seen.insert(relief.barrier_node.clone()) {
+                next.push(Activation {
+                    node: relief.barrier_node.clone(),
+                    send_arg: None,
+                });
+            }
+        }
+
         Ok(next)
     }
 
+    /// Whether `to` is reachable from `from` by following only deterministic
+    /// static routing — plain/waiting edges (`self.edges`), which resolve to
+    /// exactly one successor with no runtime decision — without ever
+    /// expanding through `stop`.
+    ///
+    /// This is what makes barrier-relief evaluation correct for a
+    /// multi-hop conditional predecessor: once a brancher's routing decision
+    /// for this step is resolved to a concrete target, whether that target
+    /// eventually leads to a barrier's conditional predecessor is a static
+    /// property of the compiled topology for any chain of plain pass-through
+    /// nodes — it does not depend on when each hop happens to run. A further
+    /// conditional/command node along the way (no `self.edges` entry) is a
+    /// second runtime decision this walk cannot resolve ahead of time, so it
+    /// conservatively reports unreachable there (falling back to the
+    /// same-superstep check).
+    fn reaches_deterministically(&self, from: &NodeId, to: &NodeId, stop: &NodeId) -> bool {
+        if from == to {
+            return true;
+        }
+        let mut current = from;
+        let mut seen: HashSet<&NodeId> = HashSet::new();
+        while let Some(next) = self.edges.get(current) {
+            if next == to {
+                return true;
+            }
+            if next == stop || !seen.insert(next) {
+                return false;
+            }
+            current = next;
+        }
+        false
+    }
+
     /// Resolves the next routing targets for `node_id`.
     ///
     /// Command `goto` (which may include [`Send`] packets) wins over static and

src/graph/compiled/test.rs

@@ -1243,6 +1243,71 @@ async fn barrier_arrivals_survive_interrupt_and_resume() {
     assert_eq!(done.state.value, 103);
 }
 
+#[tokio::test]
+async fn barrier_relief_fires_when_source_skips_relief_node() {
+    // Mixed fan-in: `m` waits on both `a` and `c`. `condition` never routes to
+    // `a` — it always takes the `skip` route to END, simulating an untaken
+    // conditional branch — so without a barrier relief `m` would deadlock
+    // forever waiting on a predecessor that never runs.
+    // `add_barrier_relief("condition", "a", "m")` registers `a`'s phantom
+    // arrival at `m` whenever `condition` completes without activating `a`,
+    // so `m` still fires once `c`'s real arrival lands.
+    let graph = GraphBuilder::<Vec<String>, String>::new()
+        .with_parallel(true)
+        .set_reducer(ClosureStateReducer::new(|mut s: Vec<String>, u: String| {
+            s.push(u);
+            Ok(s)
+        }))
+        .add_node("start", |_s, _c: NodeContext| async move {
+            Ok(NodeResult::Command(
+                Command::default().with_goto(["condition", "c"]),
+            ))
+        })
+        .add_node("condition", |_s, _c: NodeContext| async move {
+            Ok(NodeResult::Update("condition".to_string()))
+        })
+        .add_node("a", |_s, _c: NodeContext| async move {
+            Ok(NodeResult::Update("a".to_string()))
+        })
+        .add_node("c", |_s, _c: NodeContext| async move {
+            Ok(NodeResult::Update("c".to_string()))
+        })
+        .add_node("m", |_s, _c: NodeContext| async move {
+            Ok(NodeResult::Update("m".to_string()))
+        })
+        .set_entry("start")
+        .mark_command_routing("start")
+        // `condition` always takes the `skip` route to END — `a` is never
+        // reached via a real edge.
+        .add_conditional_edges(
+            "condition",
+            |_s: &Vec<String>| "skip".to_string(),
+            [("skip", END)],
+        )
+        .add_waiting_edge("a", "m")
+        .add_waiting_edge("c", "m")
+        .add_barrier_relief("condition", "a", "m")
+        .set_finish("m")
+        .compile()
+        .unwrap();
+
+    let run = graph.run(Vec::new()).await.unwrap();
+
+    assert!(
+        run.visited.iter().any(|n| n.as_str() == "m"),
+        "m must activate via the barrier relief even though `a` never ran"
+    );
+    assert!(
+        !run.visited.iter().any(|n| n.as_str() == "a"),
+        "a must never have run (condition always skipped it)"
+    );
+    assert_eq!(
+        run.state,
+        vec!["condition".to_string(), "c".to_string(), "m".to_string()],
+        "m fires off condition+c's real contributions, with no phantom `a` update"
+    );
+}
+
 #[tokio::test]
 async fn reducer_error_at_boundary_transitions_run_to_failed() {
     // A reducer error raised at the step boundary (after the node ran) must

src/graph/compiled/types.rs

@@ -4,7 +4,7 @@ use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
 use crate::graph::builder::START;
-use crate::graph::builder::{Branch, BuilderNode, NodeMeta};
+use crate::graph::builder::{BarrierRelief, Branch, BuilderNode, NodeMeta};
 use crate::graph::checkpoint::{
     CheckpointConfig, CheckpointMetadata, Checkpointer, DurabilityMode,
 };
@@ -32,6 +32,8 @@ pub struct CompiledGraph<State, Update> {
     /// Barrier/waiting edges: target -> the predecessor set that must all
     /// complete (across steps) before the target activates.
     pub(crate) waiting: Arc<HashMap<NodeId, HashSet<NodeId>>>,
+    /// Mixed fan-in barrier relief registrations; see [`BarrierRelief`].
+    pub(crate) barrier_reliefs: Arc<Vec<BarrierRelief>>,
     /// Behavior-free per-node markers/metadata surfaced by the topology export.
     pub(crate) node_meta: Arc<HashMap<NodeId, NodeMeta>>,
     pub(crate) entry: NodeId,
@@ -106,6 +108,7 @@ impl<State, Update> Clone for CompiledGraph<State, Update> {
             branches: self.branches.clone(),
             command_nodes: self.command_nodes.clone(),
             waiting: self.waiting.clone(),
+            barrier_reliefs: self.barrier_reliefs.clone(),
             node_meta: self.node_meta.clone(),
             entry: self.entry.clone(),
             reducer: self.reducer.clone(),