async_nodes/

node.rs

//! The node contract: re-executable units over consumer state `S`.
//!
//! Nodes are the consumer's computation units; the engine owns topology, scheduling, and
//! state movement. A node **never names its successors** — it reads its gathered inputs,
//! mutates [`RunContext::state`], stages outputs, and the engine decides what happens next
//! by evaluating the outgoing edges. A node may run zero, one, or many times per run.

use async_trait::async_trait;
use indexmap::IndexMap;

use super::{
    BspContext,
    error::BspError,
    graph::{NodeId, PortIndex},
    recovery::RecoveryPolicy,
    run::RunContext,
    value::BspValue,
};

/// How a node joins a frontier, i.e. how its pending activation tokens are counted.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Activation {
    /// Fire as soon as at least one incoming edge holds a pending token (control-flow
    /// default). Exactly one token is consumed per firing (from the lowest-declared
    /// pending edge), so co-pending tokens re-fire the node on later supersteps — an
    /// `Any`-join fires once per arriving branch.
    #[default]
    Any,
    /// Fire only when *every* incoming edge holds ≥1 pending token (dataflow default).
    /// One token per incoming edge is consumed.
    ///
    /// **Beware inside loops:** an `All`-join on a cycle is the classic self-deadlock —
    /// one predecessor's token arrives once per iteration while the other's arrives once
    /// ever, so the join waits forever. Joins inside loops are almost always `Any` or
    /// merger-joins; validation warns on this shape.
    All,
}

/// Defines the evaluated outcome of a single node's execution turn.
#[derive(Debug)]
pub enum NodeOutcome {
    /// Node executed successfully, returning values mapped to specific output ports.
    /// Uses `IndexMap` to guarantee deterministic iteration order.
    Success(IndexMap<PortIndex, BspValue>),

    /// Node requested execution pause (e.g., waiting for external input).
    Suspend {
        /// Arbitrary context parameters passed back to the host application.
        payload: serde_json::Value,
        /// Explicit indices of inputs that must be provided to resume.
        waiting_inputs: Vec<PortIndex>,
    },
}

/// The programmatic contract for writing execution nodes.
///
/// `S` is the consumer-owned run state: the engine is generic over it and never inspects
/// it except through explicitly defined lenses (conditions, mergers). Implementors get
/// control-flow defaults; the `#[node]` / `#[node_blocking]` macros generate dataflow
/// profile nodes (`Activation::All`, named ports) over this same trait.
#[async_trait]
pub trait BspNode<S>: Send + Sync {
    /// Semantic name used for logs, interceptors, and the GUI.
    fn name(&self) -> &str;

    /// Registry type name used for (de)serialization, e.g. `"my-crate.generate"`.
    /// Defaults to [`BspNode::name`]; override for stable cross-version identity.
    fn type_name(&self) -> &str {
        self.name()
    }

    /// Dataflow profile: the unique indices of input ports required for execution
    /// (empty for pure control-flow nodes).
    fn input_ports(&self) -> Vec<PortIndex> {
        Vec::new()
    }

    /// Dataflow profile: the unique indices of output ports produced upon success.
    fn output_ports(&self) -> Vec<PortIndex> {
        Vec::new()
    }

    /// Indicates whether output ports are evaluated dynamically at runtime.
    fn is_dynamic_output(&self) -> bool {
        false
    }

    /// How this node joins a frontier. Default: [`Activation::Any`] (control-flow).
    fn activation(&self) -> Activation {
        Activation::Any
    }

    /// Per-node recovery policy. Default: fail-fast (first error aborts the run).
    fn recovery(&self) -> RecoveryPolicy {
        RecoveryPolicy::FailFast
    }

    /// Optional name of a consumer-registered [`crate::merge::Merger`] used when this
    /// node joins concurrently produced branch states at a superstep barrier.
    fn merger(&self) -> Option<&str> {
        None
    }

