async_nodes/

engine.rs

//! The dynamic scheduler: token-based activation over BSP supersteps.
//!
//! Edge evaluation emits *tokens* onto edges; a node joins the next frontier when its
//! [`Activation`] policy is satisfiable from pending (unconsumed) tokens, and activation
//! consumes them. This is process-network/Kahn-style firing, and it makes loops fall out
//! naturally: a back-edge simply delivers another token to an earlier node, which refires.
//! There is no "executed once" set and no special loop construct.
//!
//! # Circuit breakers
//!
//! Loops make runaway execution possible by construction, so bounds are part of the core:
//!
//! - a global [`BspEngine::with_max_steps`] superstep cap producing
//!   [`EngineResult::StepLimitReached`] — a first-class result carrying the live run
//!   state, never a silent salvage;
//! - per-edge `max_iterations` budgets raising [`BspError::LoopBoundExceeded`];
//! - optional per-node `max_executions` caps;
//! - a pending-token cap per edge ([`BspEngine::with_max_pending_tokens`]) against floods.
//!
//! # Joins inside loops are almost always `Any` or merger-joins
//!
//! An `All`-join on a cycle is the classic self-deadlock — one predecessor's token
//! arrives once per iteration, the other's arrives once ever, so the join waits forever
//! and the run ends in [`BspError::Deadlock`]:
//!
//! ```text
//!        entry ──▶ J (All) ──▶ body
//!                    ▲            │
//!                    └────────────┘   J needs a token on BOTH edges per firing,
//!                      back-edge      but entry only ever fires once → stall
//! ```
//!
//! Validation warns on this shape; write the loop join as `Any` (or a merger-join).

use std::{
    collections::HashSet,
    sync::{Arc, OnceLock},
    time::{Duration, Instant},
};

use futures_util::stream::{self, StreamExt};
use indexmap::{IndexMap, IndexSet};

use super::{
    BspContext,
    condition::{Condition, ConditionEval},
    error::BspError,
    events::BspInterceptor,
    graph::{BspGraph, ERROR_PORT, EdgeId, InputPin, NodeId, OutputPin, PortIndex},
    merge::Merger,
    node::{Activation, BspNode, NodeOutcome},
    recovery::{Recovery, RecoveryPolicy},
    run::{EngineResult, PendingSuspension, RunContext, RunState, SuspendedExecution, ValueStore},
    value::{BspSkip, BspValue},
};

/// The default global superstep budget (circuit breaker). The pre-rewrite engine had no
/// cap; loops make one mandatory, so a generous default is baked in — tune it with
/// [`BspEngine::with_max_steps`].
pub const DEFAULT_MAX_STEPS: u64 = 1024;

/// The default cap on pending (unconsumed) tokens per edge (token-flood guard).
pub const DEFAULT_MAX_PENDING_TOKENS: u64 = 1024;

/// A transform applied to each branch-state clone at multi-node frontier formation,
/// right after the clone is taken from the trunk (see
/// [`BspEngine::with_branch_transform`]). The `usize` is the branch's ordinal within
/// the frontier formation (declaration order, 0-based).
pub type BranchTransform<S> = Arc<dyn Fn(&mut S, usize) + Send + Sync>;

/// The outcome of one executed (or bypassed) node task.
#[derive(Debug)]
enum TaskResult {
    /// The node finished; its outputs were written and its edges should be evaluated.
    Success {
        outputs: IndexMap<PortIndex, BspValue>,
        skipped: bool,
    },
    /// The node requested suspension.
    Suspended {
        payload: serde_json::Value,
        waiting_inputs: Vec<PortIndex>,
    },
    /// The node failed and its recovery policy routed the failure to error edges.
    RoutedError(BspError),
}

/// What the scheduler decided at the head of a superstep.
enum NextStep {
    /// Execute this frontier.
    Frontier(Vec<NodeId>),
    /// The frontier emptied with no pending tokens: natural completion.
    Complete,
    /// The global superstep budget is exhausted.
    StepLimit,
}

/// The result of graph validation: fatal errors plus non-fatal findings.
///
/// [`BspEngine::validate`] stays error-only; use [`BspEngine::validate_with_report`] to
/// also see warnings (unbounded cycles, unreachable nodes, dead ends, missing mergers…).
#[derive(Debug, Clone)]
pub struct ValidationReport {
    /// The first fatal validation error, if any.
    pub error: Option<BspError>,
    /// Non-fatal findings, in analysis order.
    pub warnings: Vec<String>,
}

impl ValidationReport {
    /// Returns whether validation found no fatal error.
    #[must_use]
    pub const fn is_ok(&self) -> bool {
        self.error.is_none()
    }
}

/// The dynamic scheduler that organizes nodes into supersteps and executes them
/// concurrently (bounded), re-executing nodes as loops deliver new activation tokens.
///
/// `S` is the consumer-owned run state; it defaults to `()` for pure dataflow graphs.
pub struct BspEngine<S = ()> {
    graph: BspGraph<S>,
    ctx: BspContext,
    interceptor: Option<Arc<dyn BspInterceptor>>,
    max_concurrency: Option<usize>,
    max_steps: u64,
    max_pending_tokens: u64,
    branch_transform: Option<BranchTransform<S>>,
    validation_cache: OnceLock<ValidationReport>,
}

impl<S> BspEngine<S> {
    /// Instantiates a new BSP engine using a compiled graph layout.
    #[must_use]
    pub fn new(graph: BspGraph<S>) -> Self {
        Self {
            graph,
            ctx: BspContext::new(),
            interceptor: None,
            max_concurrency: None,
            max_steps: DEFAULT_MAX_STEPS,
            max_pending_tokens: DEFAULT_MAX_PENDING_TOKENS,
            branch_transform: None,
            validation_cache: OnceLock::new(),
        }
    }

    /// Attaches a shared execution context for dependency injection.
    #[must_use]
    pub fn with_context(mut self, ctx: BspContext) -> Self {
        self.ctx = ctx;
        self
    }

    /// Returns a reference to the active execution context.
    #[must_use]
    pub const fn context(&self) -> &BspContext {
        &self.ctx
    }

