lc_langgraph/

node.rs

// crates/lc-langgraph/src/node.rs
//! Node definition for LangGraph
//!
//! Nodes are the execution units in a graph. Each node receives the current
//! state and returns a state update.

use crate::errors::GraphError;
use crate::state::{StateSchema, StateUpdate};
use async_trait::async_trait;
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;

/// Graph Node trait
///
/// Nodes are async functions that take a state and return a state update.
/// They represent the work units in the graph.
#[async_trait]
pub trait GraphNode<S: StateSchema>: Send + Sync + 'static {
    /// Execute the node
    ///
    /// # Parameters
    /// - `state`: Current state of the graph
    /// - `config`: Optional configuration for this execution
    ///
    /// # Returns
    /// A state update that will be merged into the current state
    async fn execute(
        &self,
        state: &S,
        config: Option<NodeConfig>,
    ) -> Result<StateUpdate<S>, GraphError>;

    /// Get node name
    fn name(&self) -> &str;
}

/// Metadata key the compiled graph uses to inject the human's decision into a
/// node that is being resumed after a runtime [`crate::errors::GraphError::InterruptRequest`].
/// A resume-aware node reads the value under this key from its `NodeConfig.metadata`.
pub const INTERRUPT_RESUME_KEY: &str = "__lc_interrupt_resume";

/// Node configuration
#[derive(Debug, Clone, Default)]
pub struct NodeConfig {
    /// Maximum recursion depth
    pub recursion_limit: usize,

    /// Custom metadata
    pub metadata: std::collections::HashMap<String, serde_json::Value>,

    /// Enable debug tracing
    pub debug: bool,
}

/// Node execution result (alias for clarity)
pub type NodeResult<S> = Result<StateUpdate<S>, GraphError>;

/// Async node function type (boxed future)
pub type AsyncNodeFn<S> =
    Box<dyn Fn(&S) -> Pin<Box<dyn Future<Output = NodeResult<S>> + Send>> + Send + Sync>;

/// AsyncFn trait for simpler async node creation
pub trait AsyncFn<S: StateSchema>: Send + Sync {
    /// Call the async function with the given state.
    fn call(&self, state: &S) -> Pin<Box<dyn Future<Output = NodeResult<S>> + Send>>;
}

impl<S: StateSchema, F, Fut> AsyncFn<S> for F
where
    F: Fn(&S) -> Fut + Send + Sync + 'static,
    Fut: Future<Output = NodeResult<S>> + Send + 'static,
{
    fn call(&self, state: &S) -> Pin<Box<dyn Future<Output = NodeResult<S>> + Send>> {
        Box::pin((self)(state))
    }
}

/// AsyncNode - Simple async node wrapper
pub struct AsyncNode<S: StateSchema, F: AsyncFn<S>> {
    name: String,
    func: F,
    _marker: PhantomData<S>,
}

impl<S: StateSchema, F: AsyncFn<S>> AsyncNode<S, F> {
    /// Create a new async node with the given name and function.
    pub fn new(name: impl Into<String>, func: F) -> Self {
        Self {
            name: name.into(),
            func,
            _marker: PhantomData,
        }
    }
}

#[async_trait]
impl<S: StateSchema, F: AsyncFn<S> + 'static> GraphNode<S> for AsyncNode<S, F> {
    async fn execute(&self, state: &S, _config: Option<NodeConfig>) -> NodeResult<S> {
        self.func.call(state).await
    }

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

/// Function-based node implementation
///
/// Wraps an async function as a GraphNode.
pub struct FunctionNode<S: StateSchema, F> {
    name: String,
    func: F,
    _marker: PhantomData<S>,
}

impl<S: StateSchema, F> FunctionNode<S, F>
where
    F: Fn(&S) -> Pin<Box<dyn Future<Output = Result<StateUpdate<S>, GraphError>> + Send>>
        + Send
        + Sync,
{
    /// Create a new function node
    pub fn new(name: impl Into<String>, func: F) -> Self {
        Self {
            name: name.into(),
            func,
            _marker: PhantomData,
        }
    }
}