    /// Optional circuit breaker on how many times this node may execute per run
    /// (poll-until style nodes).
    fn max_executions(&self) -> Option<u64> {
        None
    }

    /// Core computation. Mutates state and the value store through `run`; the engine
    /// decides what happens next from the edges — nodes never name their successors.
    async fn execute(&self, run: &mut RunContext<'_, S>) -> Result<NodeOutcome, BspError>;
}

#[async_trait]
impl<S: Send> BspNode<S> for Box<dyn BspNode<S>> {
    fn name(&self) -> &str {
        (**self).name()
    }

    fn type_name(&self) -> &str {
        (**self).type_name()
    }

    fn input_ports(&self) -> Vec<PortIndex> {
        (**self).input_ports()
    }

    fn output_ports(&self) -> Vec<PortIndex> {
        (**self).output_ports()
    }

    fn is_dynamic_output(&self) -> bool {
        (**self).is_dynamic_output()
    }

    fn activation(&self) -> Activation {
        (**self).activation()
    }

    fn recovery(&self) -> RecoveryPolicy {
        (**self).recovery()
    }

    fn merger(&self) -> Option<&str> {
        (**self).merger()
    }

    fn max_executions(&self) -> Option<u64> {
        (**self).max_executions()
    }

    async fn execute(&self, run: &mut RunContext<'_, S>) -> Result<NodeOutcome, BspError> {
        (**self).execute(run).await
    }
}

/// The dataflow-profile authoring trait: plain-Rust node authoring without macros.
///
/// The `#[node]` / `#[node_blocking]` macros generate this same shape under the hood;
/// implement `DataflowNode` directly when you want the classic
/// `(node_id, inputs, ctx) -> outputs` signature with no macro magic — e.g. when a
/// registry or GUI bridge (like the snarl one) constructs nodes from descriptors.
///
/// Every `DataflowNode` automatically implements [`BspNode`] over any consumer state
/// `S`, with join semantics ([`Activation::All`]) and outputs packed into
/// [`NodeOutcome::Success`]. The two authoring styles compose freely in one graph.
///
/// Note: implement *either* `DataflowNode` *or* [`BspNode`] for a type, never both —
/// the blanket impl would collide.
#[async_trait]
pub trait DataflowNode: Send + Sync {
    /// Semantic name used for logs, interceptors, and the GUI.
    fn name(&self) -> &str;

    /// The unique indices of input ports required for execution.
    fn input_ports(&self) -> Vec<PortIndex>;

    /// The unique indices of output ports produced upon success.
    fn output_ports(&self) -> Vec<PortIndex>;

    /// Indicates whether output ports are evaluated dynamically at runtime.
    fn is_dynamic_output(&self) -> bool {
        false
    }

    /// Core computation over the gathered pin inputs.
    async fn execute(
        &self,
        node_id: NodeId,
        inputs: IndexMap<PortIndex, BspValue>,
        ctx: &BspContext,
    ) -> Result<IndexMap<PortIndex, BspValue>, BspError>;
}

#[async_trait]
impl<S: Send, T: DataflowNode> BspNode<S> for T {
    fn name(&self) -> &str {
        DataflowNode::name(self)
    }

    fn type_name(&self) -> &str {
        DataflowNode::name(self)
    }

    fn input_ports(&self) -> Vec<PortIndex> {
        DataflowNode::input_ports(self)
    }

    fn output_ports(&self) -> Vec<PortIndex> {
        DataflowNode::output_ports(self)
    }

    fn is_dynamic_output(&self) -> bool {
        DataflowNode::is_dynamic_output(self)
    }

    fn activation(&self) -> Activation {
        Activation::All
    }

    async fn execute(&self, run: &mut RunContext<'_, S>) -> Result<NodeOutcome, BspError> {
        let inputs = std::mem::take(&mut run.inputs);
        let outputs = DataflowNode::execute(self, run.node_id, inputs, run.ctx).await?;
        Ok(NodeOutcome::Success(outputs))
    }
}