    /// The graph this engine runs.
    #[must_use]
    pub const fn graph(&self) -> &BspGraph<S> {
        &self.graph
    }

    /// Attaches a performance and audit logging interceptor.
    #[must_use]
    pub fn with_interceptor(mut self, interceptor: Arc<dyn BspInterceptor>) -> Self {
        self.interceptor = Some(interceptor);
        self
    }

    /// Configures the maximum concurrency (throttling limit) for task execution within a
    /// superstep.
    ///
    /// The throttle is scoped to this engine instance (`buffer_unordered` per superstep);
    /// a nested graph-as-node runs a child engine with its own independent throttle, so
    /// nesting can never deadlock or halve capacity per level.
    #[must_use]
    pub const fn with_max_concurrency(mut self, limit: usize) -> Self {
        self.max_concurrency = Some(limit);
        self
    }

    /// Configures the global superstep budget. Exceeding it yields
    /// [`EngineResult::StepLimitReached`] with the live run state — never a silent salvage.
    #[must_use]
    pub const fn with_max_steps(mut self, steps: u64) -> Self {
        self.max_steps = steps;
        self
    }

    /// Configures the per-edge pending-token cap (token-flood guard).
    #[must_use]
    pub const fn with_max_pending_tokens(mut self, cap: u64) -> Self {
        self.max_pending_tokens = cap;
        self
    }

    /// Attaches a transform applied to each branch state clone at multi-node frontier
    /// formation, right after the clone is taken from the trunk.
    ///
    /// The second argument is the branch's *ordinal within the frontier formation*
    /// (declaration order, 0-based). This is the hook for fork-fan-out semantics where
    /// sibling clones must not all inherit the full trunk — the reference consumer
    /// (`llm-net`) zeroes branch metrics for every sibling after the first so boundary
    /// rollups never double-count the parent's history.
    ///
    /// The default (`None`) is the identity: every branch is an exact clone. The
    /// transform never runs for single-node frontiers (they borrow the trunk directly)
    /// or on the sequential path (`run_sequential` does no branch cloning).
    #[must_use]
    pub fn with_branch_transform(mut self, transform: BranchTransform<S>) -> Self {
        self.branch_transform = Some(transform);
        self
    }

    /// Validates the graph (errors only), looking for unfireable cycles, missing links,
    /// invalid port configurations, and unregistered condition/merger references.
    pub fn validate(&self) -> Result<(), BspError> {
        let report = self.validation_report();
        match &report.error {
            Some(err) => Err(err.clone()),
            None => Ok(()),
        }
    }

    /// Validates the graph, additionally returning non-fatal warnings (loop-aware
    /// analysis: unbounded cycles, unreachable nodes, dead ends, `All`-joins on cycles,
    /// joins without mergers).
    pub fn validate_with_report(&self) -> ValidationReport {
        self.validation_report().clone()
    }

    fn validation_report(&self) -> &ValidationReport {
        self.validation_cache
            .get_or_init(|| compute_validation(&self.graph))
    }

    // =========================================================================
    // SCHEDULER PRIMITIVES
    // =========================================================================

    /// Computes the next frontier: forced nodes (entries / resumed suspensions) plus every
    /// node whose activation policy is satisfiable from pending tokens.
    fn compute_frontier(&self, state: &RunState<S>, forced: IndexSet<NodeId>) -> IndexSet<NodeId> {
        let mut frontier = forced;
        for (&edge_id, &count) in &state.pending_tokens {
            if count == 0 {
                continue;
            }
            let target = self.graph.edges[edge_id].target;
            if frontier.contains(&target) {
                continue;
            }
            let node = &self.graph.nodes[&target];
            if self.is_activatable(state, target, node.activation()) {
                frontier.insert(target);
            }
        }
        frontier
    }

    /// Assembles the next superstep's frontier, detecting the terminal conditions
    /// (natural completion, deadlock, step-limit) and firing superstep events.
    fn next_step(
        &self,
        state: &mut RunState<S>,
        forced: &mut IndexSet<NodeId>,
    ) -> Result<NextStep, BspError> {
        let frontier = self.compute_frontier(state, std::mem::take(forced));

        if frontier.is_empty() {
            if state.pending_tokens.values().any(|&c| c > 0) {
                tracing::error!("❌ [BspEngine] Deadlock detected during scheduling.");
                return Err(BspError::Deadlock);
            }
            return Ok(NextStep::Complete);
        }

        if state.step >= self.max_steps {
            tracing::error!(
                "🛑 [BspEngine] Superstep budget of {} exhausted; returning live state.",
                self.max_steps
            );
            return Ok(NextStep::StepLimit);
        }

        state.frontier = frontier.iter().copied().collect();
        // Declaration order (auto-increment ids track insertion order), so barrier folds
        // and event streams are deterministic regardless of token-emission interleavings.
        let mut frontier_nodes: Vec<NodeId> = frontier.into_iter().collect();
        frontier_nodes.sort_unstable();

        if let Some(interceptor) = &self.interceptor {
            interceptor.on_superstep_start(state.step, &frontier_nodes);
        }
        tracing::debug!(
            "🌀 [BspEngine] Superstep {}: Launching {} concurrent tasks...",
            state.step,
            frontier_nodes.len()
        );
        Ok(NextStep::Frontier(frontier_nodes))
    }

    /// Whether a node's activation policy is satisfiable from pending tokens.
    fn is_activatable(&self, state: &RunState<S>, node_id: NodeId, activation: Activation) -> bool {
        let Some(incoming) = self.graph.incoming.get(&node_id) else {
            return false;
        };
        match activation {
            Activation::Any => incoming
                .iter()
                .any(|e| state.pending_tokens.get(e).is_some_and(|&c| c > 0)),
            Activation::All => {
                !incoming.is_empty()
                    && incoming
                        .iter()
                        .all(|e| state.pending_tokens.get(e).is_some_and(|&c| c > 0))
            }
        }
    }

