async_nodes/

def.rs

//! Serializable graph definitions — the GUI / n8n-enabler.
//!
//! A [`GraphDef`] fully describes a graph as data: nodes (with stable string ids, opaque
//! consumer-owned config blobs, and inert GUI layout positions), edges (with declarative
//! conditions and loop budgets), and explicit entries. The fluent [`BspGraph`] builder
//! remains the *authoring* API; definitions are the *interchange* format, and
//! `def -> graph -> def` is a fixed point (modulo
//! [`Custom`](crate::condition::Condition::Custom) closures, which serialize as their
//! label only and fail closed at instantiation unless the label is re-registered as a
//! named condition).
//!
//! `version` has existed since the first serialized file: bump [`GRAPH_DEF_VERSION`] on
//! any breaking format change so old files are never unmarked.

use std::collections::HashMap;

use async_trait::async_trait;
use serde::{Deserialize, Serialize};

use super::{
    BspContext,
    condition::Condition,
    error::BspError,
    graph::{BspGraph, Edge, InputPin, NodeId},
    node::{Activation, BspNode, NodeOutcome},
    recovery::{Recovery, RecoveryPolicy, RetryPolicy},
    registry::NodeRegistry,
    run::RunContext,
    value::BspValue,
};

/// Projects a static-input [`BspValue`] down to JSON for [`GraphDef`] serialization.
///
/// Covers the JSON-native types (`bool` / `i64` / `u64` / `f64` / `String` /
/// `serde_json::Value`) — the exact set the default value factory produces, so
/// `def -> graph -> def` is a fixed point for static inputs. Anything else returns
/// `None` and is skipped by [`BspGraph::to_def`] (surfaced as a validation warning).
pub(crate) fn value_to_json(value: &BspValue) -> Option<serde_json::Value> {
    if let Some(v) = value.downcast_ref::<serde_json::Value>() {
        Some(v.clone())
    } else if let Some(b) = value.downcast_ref::<bool>() {
        Some(serde_json::Value::from(*b))
    } else if let Some(i) = value.downcast_ref::<i64>() {
        Some(serde_json::Value::from(*i))
    } else if let Some(u) = value.downcast_ref::<u64>() {
        Some(serde_json::Value::from(*u))
    } else if let Some(f) = value.downcast_ref::<f64>() {
        // NaN/Inf have no JSON representation; skip rather than emit `null`.
        serde_json::Number::from_f64(*f).map(serde_json::Value::Number)
    } else {
        value
            .downcast_ref::<String>()
            .map(|s| serde_json::Value::from(s.clone()))
    }
}

/// The current [`GraphDef`] schema version. Bump on any breaking format change.
pub const GRAPH_DEF_VERSION: u32 = 1;

/// A serializable static input: an initial pin value injected before execution.
///
/// Values are JSON. Instantiation maps them back to typed [`BspValue`]s through the
/// registry's value factory ([`crate::registry::default_value_factory`] by default).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct StaticInputDef {
    /// Target node id (the stable string label).
    pub node: String,
    /// Target input port.
    pub port: usize,
    /// The value as JSON.
    pub value: serde_json::Value,
}

/// A fully serializable graph description.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct GraphDef {
    /// Schema version — present from day one so old files are never unmarked.
    pub version: u32,
    /// The node definitions, in declaration order.
    pub nodes: Vec<NodeDef>,
    /// The edge definitions, in declaration order.
    pub edges: Vec<EdgeDef>,
    /// Explicit entry node ids (in addition to no-incoming-edge sources).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub entry: Vec<String>,
    /// Static pin inputs, in declaration order.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub static_inputs: Vec<StaticInputDef>,
}

impl GraphDef {
    /// Parses a graph definition from JSON.
    pub fn from_json(json: &str) -> Result<Self, BspError> {
        serde_json::from_str(json).map_err(|e| BspError::DefError(e.to_string()))
    }

