async-nodes 0.1.0

//! Loop integration tests: bounded and unbounded cycles, iteration budgets, mid-loop
//! suspension/resumption, execution caps, and token-flood protection.

use async_nodes::{
    Activation, BspEngine, BspError, BspGraph, BspNode, Condition, EngineResult, NodeOutcome,
    RunContext, TypedNodeHandle, async_trait::async_trait,
};
use indexmap::IndexMap;

// =========================================================================
// TEST NODES
// =========================================================================

#[derive(Clone, Debug, Default, PartialEq, Eq)]
struct LoopState {
    count: u64,
    suspended_once: bool,
}

/// Increments `count` each firing; suspends once when `count == suspend_at`.
struct CountNode {
    name: &'static str,
    suspend_at: Option<u64>,
}

#[async_trait]
impl BspNode<LoopState> for CountNode {
    fn name(&self) -> &str {
        self.name
    }

    async fn execute(&self, run: &mut RunContext<'_, LoopState>) -> Result<NodeOutcome, BspError> {
        run.state.count += 1;
        if self.suspend_at == Some(run.state.count) && !run.state.suspended_once {
            run.state.suspended_once = true;
            return Ok(NodeOutcome::Suspend {
                payload: serde_json::json!({ "suspended_at": run.state.count }),
                waiting_inputs: vec![],
            });
        }
        Ok(NodeOutcome::Success(IndexMap::new()))
    }
}

/// A no-op node with a configurable activation policy and execution cap.
struct GateNode {
    name: &'static str,
    activation: Activation,
    max_executions: Option<u64>,
}

impl GateNode {
    fn any(name: &'static str) -> Self {
        Self {
            name,
            activation: Activation::Any,
            max_executions: None,
        }
    }
}

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

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

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

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

/// The canonical counter loop: entry -> counter -> sink, plus a conditioned back-edge
/// counter -> counter.
struct CounterGraph {
    graph: BspGraph<LoopState>,
    counter: TypedNodeHandle<CountNode>,
}

fn counter_graph(suspend_at: Option<u64>, loop_max_iterations: u64, target: u64) -> CounterGraph {
    let mut graph = BspGraph::new();
    let entry = graph.add_node(GateNode::any("entry"));
    let counter = graph.add_node(CountNode {
        name: "counter",
        suspend_at,
    });
    let sink = graph.add_node(GateNode::any("sink"));
    graph.add_edge(entry.id, counter.id);
    graph.register_condition("more", move |s: &LoopState, _| s.count < target);
    graph.register_condition("done", move |s: &LoopState, _| s.count >= target);
    graph
        .add_edge(counter.id, sink.id)
        .when(Condition::Named("done".into()));
    graph
        .add_edge(counter.id, counter.id)
        .when(Condition::Named("more".into()))
        .with_max_iterations(loop_max_iterations);
    CounterGraph { graph, counter }
}

// =========================================================================
// LOOPS
// =========================================================================

#[tokio::test]
async fn test_counter_loop_runs_exactly_n_iterations() {
    let CounterGraph { graph, counter } = counter_graph(None, 50, 10);
    let engine = BspEngine::new(graph);
    let EngineResult::Complete(state) = engine.run(LoopState::default()).await.unwrap() else {
        panic!("expected completion");
    };

    assert_eq!(state.user.count, 10);
    assert_eq!(state.execution_count(counter.id), 10);
    // The back-edge fired once per non-final iteration.
    let total_iterations: u64 = state.iterations().values().sum();
    assert_eq!(total_iterations, 9);

    // A bounded cycle draws no unbounded-cycle warning.
    let report = engine.validate_with_report();
    assert!(report.is_ok());
    assert!(
        !report
            .warnings
            .iter()
            .any(|w| w.contains("no max_iterations budget"))
    );
}

#[tokio::test]
async fn test_unbounded_loop_trips_step_limit() {
    let mut graph = BspGraph::new();
    let looper = graph.add_node(GateNode::any("looper"));
    graph.add_edge(looper.id, looper.id); // Always, unbounded
    graph.add_entry(looper.id);

    let engine: BspEngine<LoopState> = BspEngine::new(graph);
    let outcome = engine
        .with_max_steps(25)
        .run(LoopState::default())
        .await
        .unwrap();
    let EngineResult::StepLimitReached(state) = outcome else {
        panic!("expected StepLimitReached, got {outcome:?}");
    };
    assert_eq!(state.step, 25);
    assert_eq!(state.execution_count(looper.id), 25);
}

#[tokio::test]
async fn test_bounded_loop_raises_loop_bound_exceeded() {
    let mut graph = BspGraph::new();
    let entry = graph.add_node(GateNode::any("entry"));
    let looper = graph.add_node(GateNode::any("looper"));
    graph.add_edge(entry.id, looper.id);
    graph.add_edge(looper.id, looper.id).with_max_iterations(3);

    let engine: BspEngine<LoopState> = BspEngine::new(graph);
    let err = engine.run(LoopState::default()).await.unwrap_err();
    let BspError::LoopBoundExceeded { from, to, max, .. } = err else {
        panic!("expected LoopBoundExceeded, got {err:?}");
    };
    assert_eq!((from, to), (looper.id, looper.id));
    assert_eq!(max, 3);
}

