async_nodes/

graph.rs

//! Graph topology: nodes, pins, and declarative edges.
//!
//! A [`BspGraph`] is the compiled blueprint the engine runs: re-executable nodes, edges as
//! *data* (each carrying a declarative [`Condition`]), static inputs, and the consumer's
//! named-condition / merger registries. The fluent builder here is the primary authoring
//! API; [`crate::def::GraphDef`] is the serializable interchange format derived from it.

use std::marker::PhantomData;

use indexmap::IndexMap;
use serde::{Deserialize, Serialize};

use super::{
    condition::{Condition, ConditionMap},
    merge::MergerMap,
    node::BspNode,
    value::BspValue,
};

/// Programmatic node identifier: positional, stable within one graph build.
///
/// The *serializable* definition layer uses stable string ids (see
/// [`crate::def::NodeDef`]); the graph keeps a string label per node
/// ([`BspGraph::node_label`]) so definition round-trips are lossless while the hot path
/// stays on cheap, copyable ids.
pub type NodeId = usize;
/// A port index on a node (dataflow profile).
pub type PortIndex = usize;
/// Edge identifier: the edge's index in declaration order.
pub type EdgeId = usize;

/// An output pin on a specific node.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct OutputPin {
    pub node_id: NodeId,
    pub port: PortIndex,
}

/// An input pin on a specific node.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct InputPin {
    pub node_id: NodeId,
    pub port: PortIndex,
}

/// The value-store port a failing node writes its error to when its recovery policy routes
/// to an error edge (dataflow profile; see
/// [`crate::recovery::RecoveryPolicy::RouteToErrorEdge`]).
pub const ERROR_PORT: PortIndex = usize::MAX;

/// A declarative connection between two nodes — data, not a closure.
///
/// After `source` completes, the engine evaluates `condition`; when it matches, the edge
/// emits an activation token toward `target`. Edges carrying `source_port`/`target_port`
/// additionally route a pin value (dataflow profile).
#[derive(Clone)]
pub struct Edge<S> {
    /// The node this edge starts from.
    pub source: NodeId,
    /// The node this edge points to.
    pub target: NodeId,
    /// The declarative predicate deciding whether the edge fires.
    pub condition: Condition<S>,
    /// Iteration budget. Present (and validated) on cycle-closing edges: exceeding it
    /// raises [`crate::error::BspError::LoopBoundExceeded`].
    pub max_iterations: Option<u64>,
    /// The output port routed by this edge (dataflow profile only).
    pub source_port: Option<PortIndex>,
    /// The input port routed by this edge (dataflow profile only).
    pub target_port: Option<PortIndex>,
}

impl<S> Edge<S> {
    fn control(source: NodeId, target: NodeId) -> Self {
        Self {
            source,
            target,
            condition: Condition::Always,
            max_iterations: None,
            source_port: None,
            target_port: None,
        }
    }
}

/// A legacy non-generic identifier wrapping a designated node with port-link builder logic.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct NodeHandle {
    pub id: NodeId,
}

impl NodeHandle {
    #[must_use]
    pub const fn new(id: NodeId) -> Self {
        Self { id }
    }

    #[must_use]
    pub const fn output(self, port: PortIndex) -> OutputPin {
        OutputPin {
            node_id: self.id,
            port,
        }
    }

    #[must_use]
    pub const fn input(self, port: PortIndex) -> InputPin {
        InputPin {
            node_id: self.id,
            port,
        }
    }
}

/// A compile-time type-safe identifier wrapping a designated node.
/// This allows compiler-checked, named port accessors generated by macro attributes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct TypedNodeHandle<N> {
    pub id: NodeId,
    _marker: PhantomData<fn() -> N>,
}

impl<N> TypedNodeHandle<N> {
    #[must_use]
    pub const fn new(id: NodeId) -> Self {
        Self {
            id,
            _marker: PhantomData,
        }
    }