    /// Serializes this definition to pretty JSON.
    pub fn to_json(&self) -> Result<String, BspError> {
        serde_json::to_string_pretty(self).map_err(|e| BspError::DefError(e.to_string()))
    }
}

/// A serializable node definition.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct NodeDef {
    /// Stable string id — survives graph editing (deletion, copy-paste) untouched.
    pub id: String,
    /// Registry key, e.g. `"my-crate.generate"`.
    pub type_name: String,
    /// Node-specific configuration blob; the schema is owned (and versioned) by the
    /// consumer's node type. The engine never looks inside.
    #[serde(default, skip_serializing_if = "serde_json::Value::is_null")]
    pub config: serde_json::Value,
    /// Activation override; absent means "the node type's own default".
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub activation: Option<ActivationDef>,
    /// Recovery override; absent means "the node type's own policy". A `retry` here
    /// reuses the node's own classifier when it has one, and otherwise retries every
    /// failure — prefer registering classification inside the node factory.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub recovery: Option<RecoveryDef>,
    /// Optional named merger used when this node joins branch states.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub merger: Option<String>,
    /// Optional per-run execution cap (poll-until circuit breaker).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_executions: Option<u64>,
    /// Inert GUI layout hint; the engine ignores it and round-trips it faithfully so the
    /// GUI never loses a human's arrangement.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub position: Option<(f64, f64)>,
}

/// Serializable activation policy: `"any"` (control-flow default) or `"all"` (join).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ActivationDef {
    /// Fire on the first pending token.
    Any,
    /// Fire when every incoming edge holds a pending token.
    All,
}

/// Serializable recovery policy: `"fail_fast"`, `"error_edge"`, or
/// `{"retry": {...}}` (backoff parameters only — classification never serializes).
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RecoveryDef {
    /// First error aborts the run.
    FailFast,
    /// Re-run with backoff on failure.
    Retry {
        /// Retry budget after the initial failure.
        #[serde(default = "default_max_retries")]
        max_retries: u32,
        /// Initial backoff delay, in milliseconds.
        #[serde(default = "default_base_delay_ms")]
        base_delay_ms: u64,
        /// Backoff ceiling, in milliseconds.
        #[serde(default = "default_max_delay_ms")]
        max_delay_ms: u64,
        /// Symmetric jitter factor (e.g. `0.25` = ±25%).
        #[serde(default = "default_jitter")]
        jitter: f64,
    },
    /// Route failures to `on_error` edges instead of aborting.
    ErrorEdge,
}

const fn default_max_retries() -> u32 {
    3
}
const fn default_base_delay_ms() -> u64 {
    500
}
const fn default_max_delay_ms() -> u64 {
    8_000
}
const fn default_jitter() -> f64 {
    0.25
}

/// The serializable condition subset.
///
/// Unit variants serialize as plain strings (`"always"`), compound variants as single-key
/// objects (`{"named": "..."}`, `{"and": [...]}`). `Custom` keeps only its label and fails
/// closed at instantiation unless the label is registered.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum ConditionDef {
    /// Always fires.
    #[default]
    Always,
    /// Never fires.
    Never,
    /// Fires when the source execution carried a `BspSkip` marker.
    OnSkip,
    /// Fires when the source node failed and routes to error edges.
    OnError,
    /// A consumer-registered named predicate.
    Named(String),
    /// Conjunction.
    And(Vec<ConditionDef>),
    /// Disjunction.
    Or(Vec<ConditionDef>),
    /// Negation.
    Not(Box<ConditionDef>),
    /// An opaque custom predicate placeholder: label only, closure never serializes.
    Custom {
        /// The re-registration key.
        label: String,
    },
}

/// A serializable edge definition.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct EdgeDef {
    /// Source node id.
    pub from: String,
    /// Target node id.
    pub to: String,
    /// The edge condition; defaults to `"always"`.
    #[serde(default)]
    pub condition: ConditionDef,
    /// Iteration budget (expected on cycle-closing edges).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_iterations: Option<u64>,
    /// Routed output port (dataflow profile only).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source_port: Option<usize>,
    /// Routed input port (dataflow profile only).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub target_port: Option<usize>,
}