    /// Consumes the tokens backing a node's activation: exactly one token (from the
    /// lowest-declared incoming edge holding any) for `Any` — "fire on the first pending
    /// token" — and exactly one per incoming edge for `All` (Kahn-style firing).
    fn consume_tokens(&self, state: &mut RunState<S>, node_id: NodeId, activation: Activation) {
        let Some(incoming) = self.graph.incoming.get(&node_id) else {
            return;
        };
        match activation {
            Activation::Any => {
                let selected = incoming
                    .iter()
                    .find(|e| state.pending_tokens.get(*e).is_some_and(|&c| c > 0))
                    .copied();
                if let Some(e) = selected
                    && let Some(count) = state.pending_tokens.get_mut(&e)
                {
                    *count = count.saturating_sub(1);
                    if *count == 0 {
                        state.pending_tokens.swap_remove(&e);
                    }
                }
            }
            Activation::All => {
                let mut emptied = Vec::new();
                for &e in incoming {
                    if let Some(count) = state.pending_tokens.get_mut(&e) {
                        *count = count.saturating_sub(1);
                        if *count == 0 {
                            emptied.push(e);
                        }
                    }
                }
                for e in emptied {
                    state.pending_tokens.swap_remove(&e);
                }
            }
        }
    }

    /// Prepares one frontier node for execution: enforces its execution cap, consumes its
    /// activation tokens, and gathers its inputs.
    fn prepare_node(
        &self,
        state: &mut RunState<S>,
        node_id: NodeId,
    ) -> Result<(u64, IndexMap<PortIndex, BspValue>, bool), BspError> {
        let node = &self.graph.nodes[&node_id];
        let exec_base = state.execution_count(node_id);
        if let Some(max) = node.max_executions()
            && exec_base >= max
        {
            return Err(BspError::MaxExecutionsExceeded { node_id, max });
        }
        self.consume_tokens(state, node_id, node.activation());
        let (inputs, should_skip) = self.gather_node_inputs(node_id, node.as_ref(), state);
        Ok((exec_base, inputs, should_skip))
    }

    /// Gathers input values for a node from static inputs, injected inputs, and wired
    /// edges (declaration order preserved — multi-wire ports aggregate into a `Vec`).
    fn gather_node_inputs(
        &self,
        node_id: NodeId,
        node: &dyn BspNode<S>,
        state: &RunState<S>,
    ) -> (IndexMap<PortIndex, BspValue>, bool) {
        let mut inputs = IndexMap::new();
        let mut should_skip = false;

        for &port in &node.input_ports() {
            let pin = InputPin { node_id, port };
            let mut gathered_values = Vec::new();

            // 1. Static inputs
            if let Some(static_val) = self.graph.static_inputs.get(&pin) {
                gathered_values.push(static_val.clone());
            }

            // 2. Injected inputs (from resume)
            if let Some(injected_val) = state.values.injected.get(&pin) {
                gathered_values.push(injected_val.clone());
            }

            // 3. Wired inputs from completed source nodes (latest value store writes)
            if let Some(edge_ids) = self.graph.incoming_by_pin.get(&pin) {
                for &edge_id in edge_ids {
                    let edge = &self.graph.edges[edge_id];
                    let source_pin = OutputPin {
                        node_id: edge.source,
                        port: edge.source_port.unwrap_or(0),
                    };
                    if let Some(val) = state.values.outputs.get(&source_pin) {
                        gathered_values.push(val.clone());
                    }
                }
            }

            // Dynamic Control-Flow: Detect if any incoming path carried a Skip marker
            if gathered_values.iter().any(BspValue::is::<BspSkip>) {
                should_skip = true;
            }

            if !gathered_values.is_empty() {
                if gathered_values.len() == 1 {
                    inputs.insert(port, gathered_values[0].clone());
                } else {
                    inputs.insert(port, BspValue::new(gathered_values));
                }
            }
        }

        (inputs, should_skip)
    }

    /// Emits activation tokens onto every outgoing edge of `source` whose condition
    /// matches, enforcing iteration budgets and token caps and firing edge events.
    ///
    /// Borrows are split field-wise so both the trunk-aliasing paths and the barrier path
    /// (owned branch states) can call it.
    #[allow(clippy::too_many_arguments)]
    fn emit_edges(
        &self,
        source: NodeId,
        branch: &S,
        values: &ValueStore,
        pending_tokens: &mut IndexMap<EdgeId, u64>,
        iterations: &mut IndexMap<EdgeId, u64>,
        source_skipped: bool,
        source_errored: bool,
    ) -> Result<(), BspError> {
        let Some(outgoing) = self.graph.outgoing.get(&source) else {
            return Ok(());
        };
        for &edge_id in outgoing {
            let edge = &self.graph.edges[edge_id];
            let eval = ConditionEval {
                state: branch,
                values,
                conditions: &self.graph.conditions,
                source_skipped,
                source_errored,
            };
            if !edge.condition.eval(&eval) {
                continue;
            }

            if let Some(max) = edge.max_iterations {
                let iteration = iterations.entry(edge_id).or_insert(0);
                *iteration += 1;
                let iteration = *iteration;
                if let Some(interceptor) = &self.interceptor {
                    interceptor.on_loop_iteration(edge_id, iteration, Some(max));
                }
                if iteration > max {
                    return Err(BspError::LoopBoundExceeded {
                        edge: edge_id,
                        from: edge.source,
                        to: edge.target,
                        max,
                    });
                }
            }

            let pending = pending_tokens.entry(edge_id).or_insert(0);
            if *pending >= self.max_pending_tokens {
                return Err(BspError::TokenLimitExceeded {
                    edge: edge_id,
                    from: edge.source,
                    to: edge.target,
                    cap: self.max_pending_tokens,
                });
            }
            *pending += 1;

            if let Some(interceptor) = &self.interceptor {
                interceptor.on_edge_taken(edge.source, edge.target, &edge.condition.label());
            }
        }
        Ok(())
    }

    /// Resolves the merger for a multi-task barrier fold: the first (in frontier
    /// declaration order) completed node whose outgoing edges reach a join naming a
    /// registered merger.
    fn resolve_merger(&self, frontier_nodes: &[NodeId]) -> Option<Arc<dyn Merger<S>>> {
        for &node_id in frontier_nodes {
            if let Some(outgoing) = self.graph.outgoing.get(&node_id) {
                for &edge_id in outgoing {
                    let target = self.graph.edges[edge_id].target;
                    if let Some(node) = self.graph.nodes.get(&target)
                        && let Some(name) = node.merger()
                        && let Some(merger) = self.graph.mergers.get(name)
                    {
                        return Some(merger.clone());
                    }
                }
            }
        }
        None
    }

