async-nodes 0.1.0

//! Definition & registry integration tests: `GraphDef` round-trips, the hand-written
//! n8n-ish fixture, registry completeness errors, custom-condition fail-closed behavior,
//! and suspension snapshot serde.

use async_nodes::{
    ActivationDef, BspContext, BspEngine, BspError, BspGraph, BspNode, ConditionDef, EdgeDef,
    EngineResult, GRAPH_DEF_VERSION, GraphDef, NodeDef, NodeOutcome, NodeRegistry, RecoveryDef,
    RunContext, StaticInputDef, SuspendedExecution, async_trait::async_trait,
};
use indexmap::IndexMap;
use serde::{Deserialize, Serialize};

// =========================================================================
// REGISTRY NODE TYPES (S = i64)
// =========================================================================

/// "test.set": assigns the state from `config.value`.
struct SetNode(i64);

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

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

    async fn execute(&self, run: &mut RunContext<'_, i64>) -> Result<NodeOutcome, BspError> {
        *run.state = self.0;
        Ok(NodeOutcome::Success(IndexMap::new()))
    }
}

/// "test.add": adds `config.amount` to the state.
struct AddNode(i64);

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

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

    // Declared so port-carrying edge definitions validate.
    fn output_ports(&self) -> Vec<usize> {
        vec![0]
    }

    async fn execute(&self, run: &mut RunContext<'_, i64>) -> Result<NodeOutcome, BspError> {
        *run.state += self.0;
        Ok(NodeOutcome::Success(IndexMap::new()))
    }
}

/// "test.inc": config-free increment (used by the programmatic round-trip test).
struct IncNode;

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

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

    async fn execute(&self, run: &mut RunContext<'_, i64>) -> Result<NodeOutcome, BspError> {
        *run.state += 1;
        Ok(NodeOutcome::Success(IndexMap::new()))
    }
}

/// "test.noop": does nothing.
struct NoopNode;

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

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

    // Declared so port-carrying edge definitions validate.
    fn input_ports(&self) -> Vec<usize> {
        vec![0]
    }

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

fn def_registry() -> NodeRegistry<i64> {
    let mut registry = NodeRegistry::new();
    registry.register_node("test.set", |def, _| {
        let value = def.config.get("value").and_then(serde_json::Value::as_i64);
        Ok(Box::new(SetNode(value.unwrap_or(0))))
    });
    registry.register_node("test.add", |def, _| {
        let amount = def.config.get("amount").and_then(serde_json::Value::as_i64);
        Ok(Box::new(AddNode(amount.unwrap_or(0))))
    });
    registry.register_node("test.inc", |_, _| Ok(Box::new(IncNode)));
    registry.register_node("test.noop", |_, _| Ok(Box::new(NoopNode)));
    registry.register_condition("test.below_10", |s: &i64, _| *s < 10);
    registry.register_condition("test.at_least_10", |s: &i64, _| *s >= 10);
    registry.register_merger("test.add_merge", |trunk: &mut i64, branch: i64| {
        *trunk += branch;
    });
    registry
}

// =========================================================================
// FIXED-POINT ROUND-TRIPS
// =========================================================================

/// A definition exercising every field kind: configs, activation and recovery overrides,
/// a merger name, execution caps, positions, composite conditions, ports, and entries.
fn showcase_def() -> GraphDef {
    GraphDef {
        version: GRAPH_DEF_VERSION,
        nodes: vec![
            NodeDef {
                id: "start".to_string(),
                type_name: "test.set".to_string(),
                config: serde_json::json!({ "value": 1 }),
                activation: None,
                recovery: None,
                merger: None,
                max_executions: None,
                position: Some((120.0, 80.0)),
            },
            NodeDef {
                id: "bump".to_string(),
                type_name: "test.add".to_string(),
                config: serde_json::json!({ "amount": 3 }),
                activation: Some(ActivationDef::Any),
                recovery: Some(RecoveryDef::Retry {
                    max_retries: 2,
                    base_delay_ms: 100,
                    max_delay_ms: 1_000,
                    jitter: 0.1,
                }),
                merger: Some("test.add_merge".to_string()),
                max_executions: Some(25),
                position: Some((340.0, 80.0)),
            },
            NodeDef {
                id: "finish".to_string(),
                type_name: "test.noop".to_string(),
                config: serde_json::Value::Null,
                activation: None,
                recovery: None,
                merger: None,
                max_executions: None,
                position: None,
            },
        ],
        edges: vec![
            EdgeDef {
                from: "start".to_string(),
                to: "bump".to_string(),
                condition: ConditionDef::Always,
                max_iterations: None,
                source_port: None,
                target_port: None,
            },
            EdgeDef {
                from: "bump".to_string(),
                to: "bump".to_string(),
                condition: ConditionDef::Named("test.below_10".to_string()),
                max_iterations: Some(20),
                source_port: None,
                target_port: None,
            },
            EdgeDef {
                from: "bump".to_string(),
                to: "finish".to_string(),
                condition: ConditionDef::Not(Box::new(ConditionDef::Named(
                    "test.below_10".to_string(),
                ))),
                max_iterations: None,
                source_port: Some(0),
                target_port: Some(0),
            },
        ],
        entry: vec!["start".to_string()],
        static_inputs: vec![StaticInputDef {
            node: "finish".to_string(),
            port: 0,
            value: serde_json::json!({ "seed": 42 }),
        }],
    }
}