impl ConditionDef {
    /// Instantiates a runtime condition, resolving `Custom` labels against the registry.
    fn to_condition<S>(
        &self,
        registry: &NodeRegistry<S>,
        unknown: &mut Vec<String>,
    ) -> Condition<S> {
        match self {
            Self::Always => Condition::Always,
            Self::Never => Condition::Never,
            Self::OnSkip => Condition::OnSkip,
            Self::OnError => Condition::OnError,
            Self::Named(name) => {
                if !registry.conditions.contains_key(name) {
                    unknown.push(format!("condition '{name}'"));
                }
                Condition::Named(name.clone().into())
            }
            Self::And(conds) => Condition::And(
                conds
                    .iter()
                    .map(|c| c.to_condition(registry, unknown))
                    .collect(),
            ),
            Self::Or(conds) => Condition::Or(
                conds
                    .iter()
                    .map(|c| c.to_condition(registry, unknown))
                    .collect(),
            ),
            Self::Not(cond) => Condition::Not(Box::new(cond.to_condition(registry, unknown))),
            Self::Custom { label } => match registry.conditions.get(label) {
                // Fail closed unless the label was re-registered as a named condition.
                Some(f) => Condition::Custom {
                    label: label.clone(),
                    f: f.clone(),
                },
                None => {
                    unknown.push(format!(
                        "custom condition '{label}' (re-register it to instantiate)"
                    ));
                    Condition::Never
                }
            },
        }
    }

    /// Projects a runtime condition down to its serializable form.
    fn from_condition<S>(condition: &Condition<S>) -> Self {
        match condition {
            Condition::Always => Self::Always,
            Condition::Never => Self::Never,
            Condition::OnSkip => Self::OnSkip,
            Condition::OnError => Self::OnError,
            Condition::Named(id) => Self::Named(id.0.clone()),
            Condition::And(conds) => Self::And(conds.iter().map(Self::from_condition).collect()),
            Condition::Or(conds) => Self::Or(conds.iter().map(Self::from_condition).collect()),
            Condition::Not(cond) => Self::Not(Box::new(Self::from_condition(cond))),
            Condition::Custom { label, .. } => Self::Custom {
                label: label.clone(),
            },
        }
    }
}

/// A node instantiated from a [`NodeDef`]: delegates to the factory-built node and applies
/// the definition's overrides (activation, recovery, merger, execution cap).
struct DefNode<S> {
    inner: Box<dyn BspNode<S>>,
    type_name: String,
    activation: Option<Activation>,
    recovery: Option<RecoveryPolicy>,
    merger: Option<String>,
    max_executions: Option<u64>,
}

#[async_trait]
impl<S: Send + Sync> BspNode<S> for DefNode<S> {
    fn name(&self) -> &str {
        self.inner.name()
    }

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

    fn input_ports(&self) -> Vec<super::graph::PortIndex> {
        self.inner.input_ports()
    }

    fn output_ports(&self) -> Vec<super::graph::PortIndex> {
        self.inner.output_ports()
    }

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

    fn activation(&self) -> Activation {
        self.activation.unwrap_or_else(|| self.inner.activation())
    }

    fn recovery(&self) -> RecoveryPolicy {
        match &self.recovery {
            Some(policy) => policy.clone(),
            None => self.inner.recovery(),
        }
    }

    fn merger(&self) -> Option<&str> {
        self.merger.as_deref().or_else(|| self.inner.merger())
    }