    /// Executes one node's task: skip-bypass, then `execute` wrapped in the recovery loop.
    ///
    /// The task borrows the branch state it operates on, so the concurrent path can give
    /// every frontier task its own clone while single-task frontiers borrow the trunk.
    /// Returns the number of `execute` invocations (including retries) and the outcome.
    async fn run_task(
        &self,
        node_id: NodeId,
        branch: &mut S,
        inputs: IndexMap<PortIndex, BspValue>,
        should_skip: bool,
        exec_base: u64,
    ) -> (u64, Result<TaskResult, BspError>) {
        let node = self
            .graph
            .nodes
            .get(&node_id)
            .expect("frontier nodes are graph members");
        let name = node.name().to_string();

        if let Some(interceptor) = &self.interceptor {
            interceptor.on_node_start(node_id, &name);
        }

        if should_skip {
            // Dynamic Control-Flow: bypass execution completely, propagating Skips
            let mut outputs = IndexMap::new();
            for &port in &node.output_ports() {
                outputs.insert(port, BspValue::new(BspSkip));
            }
            if let Some(interceptor) = &self.interceptor {
                interceptor.on_node_complete(node_id, &name, Duration::ZERO);
            }
            return (
                1,
                Ok(TaskResult::Success {
                    outputs,
                    skipped: true,
                }),
            );
        }

        let recovery = node.recovery();
        let start = Instant::now();
        let mut calls = 0_u32;

        loop {
            calls += 1;
            let (result, staged, skip_flag) = {
                let mut run = RunContext::new(
                    &mut *branch,
                    node_id,
                    &self.ctx,
                    exec_base + u64::from(calls) - 1,
                    inputs.clone(),
                );
                let result = node.execute(&mut run).await;
                let (staged, skip_flag) = run.take_staged();
                (result, staged, skip_flag)
            };

            match result {
                Ok(NodeOutcome::Success(mut map)) => {
                    // Staged outputs merge under the returned map (explicit return wins).
                    for (port, value) in staged {
                        map.entry(port).or_insert(value);
                    }
                    let (outputs, skipped) = if skip_flag {
                        let mut outputs = IndexMap::new();
                        for &port in &node.output_ports() {
                            outputs.insert(port, BspValue::new(BspSkip));
                        }
                        (outputs, true)
                    } else {
                        let skipped = map.values().any(BspValue::is::<BspSkip>);
                        (map, skipped)
                    };
                    let duration = start.elapsed();
                    if let Some(interceptor) = &self.interceptor {
                        interceptor.on_node_complete(node_id, &name, duration);
                    }
                    return (
                        u64::from(calls),
                        Ok(TaskResult::Success { outputs, skipped }),
                    );
                }
                Ok(NodeOutcome::Suspend {
                    payload,
                    waiting_inputs,
                }) => {
                    return (
                        u64::from(calls),
                        Ok(TaskResult::Suspended {
                            payload,
                            waiting_inputs,
                        }),
                    );
                }
                Err(err) => match &recovery {
                    RecoveryPolicy::Retry { policy, classify } => match classify(&err) {
                        Recovery::Retry { after } if calls <= policy.max_retries => {
                            let wait = policy
                                .backoff(calls - 1)
                                .max(after.unwrap_or(Duration::ZERO));
                            if let Some(interceptor) = &self.interceptor {
                                interceptor.on_node_retry(node_id, calls, wait, &err);
                            }
                            super::recovery::sleep(wait).await;
                        }
                        _ => {
                            if let Some(interceptor) = &self.interceptor {
                                interceptor.on_node_error(node_id, &name, &err);
                            }
                            return (u64::from(calls), Err(err));
                        }
                    },
                    RecoveryPolicy::RouteToErrorEdge => {
                        if let Some(interceptor) = &self.interceptor {
                            interceptor.on_node_error(node_id, &name, &err);
                        }
                        return (u64::from(calls), Ok(TaskResult::RoutedError(err)));
                    }
                    RecoveryPolicy::FailFast => {
                        if let Some(interceptor) = &self.interceptor {
                            interceptor.on_node_error(node_id, &name, &err);
                        }
                        return (u64::from(calls), Err(err));
                    }
                },
            }
        }
    }

    /// Applies one task's outcome: writes outputs, evaluates edges, folds the branch state
    /// into the trunk.
    ///
    /// With `branch == None` the task already mutated the trunk in place (single-task and
    /// sequential paths); otherwise the owned branch folds in — via the barrier's merger
    /// when one is named, deterministically last-writer-wins otherwise. Suspended branches
    /// fold too: staged state mutations must survive the suspension.
    fn process_result(
        &self,
        state: &mut RunState<S>,
        node_id: NodeId,
        branch: Option<(S, Option<&dyn Merger<S>>)>,
        result: Result<TaskResult, BspError>,
        suspensions: &mut Vec<PendingSuspension>,
    ) -> Result<(), BspError> {
        let RunState {
            user,
            values,
            pending_tokens,
            iterations,
            errors,
            ..
        } = state;

        match result {
            Ok(TaskResult::Success { outputs, skipped }) => {
                for (port, value) in outputs {
                    values.outputs.insert(OutputPin { node_id, port }, value);
                }
                let branch_state: &S = branch.as_ref().map_or(&*user, |(b, _)| b);
                self.emit_edges(
                    node_id,
                    branch_state,
                    values,
                    pending_tokens,
                    iterations,
                    skipped,
                    false,
                )?;
            }
            Ok(TaskResult::Suspended {
                payload,
                waiting_inputs,
            }) => {
                suspensions.push(PendingSuspension {
                    node_id,
                    waiting_inputs,
                    payload,
                });
            }
            Ok(TaskResult::RoutedError(err)) => {
                errors.insert(node_id, err.clone());
                values.outputs.insert(
                    OutputPin {
                        node_id,
                        port: ERROR_PORT,
                    },
                    BspValue::new(err),
                );
                let branch_state: &S = branch.as_ref().map_or(&*user, |(b, _)| b);
                self.emit_edges(
                    node_id,
                    branch_state,
                    values,
                    pending_tokens,
                    iterations,
                    false,
                    true,
                )?;
            }
            Err(err) => return Err(err),
        }

        if let Some((branch_state, merger)) = branch {
            match merger {
                Some(merger) => merger.merge(user, branch_state),
                None => *user = branch_state,
            }
        }
        Ok(())
    }