    /// Fetches an output pin dynamically by index (bypasses macro accessors).
    #[must_use]
    pub const fn get_output_pin(&self, port: PortIndex) -> OutputPin {
        OutputPin {
            node_id: self.id,
            port,
        }
    }

    /// Fetches an input pin dynamically by index (bypasses macro accessors).
    #[must_use]
    pub const fn get_input_pin(&self, port: PortIndex) -> InputPin {
        InputPin {
            node_id: self.id,
            port,
        }
    }

    #[must_use]
    pub const fn as_untyped(&self) -> NodeHandle {
        NodeHandle::new(self.id)
    }
}

impl<N> From<TypedNodeHandle<N>> for NodeHandle {
    fn from(typed: TypedNodeHandle<N>) -> Self {
        Self::new(typed.id)
    }
}

impl<N> serde::Serialize for TypedNodeHandle<N> {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        self.id.serialize(serializer)
    }
}

impl<'de, N> serde::Deserialize<'de> for TypedNodeHandle<N> {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let id = NodeId::deserialize(deserializer)?;
        Ok(Self::new(id))
    }
}

/// Builder handle for configuring a freshly added edge, e.g.
/// `graph.add_edge(a, b).when(Condition::Named("has_more".into())).with_max_iterations(30)`.
pub struct EdgeBuilder<'a, S> {
    graph: &'a mut BspGraph<S>,
    id: EdgeId,
}

impl<S> EdgeBuilder<'_, S> {
    /// Sets the edge's condition.
    pub fn when(self, condition: Condition<S>) -> Self {
        self.graph.edges[self.id].condition = condition;
        self
    }

    /// Sets the edge's iteration budget (expected on cycle-closing edges).
    pub fn with_max_iterations(self, max: u64) -> Self {
        self.graph.edges[self.id].max_iterations = Some(max);
        self
    }

    /// Attaches dataflow ports to the edge (source output port, target input port).
    pub fn with_ports(self, source_port: PortIndex, target_port: PortIndex) -> Self {
        let edge = &mut self.graph.edges[self.id];
        edge.source_port = Some(source_port);
        edge.target_port = Some(target_port);
        self.graph.reindex_edge_ports(self.id);
        self
    }

    /// Returns the edge's declaration index.
    #[must_use]
    pub const fn id(&self) -> EdgeId {
        self.id
    }
}

/// The blueprint definition containing nodes, edges, and initial inputs.
///
/// `S` is the consumer-owned run state the graph's nodes operate on; it defaults to `()`
/// for pure dataflow graphs.
pub struct BspGraph<S = ()> {
    pub(crate) nodes: IndexMap<NodeId, Box<dyn BspNode<S>>>,
    pub(crate) edges: Vec<Edge<S>>,
    pub(crate) static_inputs: IndexMap<InputPin, BspValue>,
    /// Consumer-registered named conditions, resolved during edge evaluation.
    pub(crate) conditions: ConditionMap<S>,
    /// Consumer-registered mergers for join nodes.
    pub(crate) mergers: MergerMap<S>,
    /// Stable string labels per node, emitted as `NodeDef.id` by `to_def`.
    pub(crate) labels: IndexMap<NodeId, String>,
    /// Explicit entry nodes (in addition to all no-incoming-edge source nodes).
    pub(crate) entry: Vec<NodeId>,
    /// Original `NodeDef`s retained for lossless `to_def` round-trips of def-built graphs.
    pub(crate) node_defs: IndexMap<NodeId, crate::def::NodeDef>,
    /// Precomputed incoming edge indices per node to avoid linear scans at runtime.
    pub(crate) incoming: IndexMap<NodeId, Vec<EdgeId>>,
    /// Precomputed dataflow edges per input pin (declaration order preserved).
    pub(crate) incoming_by_pin: IndexMap<InputPin, Vec<EdgeId>>,
    /// Precomputed outgoing edge indices per node (declaration order preserved).
    pub(crate) outgoing: IndexMap<NodeId, Vec<EdgeId>>,
    next_node_id: NodeId,
}