    fn max_executions(&self) -> Option<u64> {
        self.max_executions.or_else(|| self.inner.max_executions())
    }

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

impl From<ActivationDef> for Activation {
    fn from(def: ActivationDef) -> Self {
        match def {
            ActivationDef::Any => Activation::Any,
            ActivationDef::All => Activation::All,
        }
    }
}

impl From<Activation> for ActivationDef {
    fn from(activation: Activation) -> Self {
        match activation {
            Activation::Any => ActivationDef::Any,
            Activation::All => ActivationDef::All,
        }
    }
}

impl RecoveryDef {
    /// Converts to the runtime retry policy parameters.
    fn to_retry_policy(self) -> Option<RetryPolicy> {
        match self {
            Self::Retry {
                max_retries,
                base_delay_ms,
                max_delay_ms,
                jitter,
            } => Some(RetryPolicy {
                max_retries,
                base_delay: std::time::Duration::from_millis(base_delay_ms),
                max_delay: std::time::Duration::from_millis(max_delay_ms),
                jitter,
            }),
            _ => None,
        }
    }
}

impl<S> BspGraph<S> {
    /// Instantiates a graph from a serializable definition and a consumer registry.
    ///
    /// Registration is fail-fast: any unknown node `type_name`, condition name, custom
    /// label, or merger name produces a [`BspError::ValidationError`] listing *all*
    /// unknown names — never a mid-run surprise. Node and edge ids are mapped from the
    /// definition's stable strings onto positional [`NodeId`]s, with the strings retained
    /// as labels so [`BspGraph::to_def`] round-trips losslessly.
    pub fn from_def(
        def: &GraphDef,
        registry: &NodeRegistry<S>,
        ctx: &BspContext,
    ) -> Result<Self, BspError>
    where
        S: Send + Sync + 'static,
    {
        if def.version != GRAPH_DEF_VERSION {
            return Err(BspError::DefError(format!(
                "unsupported graph definition version {} (this crate understands {GRAPH_DEF_VERSION})",
                def.version
            )));
        }

        // Registry completeness: collect *all* unknown names before failing.
        let mut unknown: Vec<String> = Vec::new();
        for node_def in &def.nodes {
            if !registry.factories.contains_key(&node_def.type_name) {
                unknown.push(format!("node type '{}'", node_def.type_name));
            }
            if let Some(merger) = &node_def.merger
                && !registry.mergers.contains_key(merger)
            {
                unknown.push(format!("merger '{merger}'"));
            }
        }

        // Map stable string ids onto positional node ids (declaration order).
        let mut id_map: HashMap<&str, NodeId> = HashMap::new();
        for (idx, node_def) in def.nodes.iter().enumerate() {
            let id = idx + 1;
            if id_map.insert(node_def.id.as_str(), id).is_some() {
                return Err(BspError::ValidationError(format!(
                    "duplicate node id '{}' in graph definition",
                    node_def.id
                )));
            }
        }

        // Convert edges (conditions validate against the registry as they convert).
        let mut edges: Vec<Edge<S>> = Vec::with_capacity(def.edges.len());
        for edge_def in &def.edges {
            let source = id_map.get(edge_def.from.as_str()).copied().ok_or_else(|| {
                BspError::ValidationError(format!(
                    "edge references unknown source node '{}'",
                    edge_def.from
                ))
            })?;
            let target = id_map.get(edge_def.to.as_str()).copied().ok_or_else(|| {
                BspError::ValidationError(format!(
                    "edge references unknown target node '{}'",
                    edge_def.to
                ))
            })?;
            edges.push(Edge {
                source,
                target,
                condition: edge_def.condition.to_condition(registry, &mut unknown),
                max_iterations: edge_def.max_iterations,
                source_port: edge_def.source_port,
                target_port: edge_def.target_port,
            });
        }

        if !unknown.is_empty() {
            return Err(BspError::ValidationError(format!(
                "graph definition references unregistered names: {}",
                unknown.join(", ")
            )));
        }

        let mut graph = BspGraph::new();
        graph.conditions = registry.conditions.clone();
        graph.mergers = registry.mergers.clone();

        for (idx, node_def) in def.nodes.iter().enumerate() {
            let id = idx + 1;
            let factory = registry
                .factories
                .get(&node_def.type_name)
                .expect("completeness checked above");
            let inner = factory(node_def, ctx)?;

            // A def-level retry reuses the node's own classifier when it has one;
            // otherwise it retries every failure (documented on `NodeDef::recovery`).
            let recovery = match node_def.recovery {
                None => None,
                Some(RecoveryDef::FailFast) => Some(RecoveryPolicy::FailFast),
                Some(RecoveryDef::ErrorEdge) => Some(RecoveryPolicy::RouteToErrorEdge),
                Some(def @ RecoveryDef::Retry { .. }) => {
                    let policy = def.to_retry_policy().expect("retry variant");
                    let classify = match inner.recovery() {
                        RecoveryPolicy::Retry { classify, .. } => classify,
                        _ => std::sync::Arc::new(|_: &BspError| Recovery::Retry { after: None }),
                    };
                    Some(RecoveryPolicy::Retry { policy, classify })
                }
            };

            graph.add_node_with_id(
                id,
                Box::new(DefNode {
                    inner,
                    type_name: node_def.type_name.clone(),
                    activation: node_def.activation.map(Activation::from),
                    recovery,
                    merger: node_def.merger.clone(),
                    max_executions: node_def.max_executions,
                }),
            );
            graph.labels.insert(id, node_def.id.clone());
            graph.node_defs.insert(id, node_def.clone());
        }

        for edge in edges {
            graph.push_edge(edge);
        }

        for entry in &def.entry {
            let id = id_map.get(entry.as_str()).copied().ok_or_else(|| {
                BspError::ValidationError(format!("entry references unknown node '{entry}'"))
            })?;
            graph.add_entry(id);
        }

        // Static inputs: JSON values become typed pin values through the registry's
        // value factory (see `registry::default_value_factory`).
        for static_input in &def.static_inputs {
            let id = id_map
                .get(static_input.node.as_str())
                .copied()
                .ok_or_else(|| {
                    BspError::ValidationError(format!(
                        "static input references unknown node '{}'",
                        static_input.node
                    ))
                })?;
            graph.static_inputs.insert(
                InputPin {
                    node_id: id,
                    port: static_input.port,
                },
                (registry.value_factory)(&static_input.value),
            );
        }

        Ok(graph)
    }