    /// Packages a suspension if the superstep produced one, otherwise advances the step
    /// counter. Shared by both execution loops.
    fn epilogue(&self, state: &mut RunState<S>, suspensions: &[PendingSuspension]) -> bool {
        if suspensions.is_empty() {
            state.step += 1;
            return false;
        }
        tracing::warn!(
            "⏸️ [BspEngine] {} node(s) requested execution suspension.",
            suspensions.len()
        );
        if let Some(interceptor) = &self.interceptor {
            interceptor.on_suspended(suspensions);
        }
        state.frontier = suspensions.iter().map(|s| s.node_id).collect();
        true
    }

    /// The Bulk Synchronous Parallel execution loop.
    ///
    /// `forced` nodes join the first frontier unconditionally: graph entries on a fresh
    /// run, suspended nodes on a resume (their activation tokens were consumed when they
    /// first fired, so resumption re-arms them directly).
    async fn execute_loop(
        &self,
        mut state: RunState<S>,
        forced: IndexSet<NodeId>,
    ) -> Result<EngineResult<S>, BspError>
    where
        S: Clone,
    {
        if let Some(err) = &self.validation_report().error {
            return Err(err.clone());
        }

        let mut forced = forced;

        loop {
            let frontier_nodes = match self.next_step(&mut state, &mut forced)? {
                NextStep::Frontier(nodes) => nodes,
                NextStep::Complete => {
                    tracing::info!("✅ [BspEngine] Graph executed to completion.");
                    if let Some(interceptor) = &self.interceptor {
                        interceptor.on_complete(state.step);
                    }
                    return Ok(EngineResult::Complete(Box::new(state)));
                }
                NextStep::StepLimit => return Ok(EngineResult::StepLimitReached(Box::new(state))),
            };

            let mut suspensions = Vec::new();

            if frontier_nodes.len() == 1 {
                // Single-node frontier: borrow the trunk directly — linear chains and
                // simple loops never pay for a state clone.
                let node_id = frontier_nodes[0];
                let (exec_base, inputs, should_skip) = self.prepare_node(&mut state, node_id)?;
                let (calls, result) = self
                    .run_task(node_id, &mut state.user, inputs, should_skip, exec_base)
                    .await;
                state.executions.insert(node_id, exec_base + calls);
                self.process_result(&mut state, node_id, None, result, &mut suspensions)?;
            } else {
                // Multi-node frontier: each task gets its own clone of the trunk state
                // (Rust aliasing forbids N concurrent `&mut S`); branch states fold back
                // at the barrier in frontier declaration order — never completion order.
                let merger = self.resolve_merger(&frontier_nodes);
                if merger.is_none() {
                    tracing::warn!(
                        "🔀 [BspEngine] Folding {} branch states without a named merger: last-writer-wins.",
                        frontier_nodes.len()
                    );
                }

                let mut prepared = Vec::with_capacity(frontier_nodes.len());
                for (ordinal, &node_id) in frontier_nodes.iter().enumerate() {
                    let (exec_base, inputs, should_skip) =
                        self.prepare_node(&mut state, node_id)?;
                    prepared.push((ordinal, node_id, exec_base, inputs, should_skip));
                }

                let limit = match self.max_concurrency {
                    Some(0) | None => prepared.len(),
                    Some(n) => n,
                };
                let tasks = prepared.into_iter().map(
                    |(ordinal, node_id, exec_base, inputs, should_skip)| {
                        let mut branch = state.user.clone();
                        if let Some(transform) = &self.branch_transform {
                            transform(&mut branch, ordinal);
                        }
                        async move {
                            let (calls, result) = self
                                .run_task(node_id, &mut branch, inputs, should_skip, exec_base)
                                .await;
                            (ordinal, node_id, exec_base, calls, branch, result)
                        }
                    },
                );
                // `buffer_unordered` yields in *completion* order, but branch results must
                // fold in *declaration* order — each task carries its frontier ordinal and
                // results are re-sorted before folding.
                let mut results = stream::iter(tasks)
                    .buffer_unordered(limit)
                    .collect::<Vec<_>>()
                    .await;
                results.sort_by_key(|(ordinal, ..)| *ordinal);

                for (_, node_id, exec_base, calls, branch, result) in results {
                    state.executions.insert(node_id, exec_base + calls);
                    self.process_result(
                        &mut state,
                        node_id,
                        Some((branch, merger.as_deref())),
                        result,
                        &mut suspensions,
                    )?;
                }

                // Barrier fold complete: give the merger its one-shot finalization hook
                // (all-at-once reducers collapse the accumulated branches here).
                if let Some(merger) = &merger {
                    merger.finish(&mut state.user).await?;
                }
            }

            if self.epilogue(&mut state, &suspensions) {
                return Ok(EngineResult::Suspended(Box::new(SuspendedExecution {
                    suspensions,
                    state,
                })));
            }
        }
    }