#[tokio::test]
async fn test_graph_def_round_trip_is_a_fixed_point() {
    let registry = def_registry();
    let ctx = BspContext::new();
    let def = showcase_def();

    let graph = BspGraph::from_def(&def, &registry, &ctx).unwrap();
    let reemitted = graph.to_def();
    assert_eq!(def, reemitted, "def -> graph -> def must be a fixed point");

    // And the same through JSON.
    let json = def.to_json().unwrap();
    let parsed = GraphDef::from_json(&json).unwrap();
    assert_eq!(def, parsed);
    let graph2 = BspGraph::from_def(&parsed, &registry, &ctx).unwrap();
    assert_eq!(graph2.to_def(), def);

    // The def-built graph also runs: 1 -> +3 -> +3 -> +3 = 10, then finish.
    let engine = BspEngine::new(graph);
    let EngineResult::Complete(state) = engine.run(0_i64).await.unwrap() else {
        panic!("expected completion");
    };
    assert_eq!(state.user, 10);
}

#[tokio::test]
async fn test_programmatic_graph_round_trips_through_def() {
    // Phase 5 gate: fluent build -> to_def -> instantiate via registry -> run -> identical.
    let mut graph = BspGraph::new();
    let a = graph.add_node(IncNode);
    let b = graph.add_node(IncNode);
    let c = graph.add_node(NoopNode);
    graph.add_edge(a.id, b.id);
    graph.add_edge(b.id, c.id);

    let def = graph.to_def();
    let json = def.to_json().unwrap();

    let registry = def_registry();
    let ctx = BspContext::new();
    let rebuilt =
        BspGraph::from_def(&GraphDef::from_json(&json).unwrap(), &registry, &ctx).unwrap();

    // Fixed point through the programmatic path too (ids serialize as string labels).
    assert_eq!(rebuilt.to_def(), def);

    let first = BspEngine::new(graph).run(0_i64).await.unwrap();
    let second = BspEngine::new(rebuilt).run(0_i64).await.unwrap();
    let (EngineResult::Complete(a), EngineResult::Complete(b)) = (first, second) else {
        panic!("expected completions");
    };
    assert_eq!(a.user, 2);
    assert_eq!(a.user, b.user);
}

// =========================================================================
// THE n8n-ISH FIXTURE
// =========================================================================

#[tokio::test]
async fn test_n8n_fixture_loads_and_runs() {
    let json = include_str!("fixtures/n8n_counter.json");
    let def = GraphDef::from_json(json).unwrap();

    let registry = def_registry();
    let ctx = BspContext::new();
    let graph = BspGraph::from_def(&def, &registry, &ctx).unwrap();

    // Layout hints round-trip: the GUI never loses a human's arrangement.
    let reemitted = graph.to_def();
    let original: serde_json::Value = serde_json::from_str(json).unwrap();
    let round_tripped: serde_json::Value =
        serde_json::from_str(&reemitted.to_json().unwrap()).unwrap();
    assert_eq!(original, round_tripped);

    let engine = BspEngine::new(graph);
    let EngineResult::Complete(state) = engine.run(0_i64).await.unwrap() else {
        panic!("expected completion");
    };
    // 1 -> 4 -> 7 -> 10, then exit the loop.
    assert_eq!(state.user, 10);
    let bump_executions = state.execution_count(2);
    assert_eq!(bump_executions, 3);
}