    /// Projects this graph into a serializable definition.
    ///
    /// Def-built graphs re-emit their retained [`NodeDef`]s exactly, so
    /// `def -> graph -> def` is a fixed point. Programmatically built graphs synthesize
    /// node definitions from the live nodes (stable label ids, `type_name`s, current
    /// activation / recovery / merger / execution caps, null configs).
    #[must_use]
    pub fn to_def(&self) -> GraphDef {
        let nodes = self
            .nodes
            .iter()
            .map(|(id, node)| {
                if let Some(retained) = self.node_defs.get(id) {
                    return retained.clone();
                }
                NodeDef {
                    id: self
                        .labels
                        .get(id)
                        .cloned()
                        .unwrap_or_else(|| id.to_string()),
                    type_name: node.type_name().to_string(),
                    config: serde_json::Value::Null,
                    activation: Some(node.activation().into()),
                    recovery: Some(match node.recovery() {
                        RecoveryPolicy::FailFast => RecoveryDef::FailFast,
                        RecoveryPolicy::RouteToErrorEdge => RecoveryDef::ErrorEdge,
                        RecoveryPolicy::Retry { policy, .. } => RecoveryDef::Retry {
                            max_retries: policy.max_retries,
                            base_delay_ms: u64::try_from(policy.base_delay.as_millis())
                                .unwrap_or(u64::MAX),
                            max_delay_ms: u64::try_from(policy.max_delay.as_millis())
                                .unwrap_or(u64::MAX),
                            jitter: policy.jitter,
                        },
                    }),
                    merger: node.merger().map(str::to_string),
                    max_executions: node.max_executions(),
                    position: None,
                }
            })
            .collect();

        let edges = self
            .edges
            .iter()
            .map(|edge| EdgeDef {
                from: self
                    .labels
                    .get(&edge.source)
                    .cloned()
                    .unwrap_or_else(|| edge.source.to_string()),
                to: self
                    .labels
                    .get(&edge.target)
                    .cloned()
                    .unwrap_or_else(|| edge.target.to_string()),
                condition: ConditionDef::from_condition(&edge.condition),
                max_iterations: edge.max_iterations,
                source_port: edge.source_port,
                target_port: edge.target_port,
            })
            .collect();

        let entry = self
            .entry
            .iter()
            .map(|id| {
                self.labels
                    .get(id)
                    .cloned()
                    .unwrap_or_else(|| id.to_string())
            })
            .collect();

        // Static inputs: emit the JSON-representable ones; anything else is skipped
        // here and surfaced as a warning by `validate_with_report`.
        let static_inputs = self
            .static_inputs
            .iter()
            .filter_map(|(pin, value)| {
                value_to_json(value).map(|json| StaticInputDef {
                    node: self
                        .labels
                        .get(&pin.node_id)
                        .cloned()
                        .unwrap_or_else(|| pin.node_id.to_string()),
                    port: pin.port,
                    value: json,
                })
            })
            .collect();

        GraphDef {
            version: GRAPH_DEF_VERSION,
            nodes,
            edges,
            entry,
            static_inputs,
        }
    }
}

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

