async-nodes 0.1.0

//! Dataflow-profile integration tests: `BspSkip` propagation, `Condition::OnSkip` routing,
//! declaration-ordered multi-wire aggregation, and static inputs on the new core.

use async_nodes::{
    Activation, BspContext, BspEngine, BspError, BspGraph, BspNode, BspOutputMapExt, BspSkip,
    Condition, DataflowNode, NodeOutcome, RunContext, async_trait::async_trait, node,
};
use indexmap::IndexMap;

// =========================================================================
// MACRO NODES (THE PROFILE'S PUBLIC SURFACE)
// =========================================================================

#[node("Source")]
async fn profile_source() -> f64 {
    5.0
}

/// Emits `Some` only for large inputs; `None` compiles to the `BspSkip` marker.
#[node("MaybePass")]
async fn maybe_pass(val: f64) -> Option<f64> {
    if val > 100.0 { Some(val) } else { None }
}

#[node("Doubled")]
async fn doubled(val: f64) -> f64 {
    val * 2.0
}

#[node("Describe")]
async fn describe(val: Option<f64>) -> String {
    format!("{val:?}")
}

#[node("WordA")]
async fn word_a() -> String {
    "a".to_string()
}

#[node("WordB")]
async fn word_b() -> String {
    "b".to_string()
}

#[node("JoinWords")]
async fn join_words(input: Vec<String>) -> String {
    input.join(",")
}

/// A control-flow observer used with `Condition::OnSkip`.
struct WatchNode;

#[async_trait]
impl BspNode<()> for WatchNode {
    fn name(&self) -> &str {
        "watcher"
    }

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

    async fn execute(&self, _run: &mut RunContext<'_, ()>) -> Result<NodeOutcome, BspError> {
        Ok(NodeOutcome::Success(IndexMap::new()))
    }
}

// =========================================================================
// SKIP PROPAGATION
// =========================================================================

#[tokio::test]
async fn test_bsp_skip_propagates_downstream() {
    let mut graph = BspGraph::new();
    let src = graph.add_node(ProfileSourceNode::new());
    let maybe = graph.add_node(MaybePassNode::new());
    let dbl = graph.add_node(DoubledNode::new());
    let dsc = graph.add_node(DescribeNode::new());
    graph.add_link(src.output(), maybe.val());
    graph.add_link(maybe.output(), dbl.val());
    graph.add_link(dbl.output(), dsc.val());

    let engine = BspEngine::new(graph);
    let async_nodes::EngineResult::Complete(state) = engine.run(()).await.unwrap() else {
        panic!("expected completion");
    };

    // `maybe_pass` returned None -> BspSkip; every downstream node was bypassed and
    // carries the marker on its outputs.
    let skipped: BspSkip = state.get_typed(dbl.output()).unwrap();
    assert_eq!(skipped, BspSkip);
    let skipped: BspSkip = state.get_typed(dsc.output()).unwrap();
    assert_eq!(skipped, BspSkip);
}

#[tokio::test]
async fn test_some_values_flow_normally() {
    let mut graph = BspGraph::new();
    let maybe = graph.add_node(MaybePassNode::new());
    let dbl = graph.add_node(DoubledNode::new());
    graph.set_static_input(maybe.val(), 500.0);
    graph.add_link(maybe.output(), dbl.val());

    let engine = BspEngine::new(graph);
    let async_nodes::EngineResult::Complete(state) = engine.run(()).await.unwrap() else {
        panic!("expected completion");
    };
    let result: f64 = state.get_typed(dbl.output()).unwrap();
    assert!((result - 1000.0).abs() < f64::EPSILON);
}

#[tokio::test]
async fn test_on_skip_condition_routes_skipped_executions() {
    let build = |maybe_val: f64| {
        let mut graph = BspGraph::new();
        let maybe = graph.add_node(MaybePassNode::new());
        let watcher = graph.add_node(WatchNode);
        graph.set_static_input(maybe.val(), maybe_val);
        graph.add_edge_with(maybe.id, watcher.id, Condition::OnSkip);
        (graph, watcher)
    };

    // Skip case: the None output trips OnSkip and the watcher fires.
    let (graph, watcher) = build(1.0);
    let engine = BspEngine::new(graph);
    let async_nodes::EngineResult::Complete(state) = engine.run(()).await.unwrap() else {
        panic!("expected completion");
    };
    assert_eq!(state.execution_count(watcher.id), 1);

    // Non-skip case: OnSkip does not fire.
    let (graph, watcher) = build(500.0);
    let engine = BspEngine::new(graph);
    let async_nodes::EngineResult::Complete(state) = engine.run(()).await.unwrap() else {
        panic!("expected completion");
    };
    assert_eq!(state.execution_count(watcher.id), 0);
}

// =========================================================================
// MULTI-WIRE AGGREGATION ORDER
// =========================================================================