    /// The sequential execution loop for `!Clone` states: frontier tasks are awaited one
    /// at a time in declaration order, each borrowing the trunk state directly. Linear and
    /// loop graphs behave exactly as in the concurrent loop; multi-node frontiers run
    /// serially instead of in isolated clones (no `Clone` bound, no branch merging).
    async fn execute_loop_sequential(
        &self,
        mut state: RunState<S>,
        forced: IndexSet<NodeId>,
    ) -> Result<EngineResult<S>, BspError> {
        if let Some(err) = &self.validation_report().error {
            return Err(err.clone());
        }

        let mut forced = forced;

        loop {
            let frontier_nodes = match self.next_step(&mut state, &mut forced)? {
                NextStep::Frontier(nodes) => nodes,
                NextStep::Complete => {
                    tracing::info!("✅ [BspEngine] Graph executed to completion.");
                    if let Some(interceptor) = &self.interceptor {
                        interceptor.on_complete(state.step);
                    }
                    return Ok(EngineResult::Complete(Box::new(state)));
                }
                NextStep::StepLimit => return Ok(EngineResult::StepLimitReached(Box::new(state))),
            };

            let mut suspensions = Vec::new();
            for node_id in frontier_nodes {
                let (exec_base, inputs, should_skip) = self.prepare_node(&mut state, node_id)?;
                let (calls, result) = self
                    .run_task(node_id, &mut state.user, inputs, should_skip, exec_base)
                    .await;
                state.executions.insert(node_id, exec_base + calls);
                self.process_result(&mut state, node_id, None, result, &mut suspensions)?;
            }

            if self.epilogue(&mut state, &suspensions) {
                return Ok(EngineResult::Suspended(Box::new(SuspendedExecution {
                    suspensions,
                    state,
                })));
            }
        }
    }
}

impl<S: Clone> BspEngine<S> {
    /// Initiates a clean run of the graph with the given initial consumer state
    /// (`run(())` for pure dataflow graphs).
    ///
    /// Requires `S: Clone` because concurrent frontiers execute against per-task clones
    /// (see the module docs); single-node frontiers borrow the trunk directly. For
    /// `!Clone` states, use [`BspEngine::run_sequential`].
    pub async fn run(&self, state: S) -> Result<EngineResult<S>, BspError> {
        tracing::info!("🏁 [BspEngine] Initiating graph execution pass...");
        let forced = self.graph.entries();
        self.execute_loop(RunState::new(state), forced).await
    }

    /// Like [`BspEngine::run`], but forces the given nodes as the run's entry frontier
    /// instead of the graph's declared entries.
    ///
    /// This is the resumable-control-flow primitive for hosts that checkpoint only the
    /// consumer state `S` (not the engine's [`SuspendedExecution`] snapshot): resuming is
    /// "start a fresh run forced at the recorded cursor node(s)", exactly the model the
    /// reference consumer's `resume_at` graph cursor uses. Forced nodes join the first
    /// frontier unconditionally, even when their incoming edges hold no tokens.
    pub async fn run_with_entries(
        &self,
        state: S,
        entries: impl IntoIterator<Item = NodeId>,
    ) -> Result<EngineResult<S>, BspError> {
        tracing::info!("🏁 [BspEngine] Initiating graph execution pass with forced entries...");
        self.execute_loop(RunState::new(state), entries.into_iter().collect())
            .await
    }

    /// Resumes a suspended graph execution, injecting external values into the waiting
    /// ports. Iteration and execution counters continue — they never reset.
    pub async fn resume(
        &self,
        mut suspended: SuspendedExecution<S>,
        injections: IndexMap<InputPin, BspValue>,
    ) -> Result<EngineResult<S>, BspError> {
        tracing::info!(
            "🔄 [BspEngine] Resuming execution with {} injected inputs...",
            injections.len()
        );
        for (pin, val) in injections {
            suspended.state.values.injected.insert(pin, val);
        }
        let forced = suspended.suspensions.iter().map(|s| s.node_id).collect();
        self.execute_loop(suspended.state, forced).await
    }
}

impl<S> BspEngine<S> {
    /// Like [`BspEngine::run`] but without a `Clone` bound on `S`: frontier tasks
    /// execute sequentially, each borrowing the trunk state. Linear chains and loops
    /// behave identically; parallel frontiers serialize and share the trunk in
    /// declaration order (no isolation, no merging).
    pub async fn run_sequential(&self, state: S) -> Result<EngineResult<S>, BspError> {
        tracing::info!("🏁 [BspEngine] Initiating sequential graph execution pass...");
        let forced = self.graph.entries();
        self.execute_loop_sequential(RunState::new(state), forced)
            .await
    }

    /// Sequential counterpart of [`BspEngine::resume`] for `!Clone` states.
    pub async fn resume_sequential(
        &self,
        mut suspended: SuspendedExecution<S>,
        injections: IndexMap<InputPin, BspValue>,
    ) -> Result<EngineResult<S>, BspError> {
        tracing::info!(
            "🔄 [BspEngine] Resuming sequential execution with {} injected inputs...",
            injections.len()
        );
        for (pin, val) in injections {
            suspended.state.values.injected.insert(pin, val);
        }
        let forced = suspended.suspensions.iter().map(|s| s.node_id).collect();
        self.execute_loop_sequential(suspended.state, forced).await
    }
}

// =========================================================================
// VALIDATION
// =========================================================================

