diff --git a/src/function/agents.rs b/src/function/agents.rs index a62b2c1..d991d73 100644 --- a/src/function/agents.rs +++ b/src/function/agents.rs @@ -786,13 +786,7 @@ pub async fn run_agent_for_graph( debug!("Spawning agent '{agent_name}' for graph node as '{agent_id}'"); - let result = run_child_agent(child_ctx, input, child_abort).await; - - if let Some(registry) = &peer_registry { - registry.mark_finished(&agent_id); - } - - result + run_child_agent(child_ctx, input, child_abort).await } async fn populate_agent_mcp_runtime(ctx: &mut RequestContext, server_ids: &[String]) -> Result<()> { diff --git a/src/graph/executor.rs b/src/graph/executor.rs index aa854b5..01b4408 100644 --- a/src/graph/executor.rs +++ b/src/graph/executor.rs @@ -179,10 +179,13 @@ impl GraphExecutor { logger.node_start(&node, in_super_step); let branch_state = state.fork_for_branch_state(); let mut branch_ctx = ctx.fork_for_branch(); + let mut peer_id: Option = None; if let Some(assignment) = peer_assignments.remove(node_id) { + peer_id = Some(assignment.0.clone()); branch_ctx.peer_registry = peer_registry.clone(); branch_ctx.peer_assignment = Some(assignment); } + let registry_for_task = peer_registry.clone(); if in_super_step { branch_ctx.render_mode = RenderMode::Silent; } @@ -216,12 +219,16 @@ impl GraphExecutor { let mut state = branch_state; let mut ctx = branch_ctx; let step_ctx = StepContext { - graph: graph_clone.as_ref(), + graph: Arc::clone(&graph_clone), script_executor: &script_exec_clone, max_concurrency, abort_signal: &abort_clone, + branch_mode: false, }; let result = step(&node, &mut state, &mut ctx, &step_ctx, ¤t).await; + if let (Some(registry), Some(id)) = (®istry_for_task, &peer_id) { + registry.mark_finished(id); + } let elapsed = node_start.elapsed(); match &result { Ok(StepResult::Continue(targets)) => { @@ -398,10 +405,11 @@ fn provision_frontier_peers( } pub(super) struct StepContext<'a> { - pub graph: &'a Graph, + pub graph: Arc, pub script_executor: &'a ScriptExecutor, pub max_concurrency: usize, pub abort_signal: &'a AbortSignal, + pub branch_mode: bool, } impl StepContext<'_> { @@ -410,7 +418,7 @@ impl StepContext<'_> { } } -enum StepResult { +pub(super) enum StepResult { // The set of next-node ids the executor should add to the next super-step's // frontier. A `Vec` of length 1 for sequential routing (default) and the // full target list for fan-out (`next: [a, b, ...]`). Dynamic single-route @@ -420,7 +428,7 @@ enum StepResult { End(String), } -async fn step( +pub(super) async fn step( node: &Node, state: &mut StateManager, ctx: &mut RequestContext, @@ -430,7 +438,7 @@ async fn step( match &node.node_type { NodeType::Agent(agent_node) => { AgentNodeExecutor::execute(agent_node, state, ctx).await?; - let targets = static_next_targets(node, current, "agent")?; + let targets = static_next_targets(node, current, "agent", step_ctx.branch_mode)?; Ok(StepResult::Continue(targets)) } NodeType::Script(script_node) => { @@ -452,7 +460,7 @@ async fn step( }; let targets = match dynamic { Some(n) => vec![n], - None => static_next_targets(node, current, "script")?, + None => static_next_targets(node, current, "script", step_ctx.branch_mode)?, }; Ok(StepResult::Continue(targets)) } @@ -468,26 +476,38 @@ async fn step( NodeType::Llm(llm_node) => { let outcome = LlmNodeExecutor::execute(current, llm_node, state, ctx).await?; let targets = match outcome { - LlmExecutionOutcome::Continue => static_next_targets(node, current, "llm")?, + LlmExecutionOutcome::Continue => { + static_next_targets(node, current, "llm", step_ctx.branch_mode)? + } LlmExecutionOutcome::FellBack(target) => vec![target], }; Ok(StepResult::Continue(targets)) } NodeType::Rag(rag_node) => { RagNodeExecutor::execute(rag_node, current, state, ctx).await?; - let targets = static_next_targets(node, current, "rag")?; + let targets = static_next_targets(node, current, "rag", step_ctx.branch_mode)?; Ok(StepResult::Continue(targets)) } NodeType::End(end_node) => Ok(StepResult::End(resolve_end_output(end_node, state))), NodeType::Map(map_node) => { - let targets = static_next_targets(node, current, "map")?; + let targets = static_next_targets(node, current, "map", step_ctx.branch_mode)?; MapNodeExecutor::execute(map_node, state, ctx, step_ctx, current).await?; Ok(StepResult::Continue(targets)) } } } -fn static_next_targets(node: &Node, current: &str, kind: &str) -> Result> { +/// Inside a map branch a node without `next` simply ends the item's chain; on +/// the main flow it is a routing error. +fn static_next_targets( + node: &Node, + current: &str, + kind: &str, + branch_mode: bool, +) -> Result> { + if node.next.is_none() && branch_mode { + return Ok(Vec::new()); + } node.next .as_ref() .map(|t| t.as_slice().to_vec()) diff --git a/src/graph/map.rs b/src/graph/map.rs index 99325f9..808a65b 100644 --- a/src/graph/map.rs +++ b/src/graph/map.rs @@ -1,159 +1,289 @@ -use super::agent::AgentNodeExecutor; -use super::executor::StepContext; -use super::llm::LlmNodeExecutor; -use super::rag::RagNodeExecutor; +use super::executor::{StepContext, StepResult, step}; use super::state::StateManager; -use super::types::{ConcurrencyCap, MapNode, NodeType}; +use super::types::{ConcurrencyCap, Graph, MapNode, NodeType}; +use super::validator::branch_subgraph; use crate::config::{RenderMode, RequestContext}; use crate::graph::type_name; use crate::supervisor::mailbox::{Inbox, PeerAssignment, PeerRegistry, graph_agent_id}; -use anyhow::{Context, Result, anyhow}; -use futures_util::future::join_all; +use anyhow::{Context, Result, anyhow, bail}; +use futures_util::future::{BoxFuture, join_all}; use serde_json::Value; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; use tokio::sync::Semaphore; pub(super) struct MapNodeExecutor; impl MapNodeExecutor { - pub(super) async fn execute( - node: &MapNode, - state: &mut StateManager, - ctx: &mut RequestContext, - step_ctx: &StepContext<'_>, - node_id: &str, - ) -> Result<()> { - let over_value = state - .interpolate_raw(&node.over) - .with_context(|| format!("map node '{node_id}': evaluating `over` template"))?; + /// Boxed because item chains call back into `step`, which calls this; + /// the concrete return type breaks the otherwise-recursive opaque future. + pub(super) fn execute<'a>( + node: &'a MapNode, + state: &'a mut StateManager, + ctx: &'a mut RequestContext, + step_ctx: &'a StepContext<'a>, + node_id: &'a str, + ) -> BoxFuture<'a, Result<()>> { + Box::pin(run_map(node, state, ctx, step_ctx, node_id)) + } +} - let items = over_value.as_array().ok_or_else(|| { - anyhow!( - "map node '{}': `over` template '{}' must resolve to an array, got {}", - node_id, - node.over, - type_name(&over_value) - ) - })?; - let items = items.clone(); +async fn run_map( + node: &MapNode, + state: &mut StateManager, + ctx: &mut RequestContext, + step_ctx: &StepContext<'_>, + node_id: &str, +) -> Result<()> { + let over_value = state + .interpolate_raw(&node.over) + .with_context(|| format!("map node '{node_id}': evaluating `over` template"))?; - let branch_node = step_ctx - .graph - .get_node(&node.branch) + let items = over_value.as_array().ok_or_else(|| { + anyhow!( + "map node '{}': `over` template '{}' must resolve to an array, got {}", + node_id, + node.over, + type_name(&over_value) + ) + })?; + let items = items.clone(); + + if step_ctx.graph.get_node(&node.branch).is_none() { + bail!( + "map node '{node_id}': branch '{}' not found in graph", + node.branch + ); + } + let subgraph = Arc::new(branch_subgraph(&step_ctx.graph, &node.branch)); + + let max_conc = resolve_max_concurrency(node, state, step_ctx.max_concurrency, node_id)?; + let semaphore = Arc::new(Semaphore::new(max_conc)); + let mut sub_tasks = Vec::with_capacity(items.len()); + + let peers = provision_map_peers(&step_ctx.graph, &subgraph, &node.branch, items.len()); + + for (idx, item) in items.iter().enumerate() { + let item = item.clone(); + let as_name = node.as_name.clone(); + let mut sub_state = state.fork_for_branch_state(); + // The chain must write the result itself; an inherited parent + // value would otherwise be collected as if the item produced it. + sub_state.state_mut().remove(&node.output_key); + sub_state.state_mut().set(as_name, item); + let mut sub_ctx = ctx.fork_for_branch(); + sub_ctx.render_mode = RenderMode::Silent; + let item_peers: Option<(Arc, PeerAssignment)> = peers + .as_ref() + .map(|(registry, assignments)| (Arc::clone(registry), assignments[idx].clone())); + let script_clone = step_ctx.script_executor.clone(); + let graph = Arc::clone(&step_ctx.graph); + let subgraph = Arc::clone(&subgraph); + let entry = node.branch.clone(); + let map_id = node_id.to_string(); + let max_concurrency = step_ctx.max_concurrency; + let sem = semaphore.clone(); + let abort = step_ctx.abort_signal.clone(); + + let task = tokio::spawn(async move { + let _permit = sem + .acquire() + .await + .expect("map semaphore should not be closed"); + let mut state = sub_state; + let mut ctx = sub_ctx; + let step_ctx = StepContext { + graph, + script_executor: &script_clone, + max_concurrency, + abort_signal: &abort, + branch_mode: true, + }; + let chain = ItemChain { + map_id: &map_id, + entry: &entry, + subgraph: &subgraph, + idx, + peers: item_peers.as_ref(), + }; + let result = run_item_chain(&chain, &mut state, &mut ctx, &step_ctx).await; + (idx, state, result) + }); + sub_tasks.push(task); + } + + let joined = join_all(sub_tasks).await; + + // Collect outputs keyed by input index so order is preserved regardless of finish order. + let mut outputs: HashMap = HashMap::new(); + for join_result in joined { + let (idx, sub_state, exec_result) = + join_result.map_err(|e| anyhow!("map sub-branch panicked: {e}"))?; + + exec_result?; + + let output_value = sub_state + .state() + .get(&node.output_key) + .cloned() .ok_or_else(|| { anyhow!( - "map node '{node_id}': branch '{}' not found in graph", - node.branch - ) - })? - .clone(); - - let max_conc = resolve_max_concurrency(node, state, step_ctx.max_concurrency, node_id)?; - let semaphore = Arc::new(Semaphore::new(max_conc)); - let mut sub_tasks = Vec::with_capacity(items.len()); - - let peers = provision_map_peers(&branch_node.node_type, &node.branch, items.len()); - - for (idx, item) in items.iter().enumerate() { - let item = item.clone(); - let as_name = node.as_name.clone(); - let branch_clone = branch_node.clone(); - let mut sub_state = state.fork_for_branch_state(); - let mut sub_ctx = ctx.fork_for_branch(); - sub_ctx.render_mode = RenderMode::Silent; - if let Some((registry, assignments)) = &peers { - sub_ctx.peer_registry = Some(Arc::clone(registry)); - sub_ctx.peer_assignment = Some(assignments[idx].clone()); - } - let script_clone = step_ctx.script_executor.clone(); - let sub_branch_id = node.branch.clone(); - let sem = semaphore.clone(); - let abort = step_ctx.abort_signal.clone(); - - sub_state.state_mut().set(as_name, item); - - let task = tokio::spawn(async move { - let _permit = sem - .acquire() - .await - .expect("map semaphore should not be closed"); - if abort.aborted() { - return ( - idx, - sub_state, - Err(anyhow!("map sub-branch [{idx}] aborted")), - ); - } - let mut state = sub_state; - let mut ctx = sub_ctx; - - let exec_result: Result<()> = match &branch_clone.node_type { - NodeType::Llm(n) => { - LlmNodeExecutor::execute(&branch_clone.id, n, &mut state, &mut ctx) - .await - .map(|_| ()) - } - NodeType::Agent(n) => AgentNodeExecutor::execute(n, &mut state, &mut ctx) - .await - .map(|_| ()), - NodeType::Rag(n) => { - RagNodeExecutor::execute(n, &sub_branch_id, &mut state, &mut ctx).await - } - NodeType::Script(n) => script_clone.execute(n, &mut state).await.map(|_| ()), - _ => Err(anyhow!( - "map branch '{}' has type that cannot run inside a map \ - (validator should have caught this; internal error)", - branch_clone.id - )), - }; - - (idx, state, exec_result) - }); - sub_tasks.push(task); - } - - let joined = join_all(sub_tasks).await; - - // Collect outputs keyed by input index so order is preserved regardless of finish order. - let mut outputs: HashMap = HashMap::new(); - for join_result in joined { - let (idx, sub_state, exec_result) = - join_result.map_err(|e| anyhow!("map sub-branch panicked: {e}"))?; - - exec_result - .with_context(|| format!("map node '{node_id}': sub-branch [{idx}] failed"))?; - - let output_value = sub_state - .state() - .get(&node.output_key) - .cloned() - .ok_or_else(|| { - anyhow!( - "map node '{node_id}': sub-branch [{idx}] did not write \ - `output_key` '{}'", - node.output_key - ) - })?; - - outputs.insert(idx, output_value); - } - - let mut collected = Vec::with_capacity(items.len()); - for idx in 0..items.len() { - let value = outputs.remove(&idx).ok_or_else(|| { - anyhow!( - "map node '{node_id}': internal error: missing result for sub-branch [{idx}]" + "map node '{node_id}': sub-branch [{idx}] did not write output_key '{}'; \ + write it via state_updates, output_schema, or a script's JSON output \ + (the parent's value is not inherited inside a map branch)", + node.output_key ) })?; - collected.push(value); + + outputs.insert(idx, output_value); + } + + let mut collected = Vec::with_capacity(items.len()); + for idx in 0..items.len() { + let value = outputs.remove(&idx).ok_or_else(|| { + anyhow!("map node '{node_id}': internal error: missing result for sub-branch [{idx}]") + })?; + collected.push(value); + } + + state + .state_mut() + .set(node.collect_into.clone(), Value::Array(collected)); + + Ok(()) +} + +/// One item's slice of a map fan-out: where its chain starts, which nodes +/// it may visit, and (when the fan-out is a team) the identity it speaks as. +struct ItemChain<'a> { + map_id: &'a str, + entry: &'a str, + subgraph: &'a HashSet, + idx: usize, + peers: Option<&'a (Arc, PeerAssignment)>, +} + +/// Runs one item's chain to completion on its forked state and ctx. Owns the +/// item's teammate identity for the whole chain: arms it only before +/// `teammates: true` agent steps and retires it when the chain exits, +/// successfully or not. +async fn run_item_chain( + chain: &ItemChain<'_>, + state: &mut StateManager, + ctx: &mut RequestContext, + step_ctx: &StepContext<'_>, +) -> Result<()> { + let result = run_chain_steps(chain, state, ctx, step_ctx).await; + if let Some((registry, (id, _))) = chain.peers { + registry.mark_finished(id); + } + result +} + +async fn run_chain_steps( + chain: &ItemChain<'_>, + state: &mut StateManager, + ctx: &mut RequestContext, + step_ctx: &StepContext<'_>, +) -> Result<()> { + let mut current = chain.entry.to_string(); + let mut step_no = 0usize; + loop { + step_no += 1; + let next = run_chain_step(chain, state, ctx, step_ctx, ¤t) + .await + .with_context(|| { + format!( + "map node '{}': sub-branch [{}] failed at node '{current}' (step {step_no})", + chain.map_id, chain.idx + ) + })?; + match next { + Some(target) => current = target, + None => return Ok(()), } + } +} - state - .state_mut() - .set(node.collect_into.clone(), Value::Array(collected)); +/// Runs a single node of the chain and returns the node to visit next, or +/// `None` once the chain has ended. +async fn run_chain_step( + chain: &ItemChain<'_>, + state: &mut StateManager, + ctx: &mut RequestContext, + step_ctx: &StepContext<'_>, + current: &str, +) -> Result> { + let ItemChain { map_id, idx, .. } = *chain; + if step_ctx.abort_signal.aborted() { + bail!("map sub-branch [{idx}] aborted"); + } + let node = step_ctx.graph.get_node(current).ok_or_else(|| { + anyhow!("map node '{map_id}': sub-branch [{idx}] routed to unknown node '{current}'") + })?; + if !matches!( + node.node_type, + NodeType::Llm(_) | NodeType::Agent(_) | NodeType::Rag(_) | NodeType::Script(_) + ) { + bail!( + "map branch '{current}' has type that cannot run inside a map \ + (validator should have caught this; internal error)" + ); + } - Ok(()) + state.state_mut().visit_node(current); + let visits = state.state().loop_count(current); + let max_loops = step_ctx.graph.settings.max_loop_iterations; + if visits > max_loops { + bail!( + "node '{current}' visited {visits} times in map branch [{idx}] \ + (max_loop_iterations={max_loops})" + ); + } + + if let (Some((registry, assignment)), NodeType::Agent(a)) = (chain.peers, &node.node_type) + && a.teammates + { + ctx.peer_registry = Some(Arc::clone(registry)); + ctx.peer_assignment = Some(assignment.clone()); + } + + match step(node, state, ctx, step_ctx, current).await? { + StepResult::Continue(targets) => match targets.as_slice() { + [] => { + debug!( + "[graph:{}] map '{map_id}' [{idx}] {current} → END", + step_ctx.graph_name() + ); + Ok(None) + } + [target] => { + if !chain.subgraph.contains(target) { + let mut branch_nodes: Vec<&str> = + chain.subgraph.iter().map(String::as_str).collect(); + branch_nodes.sort_unstable(); + bail!( + "map node '{map_id}': sub-branch [{idx}] routed to '{target}' which is \ + outside the branch subgraph rooted at '{}' (branch nodes: {}). Script \ + `_next` targets inside a map branch must stay within the branch.", + chain.entry, + branch_nodes.join(", ") + ); + } + debug!( + "[graph:{}] map '{map_id}' [{idx}] {current} → {target}", + step_ctx.graph_name() + ); + Ok(Some(target.clone())) + } + many => bail!( + "map node '{map_id}': sub-branch [{idx}] node '{current}' fanned out to {many:?}; \ + a map branch must route to a single node" + ), + }, + StepResult::End(_) => bail!( + "map node '{map_id}': sub-branch [{idx}] reached end node '{current}' inside a map branch" + ), } } @@ -193,36 +323,49 @@ fn resolve_max_concurrency( } } -/// Pre-provision teammate identities before any branch starts, so a fast -/// sibling can message one still waiting on the semaphore (the message -/// queues in the pre-created inbox). A single-item fan-out has no peers, so -/// it gets no registry. +/// Pre-provision one teammate identity per item before any chain starts, so a +/// fast sibling can message one still waiting on the semaphore (the message +/// queues in the pre-created inbox). The identity lives for the item's whole +/// chain; every `teammates: true` agent step in it speaks as that item. A +/// single-item fan-out has no peers, so it gets no registry. fn provision_map_peers( - branch: &NodeType, - branch_id: &str, + graph: &Graph, + subgraph: &HashSet, + entry: &str, item_count: usize, ) -> Option<(Arc, Vec)> { - match branch { - NodeType::Agent(n) if n.teammates && item_count >= 2 => { - let registry = Arc::new(PeerRegistry::new()); - let assignments = (0..item_count) - .map(|i| { - let id = graph_agent_id(&n.agent); - let inbox = Arc::new(Inbox::new()); - registry.insert(id.clone(), format!("{branch_id}[{i}]"), inbox.clone()); - (id, inbox) - }) - .collect(); - Some((registry, assignments)) - } - _ => None, + if item_count < 2 { + return None; } + let flagged_agent = |node_id: &str| match graph.get_node(node_id).map(|n| &n.node_type) { + Some(NodeType::Agent(a)) if a.teammates => Some(a.agent.as_str()), + _ => None, + }; + let agent_name = flagged_agent(entry).or_else(|| { + graph + .nodes + .keys() + .filter(|id| subgraph.contains(*id)) + .find_map(|id| flagged_agent(id)) + })?; + + let registry = Arc::new(PeerRegistry::new()); + let assignments = (0..item_count) + .map(|i| { + let id = graph_agent_id(agent_name); + let inbox = Arc::new(Inbox::new()); + registry.insert(id.clone(), format!("{entry}[{i}]"), inbox.clone()); + (id, inbox) + }) + .collect(); + Some((registry, assignments)) } #[cfg(test)] mod tests { - use super::super::types::{AgentNode, EndNode}; + use super::super::types::{AgentNode, EndNode, GraphSettings, Node}; use super::*; + use indexmap::IndexMap; fn agent_branch(teammates: bool) -> NodeType { NodeType::Agent(AgentNode { @@ -235,10 +378,59 @@ mod tests { }) } + fn graph_with_branch(node_type: NodeType) -> Graph { + let mut nodes: IndexMap = IndexMap::new(); + nodes.insert( + "shards".into(), + Node { + id: "shards".into(), + description: String::new(), + node_type, + next: None, + }, + ); + + Graph { + name: "t".into(), + description: String::new(), + version: "1.0".into(), + model: None, + temperature: None, + top_p: None, + reasoning_effort: None, + max_concurrent_jobs: None, + can_spawn_agents: None, + max_concurrent_agents: None, + max_agent_depth: None, + global_tools: Vec::new(), + mcp_servers: Vec::new(), + mcp_tools: None, + skills_enabled: None, + enabled_skills: None, + inject_skill_instructions: None, + skill_instructions: None, + conversation_starters: Vec::new(), + variables: Vec::new(), + settings: GraphSettings::default(), + initial_state: HashMap::new(), + reducers: HashMap::new(), + start: "shards".into(), + nodes, + } + } + + fn provision( + node_type: NodeType, + n: usize, + ) -> Option<(Arc, Vec)> { + let graph = graph_with_branch(node_type); + let subgraph = HashSet::from(["shards".to_string()]); + provision_map_peers(&graph, &subgraph, "shards", n) + } + #[test] fn provision_map_peers_registers_every_branch() { - let (registry, assignments) = - provision_map_peers(&agent_branch(true), "shards", 3).expect("should provision"); + let (registry, assignments) = provision(agent_branch(true), 3).expect("should provision"); assert_eq!(assignments.len(), 3); let roster = registry.roster(); @@ -255,12 +447,12 @@ mod tests { #[test] fn provision_map_peers_single_item_gets_no_registry() { - assert!(provision_map_peers(&agent_branch(true), "shards", 1).is_none()); + assert!(provision(agent_branch(true), 1).is_none()); } #[test] fn provision_map_peers_unflagged_gets_no_registry() { - assert!(provision_map_peers(&agent_branch(false), "shards", 3).is_none()); + assert!(provision(agent_branch(false), 3).is_none()); } #[test] @@ -269,7 +461,7 @@ mod tests { output: String::new(), state_updates: None, }); - assert!(provision_map_peers(&branch, "shards", 3).is_none()); + assert!(provision(branch, 3).is_none()); } fn map_with_cap(cap: Option) -> MapNode { diff --git a/src/graph/types.rs b/src/graph/types.rs index 39f3143..5d82493 100644 --- a/src/graph/types.rs +++ b/src/graph/types.rs @@ -517,6 +517,10 @@ impl GraphState { self.data.insert(key, value); } + pub fn remove(&mut self, key: &str) -> Option { + self.data.remove(key) + } + pub fn merge(&mut self, json_obj: &serde_json::Map) { for (key, value) in json_obj { self.data.insert(key.clone(), value.clone()); diff --git a/src/graph/validator.rs b/src/graph/validator.rs index 1432d1a..58f2514 100644 --- a/src/graph/validator.rs +++ b/src/graph/validator.rs @@ -862,6 +862,44 @@ fn find_reachable_nodes(graph: &Graph) -> HashSet { reachable } +/// Nodes reachable from `entry` over the edges a chain can follow at run +/// time: `next` targets and script/llm `fallback`s. Approval routes and a +/// nested map's `branch` are not followed — neither may appear inside a +/// branch, and the validator reports them separately. +pub(super) fn branch_subgraph(graph: &Graph, entry: &str) -> HashSet { + let mut reachable: HashSet = HashSet::new(); + let mut queue: VecDeque = VecDeque::new(); + + if !graph.has_node(entry) { + return reachable; + } + + reachable.insert(entry.to_string()); + queue.push_back(entry.to_string()); + + while let Some(id) = queue.pop_front() { + let Some(node) = graph.get_node(&id) else { + continue; + }; + let mut edges: Vec<&String> = node + .next + .as_ref() + .map(|t| t.as_slice().iter().collect()) + .unwrap_or_default(); + match &node.node_type { + NodeType::Script(s) => edges.extend(s.fallback.as_ref()), + NodeType::Llm(l) => edges.extend(l.fallback.as_ref()), + _ => {} + } + for next in edges { + if graph.has_node(next) && reachable.insert(next.clone()) { + queue.push_back(next.clone()); + } + } + } + reachable +} + // v1 parallel-group detection: only the immediate `next` targets of a fan-out node count as a parallel group. Map // branches are handled separately by `validate_map_branches` (the branch's self-parallelism is checked via strict-mode // rules on the branch node itself, not via group membership).