#[async_trait]
impl<S: StateSchema, F: 'static> GraphNode<S> for FunctionNode<S, F>
where
    F: Fn(&S) -> Pin<Box<dyn Future<Output = Result<StateUpdate<S>, GraphError>> + Send>>
        + Send
        + Sync,
{
    async fn execute(
        &self,
        state: &S,
        _config: Option<NodeConfig>,
    ) -> Result<StateUpdate<S>, GraphError> {
        (self.func)(state).await
    }

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

/// Simple sync node wrapper
///
/// For nodes that don't need async execution.
pub struct SyncNode<S: StateSchema, F> {
    name: String,
    func: F,
    _marker: PhantomData<S>,
}

impl<S: StateSchema, F> SyncNode<S, F>
where
    F: Fn(&S) -> Result<StateUpdate<S>, GraphError> + Send + Sync,
{
    /// Create a new sync node
    pub fn new(name: impl Into<String>, func: F) -> Self {
        Self {
            name: name.into(),
            func,
            _marker: PhantomData,
        }
    }
}

#[async_trait]
impl<S: StateSchema, F: 'static> GraphNode<S> for SyncNode<S, F>
where
    F: Fn(&S) -> Result<StateUpdate<S>, GraphError> + Send + Sync,
{
    async fn execute(
        &self,
        state: &S,
        _config: Option<NodeConfig>,
    ) -> Result<StateUpdate<S>, GraphError> {
        (self.func)(state)
    }

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

/// Resume-aware node (LangGraph-style interrupt).
///
/// The wrapped closure is called twice per runtime interrupt:
/// - first pass: `resume` is `None` — the node either completes normally or
///   suspends itself by returning `Err(GraphError::InterruptRequest { payload })`;
/// - resume: after a human answers, the node is re-entered with `resume` =
///   `Some(decision)` so the closure can continue past the suspension.
///
/// Branch on that second argument instead of re-walking expensive side effects
/// (an API call, a DB write) that already happened on the first pass — keep any
/// such work memoized in the state between the two passes.
pub struct InterruptibleNode<S: StateSchema, F> {
    name: String,
    func: F,
    _marker: PhantomData<S>,
}

impl<S: StateSchema, F> InterruptibleNode<S, F>
where
    F: Fn(&S, Option<&serde_json::Value>) -> Pin<Box<dyn Future<Output = NodeResult<S>> + Send>>
        + Send
        + Sync,
{
    /// Create a resume-aware node. The closure takes the current state and the
    /// injected resume decision (`None` on first pass, `Some(decision)` on resume).
    pub fn new(name: impl Into<String>, func: F) -> Self {
        Self {
            name: name.into(),
            func,
            _marker: PhantomData,
        }
    }
}

#[async_trait]
impl<S: StateSchema, F: 'static> GraphNode<S> for InterruptibleNode<S, F>
where
    F: Fn(&S, Option<&serde_json::Value>) -> Pin<Box<dyn Future<Output = NodeResult<S>> + Send>>
        + Send
        + Sync,
{
    async fn execute(
        &self,
        state: &S,
        config: Option<NodeConfig>,
    ) -> Result<StateUpdate<S>, GraphError> {
        let resume = config
            .as_ref()
            .and_then(|c| c.metadata.get(INTERRUPT_RESUME_KEY))
            .cloned();
        (self.func)(state, resume.as_ref()).await
    }

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

/// Placeholder node for entry/exit points
pub struct SentinelNode {
    name: String,
}

impl SentinelNode {
    /// Create a sentinel node for the `START` marker.
    pub fn start() -> Self {
        Self {
            name: crate::START.to_string(),
        }
    }

    /// Create a sentinel node for the `END` marker.
    pub fn end() -> Self {
        Self {
            name: crate::END.to_string(),
        }
    }

    /// Create a custom-named sentinel node.
    pub fn custom(name: impl Into<String>) -> Self {
        Self { name: name.into() }
    }
}

#[async_trait]
impl<S: StateSchema> GraphNode<S> for SentinelNode {
    async fn execute(
        &self,
        _state: &S,
        _config: Option<NodeConfig>,
    ) -> Result<StateUpdate<S>, GraphError> {
        // Sentinel nodes don't modify state
        Ok(StateUpdate::unchanged())
    }

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

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

    #[tokio::test]
    async fn test_sync_node() {
        let node = SyncNode::new("test", |state: &AgentState| {
            Ok(StateUpdate::full(AgentState::new(state.input.clone())))
        });

        let state = AgentState::new("Hello".to_string());
        let result = node.execute(&state, None).await;
        assert!(result.is_ok());
    }

    #[test]
    fn test_sentinel_nodes() {
        let start: SentinelNode = SentinelNode::start();
        assert_eq!(GraphNode::<AgentState>::name(&start), crate::START);

        let end: SentinelNode = SentinelNode::end();
        assert_eq!(GraphNode::<AgentState>::name(&end), crate::END);
    }
}