/// Runs the full validation analysis: structural errors first, then loop-aware warnings.
#[allow(clippy::too_many_lines)]
fn compute_validation<S>(graph: &BspGraph<S>) -> ValidationReport {
    let mut warnings = Vec::new();

    macro_rules! fail {
        ($err:expr) => {
            return ValidationReport {
                error: Some($err),
                warnings,
            }
        };
    }

    // -- Structural: duplicate edges --------------------------------------------------
    // Dataflow links with identical pins are always bugs. Control-flow edges may share
    // endpoints legitimately (branch edges with different conditions), so the condition
    // label is part of the identity.
    let mut unique_edges = HashSet::new();
    for edge in &graph.edges {
        let key = (
            edge.source,
            edge.target,
            edge.source_port,
            edge.target_port,
            edge.condition.label(),
        );
        if !unique_edges.insert(key) {
            let fmt_port = |p: Option<PortIndex>| p.map_or("-".to_string(), |p| p.to_string());
            fail!(BspError::ValidationError(format!(
                "Duplicate link detected from Node {} (Port {}) to Node {} (Port {})",
                edge.source,
                fmt_port(edge.source_port),
                edge.target,
                fmt_port(edge.target_port)
            )));
        }
    }

    // -- Structural: edge endpoints and ports ------------------------------------------
    for edge in &graph.edges {
        let Some(source_node) = graph.nodes.get(&edge.source) else {
            fail!(BspError::NodeNotFound {
                node_id: edge.source
            });
        };
        let Some(target_node) = graph.nodes.get(&edge.target) else {
            fail!(BspError::NodeNotFound {
                node_id: edge.target
            });
        };

        if let Some(port) = edge.source_port
            && !source_node.is_dynamic_output()
            && !source_node.output_ports().contains(&port)
        {
            fail!(BspError::ValidationError(format!(
                "Link setup references invalid output port {} on Node {} ('{}')",
                port,
                edge.source,
                source_node.name()
            )));
        }

        if let Some(port) = edge.target_port
            && !target_node.input_ports().contains(&port)
        {
            fail!(BspError::ValidationError(format!(
                "Link setup references invalid input port {} on Node {} ('{}')",
                port,
                edge.target,
                target_node.name()
            )));
        }
    }

    // -- Structural: static inputs ------------------------------------------------------
    for pin in graph.static_inputs.keys() {
        let Some(node) = graph.nodes.get(&pin.node_id) else {
            fail!(BspError::NodeNotFound {
                node_id: pin.node_id
            });
        };
        if !node.input_ports().contains(&pin.port) {
            fail!(BspError::ValidationError(format!(
                "Static input targets invalid port {} on Node {} ('{}')",
                pin.port,
                pin.node_id,
                node.name()
            )));
        }
    }

    // -- Registry completeness (programmatic graphs too: fail fast, never mid-run) ------
    let mut missing: Vec<String> = Vec::new();
    for edge in &graph.edges {
        let mut named = Vec::new();
        edge.condition.collect_named(&mut named);
        for name in named {
            if !graph.conditions.contains_key(&name) {
                missing.push(format!("condition '{name}'"));
            }
        }
    }
    for node in graph.nodes.values() {
        if let Some(merger) = node.merger()
            && !graph.mergers.contains_key(merger)
        {
            missing.push(format!("merger '{merger}'"));
        }
    }
    if !missing.is_empty() {
        missing.sort();
        missing.dedup();
        fail!(BspError::ValidationError(format!(
            "graph references unregistered names: {}",
            missing.join(", ")
        )));
    }

    // -- Topology analysis (Never edges are treated as absent) ---------------------------
    let live_edges: Vec<EdgeId> = graph
        .edges
        .iter()
        .enumerate()
        .filter(|(_, e)| !matches!(e.condition, Condition::Never))
        .map(|(id, _)| id)
        .collect();

    let mut incoming_live: IndexMap<NodeId, Vec<EdgeId>> = IndexMap::new();
    let mut adjacency: IndexMap<NodeId, Vec<NodeId>> = IndexMap::new();
    for &edge_id in &live_edges {
        let edge = &graph.edges[edge_id];
        incoming_live.entry(edge.target).or_default().push(edge_id);
        adjacency.entry(edge.source).or_default().push(edge.target);
    }

    // Can-fire fixpoint: entries fire; a node fires when its activation is reachable.
    let mut fireable: IndexSet<NodeId> = IndexSet::new();
    for &id in &graph.entry {
        fireable.insert(id);
    }
    for &id in graph.nodes.keys() {
        if incoming_live.get(&id).is_none_or(Vec::is_empty) {
            fireable.insert(id);
        }
    }
    loop {
        let mut changed = false;
        for (&id, node) in &graph.nodes {
            if fireable.contains(&id) {
                continue;
            }
            let incoming = &incoming_live[&id];
            let fires = match node.activation() {
                Activation::Any => incoming
                    .iter()
                    .any(|e| fireable.contains(&graph.edges[*e].source)),
                Activation::All => incoming
                    .iter()
                    .all(|e| fireable.contains(&graph.edges[*e].source)),
            };
            if fires {
                fireable.insert(id);
                changed = true;
            }
        }
        if !changed {
            break;
        }
    }

    // Strongly connected components for cycle analysis.
    let sccs = strongly_connected_components(&adjacency, graph.nodes.keys().copied());
    let mut cyclic_nodes: HashSet<NodeId> = HashSet::new();
    for scc in &sccs {
        if scc.len() > 1 {
            cyclic_nodes.extend(scc.iter().copied());
        }
    }
    for &edge_id in &live_edges {
        let edge = &graph.edges[edge_id];
        if edge.source == edge.target {
            cyclic_nodes.insert(edge.source);
        }
    }

    // -- Unfireable nodes: a cycle that can never start is an error ----------------------
    let unfireable: Vec<NodeId> = graph
        .nodes
        .keys()
        .copied()
        .filter(|id| !fireable.contains(id))
        .collect();
    if !unfireable.is_empty() {
        if let Some(&first) = unfireable.iter().find(|id| cyclic_nodes.contains(id)) {
            fail!(BspError::CycleDetected { node_id: first });
        }
        // Unfireable without a cycle (e.g. reachable only through Never edges): warn.
        for id in &unfireable {
            warnings.push(format!(
                "Node {id} ('{}') is unreachable: no live path from any entry",
                graph.nodes[id].name()
            ));
        }
    }

    // -- Loop-aware warnings -------------------------------------------------------------
    for scc in &sccs {
        let is_cycle = scc.len() > 1
            || (scc.len() == 1 && {
                let id = scc[0];
                live_edges.iter().any(|&e| {
                    let edge = &graph.edges[e];
                    edge.source == id && edge.target == id
                })
            });
        if !is_cycle {
            continue;
        }
        let members: HashSet<NodeId> = scc.iter().copied().collect();
        let bounded = live_edges.iter().any(|&e| {
            let edge = &graph.edges[e];
            members.contains(&edge.source)
                && members.contains(&edge.target)
                && edge.max_iterations.is_some()
        });
        if !bounded {
            warnings.push(format!(
                "Cycle involving nodes {scc:?} has no max_iterations budget on any edge; \
                 an unbounded loop only stops at the global max_steps limit"
            ));
        }
        for &id in scc {
            // Starvation needs ≥2 live incoming edges: one arriving once-ever, another
            // once-per-iteration. A single-incoming All node on a cycle cannot starve.
            let incoming_arity = incoming_live.get(&id).map_or(0, Vec::len);
            if incoming_arity >= 2
                && fireable.contains(&id)
                && graph.nodes[&id].activation() == Activation::All
            {
                warnings.push(format!(
                    "Node {id} ('{}') is an All-join on a cycle: tokens on one incoming edge \
                     arrive once per iteration while another's may arrive once ever, starving \
                     the join. Prefer Any or a merger-join inside loops",
                    graph.nodes[&id].name()
                ));
            }
        }
    }

    for (&id, node) in &graph.nodes {
        if let Some(outgoing) = graph.outgoing.get(&id) {
            let live: Vec<EdgeId> = outgoing
                .iter()
                .copied()
                .filter(|e| !matches!(graph.edges[*e].condition, Condition::Never))
                .collect();
            // Dead-end: outgoing live edges exist but none is an unconditional fallback.
            if !live.is_empty()
                && !live
                    .iter()
                    .any(|&e| matches!(graph.edges[e].condition, Condition::Always))
            {
                warnings.push(format!(
                    "Node {id} ('{}') has no Always fallback on its outgoing edges; if no \
                     condition matches the run dead-ends here",
                    node.name()
                ));
            }
        }

        // Error routing declared but no error edge to route to (applies even when the
        // node has no outgoing edges at all).
        if matches!(node.recovery(), RecoveryPolicy::RouteToErrorEdge) {
            let has_error_edge = graph.outgoing.get(&id).is_some_and(|outgoing| {
                outgoing
                    .iter()
                    .any(|&e| matches!(graph.edges[e].condition, Condition::OnError))
            });
            if !has_error_edge {
                warnings.push(format!(
                    "Node {id} ('{}') uses RouteToErrorEdge recovery but has no outgoing \
                     OnError edge; failures will abort the run",
                    node.name()
                ));
            }
        }

        // Join without a merger: branch states fold last-writer-wins at the barrier.
        if let Some(incoming) = graph.incoming.get(&id) {
            let distinct_sources: HashSet<NodeId> =
                incoming.iter().map(|&e| graph.edges[e].source).collect();
            if distinct_sources.len() >= 2 && node.merger().is_none() {
                warnings.push(format!(
                    "Node {id} ('{}') joins multiple branches without a named merger; \
                     concurrent branch states fold last-writer-wins (harmless only if nodes \
                     do not mutate shared state)",
                    node.name()
                ));
            }
        }
    }

    // Static inputs a GraphDef cannot represent are skipped by `to_def` — warn here so
    // the loss is visible before a save/load round-trip silently drops them.
    for (pin, value) in &graph.static_inputs {
        if crate::def::value_to_json(value).is_none()
            && let Some(node) = graph.nodes.get(&pin.node_id)
        {
            warnings.push(format!(
                "Static input on node {} ('{}') port {} is not GraphDef-serializable; \
                 `to_def` will skip it",
                pin.node_id,
                node.name(),
                pin.port
            ));
        }
    }

    ValidationReport {
        error: None,
        warnings,
    }
}