#[tokio::test]
async fn test_suspend_mid_loop_resume_continues_counters() {
    // Suspend on iteration 5 of a 10-count loop; resume must continue at 5, not 0.
    let CounterGraph { graph, counter } = counter_graph(Some(5), 50, 10);

    #[derive(Default)]
    struct SuspensionEvents {
        suspended: std::sync::Mutex<Vec<usize>>,
    }
    impl async_nodes::BspInterceptor for SuspensionEvents {
        fn on_suspended(&self, suspensions: &[async_nodes::PendingSuspension]) {
            self.suspended.lock().unwrap().push(suspensions.len());
        }
    }
    let events = std::sync::Arc::new(SuspensionEvents::default());

    let engine = BspEngine::new(graph).with_interceptor(events.clone());

    let EngineResult::Suspended(suspended) = engine.run(LoopState::default()).await.unwrap() else {
        panic!("expected suspension");
    };
    assert_eq!(suspended.suspensions.len(), 1);
    assert_eq!(suspended.suspensions[0].node_id, counter.id);
    assert_eq!(
        suspended.suspensions[0].payload,
        serde_json::json!({ "suspended_at": 5 })
    );
    // The suspended branch's state folded into the trunk: count and iteration budget are
    // exactly where the loop left them.
    assert_eq!(suspended.state().user.count, 5);
    let iterations_at_suspend: u64 = suspended.state().iterations().values().sum();
    assert_eq!(iterations_at_suspend, 4);

    let EngineResult::Complete(state) = engine.resume(*suspended, IndexMap::new()).await.unwrap()
    else {
        panic!("expected completion after resume");
    };
    assert_eq!(state.user.count, 10);
    assert!(state.user.suspended_once);
    // Continuation proof: the back-edge counter kept counting from 4 (8 total), and the
    // node executed 10 times overall — a reset would show 4 iterations / 9 executions.
    let total_iterations: u64 = state.iterations().values().sum();
    assert_eq!(total_iterations, 8);
    assert_eq!(state.execution_count(counter.id), 10);
    // The suspension surfaced as an event, once.
    assert_eq!(events.suspended.lock().unwrap().as_slice(), [1]);
}

#[tokio::test]
async fn test_all_join_on_cycle_warns_and_deadlocks() {
    // The classic self-deadlock: an All-join inside a loop whose entry-edge token arrives
    // once ever while the back-edge token arrives per iteration.
    let mut graph = BspGraph::new();
    let entry = graph.add_node(GateNode::any("entry"));
    let join = graph.add_node(GateNode {
        activation: Activation::All,
        ..GateNode::any("join")
    });
    let body = graph.add_node(GateNode::any("body"));
    graph.add_edge(entry.id, join.id);
    graph.add_edge(join.id, body.id);
    graph.add_edge(body.id, join.id).with_max_iterations(5);
    // Force the cycle to be fireable so the warning (not the unfireable-cycle error)
    // is what fires.
    graph.add_entry(join.id);

    let engine: BspEngine<LoopState> = BspEngine::new(graph);
    let report = engine.validate_with_report();
    assert!(report.is_ok());
    assert!(
        report
            .warnings
            .iter()
            .any(|w| w.contains("All-join on a cycle")),
        "expected an All-join-on-cycle warning, got {:?}",
        report.warnings
    );

    // At runtime the join starves once the entry token is consumed: the run deadlocks.
    let err = engine.run(LoopState::default()).await.unwrap_err();
    assert!(matches!(err, BspError::Deadlock), "got {err:?}");
}

#[tokio::test]
async fn test_max_executions_caps_poll_style_nodes() {
    let mut graph = BspGraph::new();
    let entry = graph.add_node(GateNode::any("entry"));
    let poller = graph.add_node(GateNode {
        max_executions: Some(3),
        ..GateNode::any("poller")
    });
    graph.add_edge(entry.id, poller.id);
    graph.add_edge(poller.id, poller.id);

    let engine: BspEngine<LoopState> = BspEngine::new(graph);
    let err = engine.run(LoopState::default()).await.unwrap_err();
    let BspError::MaxExecutionsExceeded { node_id, max } = err else {
        panic!("expected MaxExecutionsExceeded, got {err:?}");
    };
    assert_eq!(node_id, poller.id);
    assert_eq!(max, 3);
}

#[tokio::test]
async fn test_token_flood_protection() {
    // A fast unbounded loop feeds an All-join whose other input never re-fires: tokens
    // accumulate on the fast edge until the per-edge cap trips.
    let mut graph = BspGraph::new();
    let fast = graph.add_node(GateNode::any("fast"));
    let slow = graph.add_node(GateNode::any("slow"));
    let join = graph.add_node(GateNode {
        activation: Activation::All,
        ..GateNode::any("join")
    });
    graph.add_edge(fast.id, fast.id); // unbounded self-loop
    graph.add_edge(fast.id, join.id);
    graph.add_edge(slow.id, join.id);
    // The self-loop cannot start on its own; make the fast node an explicit entry.
    graph.add_entry(fast.id);

    let engine: BspEngine<LoopState> = BspEngine::new(graph);
    let err = engine
        .with_max_pending_tokens(3)
        .with_max_steps(1_000)
        .run(LoopState::default())
        .await
        .unwrap_err();
    assert!(
        matches!(err, BspError::TokenLimitExceeded { cap: 3, .. }),
        "got {err:?}"
    );
}

#[tokio::test]
async fn test_step_limit_is_a_first_class_result_on_stateless_runs() {
    // The unified three-outcome result expresses a tripped budget directly — never an
    // error, never a silent salvage.
    let mut graph: BspGraph = BspGraph::new();
    let entry = graph.add_node(GateNode::any("entry"));
    let looper = graph.add_node(GateNode::any("looper"));
    graph.add_edge(entry.id, looper.id);
    graph.add_edge(looper.id, looper.id);

    let engine = BspEngine::new(graph).with_max_steps(4);
    let outcome = engine.run(()).await.unwrap();
    let EngineResult::StepLimitReached(state) = outcome else {
        panic!("expected StepLimitReached, got {outcome:?}");
    };
    assert_eq!(state.step, 4);
}