// =========================================================================
// REGISTRY COMPLETENESS
// =========================================================================

#[test]
fn test_unknown_type_name_and_condition_are_validation_errors() {
    let mut def = showcase_def();
    def.nodes[0].type_name = "nope.missing".to_string();
    def.edges[0].condition = ConditionDef::Named("nope.condition".to_string());

    let registry = def_registry();
    let ctx = BspContext::new();
    let result = BspGraph::from_def(&def, &registry, &ctx);
    let Err(BspError::ValidationError(msg)) = result else {
        panic!("expected a validation error");
    };
    // Fail-fast registration lists *all* unknown names, not just the first.
    assert!(msg.contains("nope.missing"), "missing type name in: {msg}");
    assert!(
        msg.contains("nope.condition"),
        "missing condition in: {msg}"
    );
}

#[test]
fn test_unknown_merger_is_a_validation_error() {
    let mut def = showcase_def();
    def.nodes[1].merger = Some("nope.merger".to_string());
    let registry = def_registry();
    let ctx = BspContext::new();
    let result = BspGraph::from_def(&def, &registry, &ctx);
    let Err(BspError::ValidationError(msg)) = result else {
        panic!("expected a validation error");
    };
    assert!(msg.contains("nope.merger"), "missing merger in: {msg}");
}

#[test]
fn test_unsupported_version_is_a_def_error() {
    let mut def = showcase_def();
    def.version = 99;
    let registry = def_registry();
    let ctx = BspContext::new();
    let result = BspGraph::from_def(&def, &registry, &ctx);
    assert!(matches!(result, Err(BspError::DefError(_))));
}

// =========================================================================
// CUSTOM CONDITIONS: FAIL CLOSED, THEN RE-REGISTER
// =========================================================================

fn custom_condition_def() -> GraphDef {
    let mut def = showcase_def();
    def.edges[0].condition = ConditionDef::Custom {
        label: "test.custom_positive".to_string(),
    };
    def
}

#[tokio::test]
async fn test_custom_condition_fails_closed_unless_re_registered() {
    let ctx = BspContext::new();
    let def = custom_condition_def();

    // Without a registration the definition cannot instantiate.
    let registry = def_registry();
    let result = BspGraph::from_def(&def, &registry, &ctx);
    let Err(BspError::ValidationError(msg)) = result else {
        panic!("expected a validation error");
    };
    assert!(msg.contains("test.custom_positive"), "got: {msg}");

    // Re-registering the label revives the definition, and the predicate routes.
    let registry = registry.with_condition("test.custom_positive", |s: &i64, _| *s > 0);
    let graph = BspGraph::from_def(&def, &registry, &ctx).unwrap();
    let engine = BspEngine::new(graph);
    let EngineResult::Complete(state) = engine.run(0_i64).await.unwrap() else {
        panic!("expected completion");
    };
    assert_eq!(state.user, 10);
}

// =========================================================================
// SUSPENSION SERDE
// =========================================================================

#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
struct SerdeState {
    flag: bool,
    count: u64,
}

/// Writes a marker value to output port 0 (to prove pin values never serialize).
struct MarkerNode;

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

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

    async fn execute(&self, run: &mut RunContext<'_, SerdeState>) -> Result<NodeOutcome, BspError> {
        let mut outputs = IndexMap::new();
        outputs.insert(0, async_nodes::BspValue::new("secret-marker".to_string()));
        let _ = run;
        Ok(NodeOutcome::Success(outputs))
    }
}

/// Suspends once, waiting for a boolean on input port 0; echoes it to output port 0.
struct ApprovalNode;

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

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

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

    async fn execute(&self, run: &mut RunContext<'_, SerdeState>) -> Result<NodeOutcome, BspError> {
        run.state.count += 1;
        if !run.state.flag {
            run.state.flag = true;
            return Ok(NodeOutcome::Suspend {
                payload: serde_json::json!({ "need": "approval" }),
                waiting_inputs: vec![0],
            });
        }
        let approved = run.input::<bool>(0).unwrap_or(false);
        let mut outputs = IndexMap::new();
        outputs.insert(0, async_nodes::BspValue::new(approved));
        Ok(NodeOutcome::Success(outputs))
    }
}