#[tokio::test]
async fn test_multi_wire_aggregation_is_declaration_ordered() {
    let mut graph = BspGraph::new();
    let a = graph.add_node(WordANode::new());
    let b = graph.add_node(WordBNode::new());
    let join = graph.add_node(JoinWordsNode::new());
    // Wire B *before* A: the aggregated Vec must follow link declaration order.
    graph.add_link(b.output(), join.input());
    graph.add_link(a.output(), join.input());

    let engine = BspEngine::new(graph);
    let async_nodes::EngineResult::Complete(state) = engine.run(()).await.unwrap() else {
        panic!("expected completion");
    };
    let joined: String = state.get_typed(join.output()).unwrap();
    assert_eq!(joined, "b,a");
}

// =========================================================================
// DATAFLOWNODE TRAIT AUTHORING (MACRO-FREE PATH)
// =========================================================================

/// An adder authored directly against the `DataflowNode` trait — the non-macro path
/// used by registries and GUI bridges constructing nodes from descriptors.
struct TraitAdd;

#[async_trait]
impl DataflowNode for TraitAdd {
    fn name(&self) -> &str {
        "trait-add"
    }

    fn input_ports(&self) -> Vec<usize> {
        vec![0, 1]
    }

    fn output_ports(&self) -> Vec<usize> {
        vec![0]
    }

    async fn execute(
        &self,
        _node_id: async_nodes::NodeId,
        inputs: IndexMap<usize, async_nodes::BspValue>,
        _ctx: &BspContext,
    ) -> Result<IndexMap<usize, async_nodes::BspValue>, BspError> {
        let read = |port: usize| {
            inputs
                .get(&port)
                .and_then(|v| v.downcast_ref::<i64>())
                .copied()
                .unwrap_or(0)
        };
        let mut outputs = IndexMap::new();
        outputs.insert(0, async_nodes::BspValue::new(read(0) + read(1)));
        Ok(outputs)
    }
}

/// The identical computation authored via the `#[node]` macro.
#[node("MacroAdd")]
async fn macro_add(a: i64, b: i64) -> i64 {
    a + b
}

fn build_trait_add_graph() -> (BspGraph, async_nodes::OutputPin) {
    let mut graph = BspGraph::new();
    let add = graph.add_node(TraitAdd);
    graph.set_static_input(add.get_input_pin(0), 40_i64);
    graph.set_static_input(add.get_input_pin(1), 2_i64);
    let out = add.get_output_pin(0);
    (graph, out)
}

#[tokio::test]
async fn test_dataflow_node_trait_receives_typed_static_inputs() {
    let (graph, out) = build_trait_add_graph();
    let engine = BspEngine::new(graph);
    let async_nodes::EngineResult::Complete(state) = engine.run(()).await.unwrap() else {
        panic!("expected completion");
    };
    let sum: i64 = state.get_typed(out).unwrap();
    assert_eq!(sum, 42);
}

#[tokio::test]
async fn test_trait_and_macro_authoring_agree_on_the_same_graph() {
    let (trait_graph, trait_out) = build_trait_add_graph();
    let trait_result = BspEngine::new(trait_graph).run(()).await.unwrap();

    let mut macro_graph = BspGraph::new();
    let add = macro_graph.add_node(MacroAddNode::new());
    macro_graph.set_static_input(add.get_input_pin(0), 40_i64);
    macro_graph.set_static_input(add.get_input_pin(1), 2_i64);
    let macro_result = BspEngine::new(macro_graph).run(()).await.unwrap();

    let (
        async_nodes::EngineResult::Complete(trait_state),
        async_nodes::EngineResult::Complete(macro_state),
    ) = (trait_result, macro_result)
    else {
        panic!("expected both graphs to complete");
    };
    let a: i64 = trait_state.get_typed(trait_out).unwrap();
    let b: i64 = macro_state.get_typed(add.get_output_pin(0)).unwrap();
    assert_eq!(a, b);
}

#[tokio::test]
async fn test_trait_nodes_wire_to_macro_nodes() {
    // A macro-authored source feeding a trait-authored doubler through a port link.
    struct TraitDouble;

    #[async_trait]
    impl DataflowNode for TraitDouble {
        fn name(&self) -> &str {
            "trait-double"
        }

        fn input_ports(&self) -> Vec<usize> {
            vec![0]
        }

        fn output_ports(&self) -> Vec<usize> {
            vec![0]
        }

        async fn execute(
            &self,
            _node_id: async_nodes::NodeId,
            inputs: IndexMap<usize, async_nodes::BspValue>,
            _ctx: &BspContext,
        ) -> Result<IndexMap<usize, async_nodes::BspValue>, BspError> {
            let val = inputs
                .get(&0)
                .and_then(|v| v.downcast_ref::<f64>())
                .copied()
                .unwrap_or(0.0);
            let mut outputs = IndexMap::new();
            outputs.insert(0, async_nodes::BspValue::new(val * 2.0));
            Ok(outputs)
        }
    }

    let mut graph = BspGraph::new();
    let src = graph.add_node(ProfileSourceNode::new());
    let dbl = graph.add_node(TraitDouble);
    graph.add_link(src.output(), dbl.get_input_pin(0));

    let engine = BspEngine::new(graph);
    let async_nodes::EngineResult::Complete(state) = engine.run(()).await.unwrap() else {
        panic!("expected completion");
    };
    let doubled: f64 = state.get_typed(dbl.get_output_pin(0)).unwrap();
    assert!((doubled - 10.0).abs() < f64::EPSILON);
}