/// Kosaraju's algorithm: strongly connected components over the live-edge adjacency.
fn strongly_connected_components(
    adjacency: &IndexMap<NodeId, Vec<NodeId>>,
    nodes: impl Iterator<Item = NodeId>,
) -> Vec<Vec<NodeId>> {
    let all: Vec<NodeId> = nodes.collect();

    // Pass 1: finishing order via iterative post-order DFS.
    let mut visited: HashSet<NodeId> = HashSet::new();
    let mut finish_order: Vec<NodeId> = Vec::new();
    for &start in &all {
        if visited.contains(&start) {
            continue;
        }
        let mut stack: Vec<(NodeId, usize)> = vec![(start, 0)];
        visited.insert(start);
        while let Some((node, child_idx)) = stack.pop() {
            let neighbors: &[NodeId] = adjacency.get(&node).map_or(&[], Vec::as_slice);
            if child_idx < neighbors.len() {
                stack.push((node, child_idx + 1));
                let next = neighbors[child_idx];
                if visited.insert(next) {
                    stack.push((next, 0));
                }
            } else {
                finish_order.push(node);
            }
        }
    }

    // Pass 2: DFS on the reversed graph in reverse finishing order.
    let mut reverse: IndexMap<NodeId, Vec<NodeId>> = IndexMap::new();
    for (from, targets) in adjacency {
        for &to in targets {
            reverse.entry(to).or_default().push(*from);
        }
    }

    let mut visited: HashSet<NodeId> = HashSet::new();
    let mut sccs = Vec::new();
    for &start in finish_order.iter().rev() {
        if visited.contains(&start) {
            continue;
        }
        let mut scc = Vec::new();
        let mut stack = vec![start];
        visited.insert(start);
        while let Some(node) = stack.pop() {
            scc.push(node);
            if let Some(predecessors) = reverse.get(&node) {
                for &pred in predecessors {
                    if visited.insert(pred) {
                        stack.push(pred);
                    }
                }
            }
        }
        sccs.push(scc);
    }
    sccs
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_sccs_find_cycles() {
        let mut adjacency: IndexMap<NodeId, Vec<NodeId>> = IndexMap::new();
        adjacency.insert(1, vec![2]);
        adjacency.insert(2, vec![1, 3]);
        adjacency.insert(3, vec![4]);
        adjacency.insert(4, vec![3]);
        let sccs = strongly_connected_components(&adjacency, adjacency.keys().copied());
        let mut sizes: Vec<usize> = sccs.iter().map(Vec::len).collect();
        sizes.sort_unstable();
        assert_eq!(sizes, vec![2, 2]);
    }

    #[test]
    fn test_sccs_of_dag_are_singletons() {
        let mut adjacency: IndexMap<NodeId, Vec<NodeId>> = IndexMap::new();
        adjacency.insert(1, vec![2, 3]);
        adjacency.insert(2, vec![4]);
        adjacency.insert(3, vec![4]);
        let sccs = strongly_connected_components(&adjacency, adjacency.keys().copied());
        assert!(sccs.iter().all(|scc| scc.len() == 1));
        assert_eq!(sccs.len(), 4);
    }
}