    #[test]
    fn test_condition_def_json_shapes() {
        assert_eq!(
            serde_json::to_string(&ConditionDef::Always).unwrap(),
            "\"always\""
        );
        assert_eq!(
            serde_json::to_string(&ConditionDef::Named("has_more".to_string())).unwrap(),
            "{\"named\":\"has_more\"}"
        );
        assert_eq!(
            serde_json::to_string(&ConditionDef::Not(Box::new(ConditionDef::OnError))).unwrap(),
            "{\"not\":\"on_error\"}"
        );
        assert_eq!(
            serde_json::to_string(&ConditionDef::Custom {
                label: "hand_rolled".to_string()
            })
            .unwrap(),
            "{\"custom\":{\"label\":\"hand_rolled\"}}"
        );
    }

    #[test]
    fn test_activation_and_recovery_def_json_shapes() {
        assert_eq!(
            serde_json::to_string(&ActivationDef::All).unwrap(),
            "\"all\""
        );
        assert_eq!(
            serde_json::to_string(&RecoveryDef::FailFast).unwrap(),
            "\"fail_fast\""
        );
        let retry = RecoveryDef::Retry {
            max_retries: 2,
            base_delay_ms: 100,
            max_delay_ms: 1_000,
            jitter: 0.1,
        };
        let json = serde_json::to_string(&retry).unwrap();
        assert!(json.starts_with("{\"retry\":{"));
        // Missing retry fields fall back to the shared defaults.
        let parsed: RecoveryDef =
            serde_json::from_str("{\"retry\": {\"max_retries\": 1}}").unwrap();
        assert_eq!(
            parsed.to_retry_policy().unwrap().base_delay,
            std::time::Duration::from_millis(500)
        );
    }

    #[test]
    fn test_graph_def_version_roundtrip_and_mismatch() {
        let json = r#"{"version": 1, "nodes": [], "edges": []}"#;
        let def = GraphDef::from_json(json).unwrap();
        assert_eq!(def.version, GRAPH_DEF_VERSION);
        assert!(def.entry.is_empty());

        let registry: NodeRegistry<()> = NodeRegistry::new();
        let ctx = BspContext::new();
        let result = BspGraph::from_def(
            &GraphDef {
                version: 99,
                nodes: Vec::new(),
                edges: Vec::new(),
                entry: Vec::new(),
                static_inputs: Vec::new(),
            },
            &registry,
            &ctx,
        );
        assert!(matches!(result, Err(BspError::DefError(_))));
    }
}