#[tokio::test]
async fn test_suspension_snapshot_serde_round_trip() {
    let mut graph = BspGraph::new();
    let marker = graph.add_node(MarkerNode);
    let approval = graph.add_node(ApprovalNode);
    let _ = marker;

    let engine = BspEngine::new(graph);
    let EngineResult::Suspended(suspended) = engine.run(SerdeState::default()).await.unwrap()
    else {
        panic!("expected suspension");
    };
    assert_eq!(suspended.suspensions[0].waiting_inputs, vec![0]);

    let json = suspended.to_json().unwrap();
    // **The value store is never serialized**: the marker pin value must not appear.
    assert!(!json.contains("secret-marker"));
    // The suspension payload and counters do serialize.
    assert!(json.contains("approval"));
    assert!(json.contains("\"count\":1"));

    // Restore from JSON and resume with the waiting input re-injected.
    let restored = SuspendedExecution::<SerdeState>::from_json(&json).unwrap();
    let mut injections = IndexMap::new();
    injections.insert(
        async_nodes::InputPin {
            node_id: approval.id,
            port: 0,
        },
        async_nodes::BspValue::new(true),
    );
    let EngineResult::Complete(state) = engine.resume(restored, injections).await.unwrap() else {
        panic!("expected completion after serde resume");
    };
    assert_eq!(state.execution_count(approval.id), 2);
    let approved: bool = state.get_output_typed(approval.get_output_pin(0)).unwrap();
    assert!(approved);
}

// =========================================================================
// STATIC INPUTS (GraphDef round-trips + value factory)
// =========================================================================

#[test]
fn test_static_inputs_round_trip_preserves_json() {
    let json = r#"{
        "version": 1,
        "nodes": [{"id": "sink", "type_name": "test.noop"}],
        "edges": [],
        "static_inputs": [
            {"node": "sink", "port": 0, "value": true},
            {"node": "sink", "port": 1, "value": 41},
            {"node": "sink", "port": 2, "value": 2.5},
            {"node": "sink", "port": 3, "value": "hello"},
            {"node": "sink", "port": 4, "value": {"nested": [1, 2]}}
        ]
    }"#;
    let def = GraphDef::from_json(json).unwrap();
    let registry = def_registry();
    let ctx = BspContext::new();
    let graph = BspGraph::from_def(&def, &registry, &ctx).unwrap();

    // The default value factory produces typed values, and `to_def` projects them back
    // losslessly — the static-input section is a fixed point.
    let projected = graph.to_def();
    assert_eq!(projected.static_inputs, def.static_inputs);
}

#[test]
fn test_static_input_unknown_node_is_a_validation_error() {
    let def = GraphDef::from_json(
        r#"{
            "version": 1,
            "nodes": [{"id": "sink", "type_name": "test.noop"}],
            "edges": [],
            "static_inputs": [{"node": "ghost", "port": 0, "value": 1}]
        }"#,
    )
    .unwrap();
    let registry = def_registry();
    let ctx = BspContext::new();
    let result = BspGraph::<i64>::from_def(&def, &registry, &ctx);
    let Err(err) = result else {
        panic!("expected a validation error for the unknown static-input node");
    };
    assert!(matches!(err, BspError::ValidationError(_)), "got {err:?}");
}

#[test]
fn test_value_factory_override_is_honored() {
    let def = GraphDef::from_json(
        r#"{
            "version": 1,
            "nodes": [{"id": "sink", "type_name": "test.noop"}],
            "edges": [],
            "static_inputs": [{"node": "sink", "port": 0, "value": 41}]
        }"#,
    )
    .unwrap();
    // Override: every static input arrives as the string "wrapped" — proves the factory
    // is the single conversion point between def JSON and typed pin values.
    let registry: NodeRegistry<i64> =
        def_registry().with_value_factory(|_| async_nodes::BspValue::new("wrapped".to_string()));
    let ctx = BspContext::new();
    let graph = BspGraph::<i64>::from_def(&def, &registry, &ctx).unwrap();
    let projected = graph.to_def();
    assert_eq!(
        projected.static_inputs[0].value,
        serde_json::json!("wrapped")
    );
}