impl<S> Default for BspGraph<S> {
    fn default() -> Self {
        Self::new()
    }
}

impl<S> BspGraph<S> {
    /// Instantiates a new empty BSP execution graph.
    #[must_use]
    pub fn new() -> Self {
        Self {
            nodes: IndexMap::new(),
            edges: Vec::new(),
            static_inputs: IndexMap::new(),
            conditions: ConditionMap::default(),
            mergers: MergerMap::default(),
            labels: IndexMap::new(),
            entry: Vec::new(),
            node_defs: IndexMap::new(),
            incoming: IndexMap::new(),
            incoming_by_pin: IndexMap::new(),
            outgoing: IndexMap::new(),
            next_node_id: 1,
        }
    }

    /// Registers a node with a self-assigned auto-incrementing identifier.
    pub fn add_node<N: BspNode<S> + 'static>(&mut self, node: N) -> TypedNodeHandle<N> {
        let id = self.next_node_id;
        self.next_node_id += 1;
        self.nodes.insert(id, Box::new(node));
        self.labels.insert(id, id.to_string());
        TypedNodeHandle::new(id)
    }

    /// Registers a node within the graph's designated manual identifier namespace.
    pub fn add_node_with_id(&mut self, id: impl Into<NodeId>, node: Box<dyn BspNode<S>>) {
        let id = id.into();
        self.nodes.insert(id, node);
        self.labels.entry(id).or_insert_with(|| id.to_string());
        if id >= self.next_node_id {
            self.next_node_id = id + 1;
        }
    }

    /// Connects an output pin of a source node to an input pin of a target node.
    ///
    /// This is the dataflow profile's wiring primitive: an edge with
    /// [`Condition::Always`] carrying both ports.
    pub fn add_link(&mut self, source: OutputPin, target: InputPin) {
        let id = self.edges.len();
        let mut edge = Edge::control(source.node_id, target.node_id);
        edge.source_port = Some(source.port);
        edge.target_port = Some(target.port);
        self.push_edge(edge);
        debug_assert_eq!(id, self.edges.len() - 1);
    }

    /// Back-compatible helper targeting index indices.
    pub fn add_link_indices(
        &mut self,
        source_node: NodeId,
        source_port: PortIndex,
        target_node: NodeId,
        target_port: PortIndex,
    ) {
        self.add_link(
            OutputPin {
                node_id: source_node,
                port: source_port,
            },
            InputPin {
                node_id: target_node,
                port: target_port,
            },
        );
    }

    /// Adds a control-flow edge between two nodes, returning a builder for the condition
    /// and iteration budget. The default condition is [`Condition::Always`].
    pub fn add_edge(
        &mut self,
        source: impl Into<NodeId>,
        target: impl Into<NodeId>,
    ) -> EdgeBuilder<'_, S> {
        let id = self.edges.len();
        self.push_edge(Edge::control(source.into(), target.into()));
        EdgeBuilder { graph: self, id }
    }

    /// Adds a control-flow edge with an explicit condition in one call.
    pub fn add_edge_with(
        &mut self,
        source: impl Into<NodeId>,
        target: impl Into<NodeId>,
        condition: Condition<S>,
    ) -> EdgeId {
        let id = self.edges.len();
        let mut edge = Edge::control(source.into(), target.into());
        edge.condition = condition;
        self.push_edge(edge);
        id
    }

    /// Injects an initial static parameter into a specific node's input pin.
    pub fn set_static_input<T: std::any::Any + Send + Sync>(&mut self, target: InputPin, value: T) {
        self.static_inputs.insert(target, BspValue::new(value));
    }

    /// Registers a named condition predicate (the serializable routing mechanism).
    pub fn register_condition(
        &mut self,
        name: impl Into<String>,
        f: impl Fn(&S, &crate::run::ValueStore) -> bool + Send + Sync + 'static,
    ) {
        self.conditions.insert(name.into(), std::sync::Arc::new(f));
    }

    /// Registers a named merger used by join nodes to fold branch states.
    pub fn register_merger<M: crate::merge::Merger<S> + 'static>(
        &mut self,
        name: impl Into<String>,
        merger: M,
    ) {
        self.mergers
            .insert(name.into(), std::sync::Arc::new(merger));
    }

    /// Marks a node as an explicit entry (in addition to no-incoming-edge sources).
    pub fn add_entry(&mut self, id: impl Into<NodeId>) {
        self.entry.push(id.into());
    }

    /// The stable string label for a node (its [`crate::def::NodeDef`] id).
    #[must_use]
    pub fn node_label(&self, id: NodeId) -> Option<&str> {
        self.labels.get(&id).map(String::as_str)
    }

    /// The nodes of this graph, in declaration order.
    pub fn node_ids(&self) -> impl Iterator<Item = NodeId> + '_ {
        self.nodes.keys().copied()
    }

    /// The edges of this graph, in declaration order.
    #[must_use]
    pub fn edges(&self) -> &[Edge<S>] {
        &self.edges
    }

    /// The initial frontier: explicit entries plus every node with no incoming edges.
    pub(crate) fn entries(&self) -> indexmap::IndexSet<NodeId> {
        let mut entries: indexmap::IndexSet<NodeId> = indexmap::IndexSet::new();
        for &id in &self.entry {
            entries.insert(id);
        }
        for &id in self.nodes.keys() {
            if self.incoming.get(&id).is_none_or(Vec::is_empty) {
                entries.insert(id);
            }
        }
        entries
    }

    /// Appends an edge and maintains the lookup indexes. Edge ids are declaration indices.
    pub(crate) fn push_edge(&mut self, edge: Edge<S>) {
        let id = self.edges.len();
        if let Some(target_port) = edge.target_port {
            self.incoming_by_pin
                .entry(InputPin {
                    node_id: edge.target,
                    port: target_port,
                })
                .or_default()
                .push(id);
        }
        self.incoming.entry(edge.target).or_default().push(id);
        self.outgoing.entry(edge.source).or_default().push(id);
        self.edges.push(edge);
    }

    /// Re-indexes an edge whose ports were attached after insertion.
    fn reindex_edge_ports(&mut self, id: EdgeId) {
        let edge = &self.edges[id];
        if let Some(target_port) = edge.target_port {
            let pin = InputPin {
                node_id: edge.target,
                port: target_port,
            };
            let list = self.incoming_by_pin.entry(pin).or_default();
            if !list.contains(&id) {
                list.push(id);
            }
        }
    }

    // =========================================================================
    // BUILDER PATTERN EXTENSIONS
    // =========================================================================

    #[must_use]
    pub fn with_node_id(mut self, id: impl Into<NodeId>, node: Box<dyn BspNode<S>>) -> Self {
        self.add_node_with_id(id, node);
        self
    }

    #[must_use]
    pub fn with_link(mut self, source: OutputPin, target: InputPin) -> Self {
        self.add_link(source, target);
        self
    }

    #[must_use]
    pub fn with_static_input<T: std::any::Any + Send + Sync>(
        mut self,
        target: InputPin,
        value: T,
    ) -> Self {
        self.set_static_input(target, value);
        self
    }

    #[must_use]
    pub fn with_condition(
        mut self,
        name: impl Into<String>,
        f: impl Fn(&S, &crate::run::ValueStore) -> bool + Send + Sync + 'static,
    ) -> Self {
        self.register_condition(name, f);
        self
    }

    #[must_use]
    pub fn with_merger<M: crate::merge::Merger<S> + 'static>(
        mut self,
        name: impl Into<String>,
        merger: M,
    ) -> Self {
        self.register_merger(name, merger);
        self
    }
}