lc_langgraph/compiled/
graph.rs

1// crates/lc-langgraph/src/compiled/graph.rs
2//! CompiledGraph struct definition, constructors, getters, and runtime injection
3
4use super::types::{DynamicTask, GraphExecution};
5use crate::checkpointer::Checkpointer;
6use crate::edge::{ConditionalEdge, GraphEdge};
7use crate::errors::{GraphError, GraphResult};
8use crate::node::GraphNode;
9use crate::state::{Reducer, StateSchema};
10use crate::subgraph::SubgraphNode;
11use std::collections::HashMap;
12use std::sync::Arc;
13use tokio::sync::Mutex;
14use tokio::sync::RwLock;
15
16/// CompiledGraph - Ready-to-execute graph
17///
18/// Created from StateGraph.compile(). Handles:
19/// - State management and updates
20/// - Edge routing (fixed and conditional)
21/// - Execution with recursion limits
22/// - Checkpointing for persistence
23#[allow(clippy::type_complexity)]
24#[derive(Clone)]
25pub struct CompiledGraph<S: StateSchema> {
26    pub(super) nodes: HashMap<String, Arc<dyn GraphNode<S>>>,
27    pub(super) edges: Vec<GraphEdge>,
28    pub(super) entry_point: String,
29    pub(super) default_reducer: Arc<dyn Reducer<S>>,
30    pub(super) conditional_routers: HashMap<String, Arc<dyn ConditionalEdge<S>>>,
31    pub(super) checkpointer: Option<Arc<Mutex<dyn Checkpointer<S> + Send>>>,
32    pub(super) recursion_limit: usize,
33    pub(super) interrupt_before: Vec<String>,
34    pub(super) interrupt_after: Vec<String>,
35
36    pub(super) runtime_nodes: Arc<RwLock<HashMap<String, Arc<dyn GraphNode<S>>>>>,
37    pub(super) runtime_edges: Arc<RwLock<Vec<GraphEdge>>>,
38    pub(super) runtime_conditional_routers:
39        Arc<RwLock<HashMap<String, Arc<dyn ConditionalEdge<S>>>>>,
40    pub(super) task_inbox: Arc<Mutex<Vec<DynamicTask>>>,
41}
42
43impl<S: StateSchema> std::fmt::Debug for CompiledGraph<S> {
44    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
45        f.debug_struct("CompiledGraph")
46            .field("nodes", &self.nodes.keys().collect::<Vec<_>>())
47            .field("edges", &self.edges)
48            .field("entry_point", &self.entry_point)
49            .field("recursion_limit", &self.recursion_limit)
50            .field("interrupt_before", &self.interrupt_before)
51            .field("interrupt_after", &self.interrupt_after)
52            .finish()
53    }
54}
55
56impl<S: StateSchema> CompiledGraph<S> {
57    pub(crate) fn new(
58        nodes: HashMap<String, Arc<dyn GraphNode<S>>>,
59        edges: Vec<GraphEdge>,
60        entry_point: String,
61        default_reducer: Arc<dyn Reducer<S>>,
62    ) -> Self {
63        Self {
64            nodes,
65            edges,
66            entry_point,
67            default_reducer,
68            conditional_routers: HashMap::new(),
69            checkpointer: None,
70            recursion_limit: 25,
71            interrupt_before: Vec::new(),
72            interrupt_after: Vec::new(),
73            runtime_nodes: Arc::new(RwLock::new(HashMap::new())),
74            runtime_edges: Arc::new(RwLock::new(Vec::new())),
75            runtime_conditional_routers: Arc::new(RwLock::new(HashMap::new())),
76            task_inbox: Arc::new(Mutex::new(Vec::new())),
77        }
78    }
79
80    pub(crate) fn add_router(&mut self, name: String, router: Arc<dyn ConditionalEdge<S>>) {
81        self.conditional_routers.insert(name, router);
82    }
83
84    /// Attach a checkpointer to this graph for state persistence.
85    pub fn with_checkpointer<C: Checkpointer<S> + 'static>(mut self, checkpointer: C) -> Self {
86        self.checkpointer = Some(Arc::new(Mutex::new(checkpointer)));
87        self
88    }
89
90    /// Set the maximum recursion depth before execution aborts.
91    pub fn with_recursion_limit(mut self, limit: usize) -> Self {
92        self.recursion_limit = limit;
93        self
94    }
95
96    /// Set the node names before which execution should be interrupted.
97    pub fn with_interrupt_before(mut self, nodes: Vec<String>) -> Self {
98        self.interrupt_before = nodes;
99        self
100    }
101
102    /// Set the node names after which execution should be interrupted.
103    pub fn with_interrupt_after(mut self, nodes: Vec<String>) -> Self {
104        self.interrupt_after = nodes;
105        self
106    }
107
108    /// Return the names of all static nodes registered in the graph.
109    pub fn node_names(&self) -> Vec<String> {
110        self.nodes.keys().cloned().collect()
111    }
112
113    /// Return the graph's fixed edges.
114    pub fn get_edges(&self) -> &[GraphEdge] {
115        &self.edges
116    }
117
118    /// Return the entry point node name.
119    pub fn entry_point(&self) -> &str {
120        &self.entry_point
121    }
122
123    /// Return the configured recursion limit.
124    pub fn recursion_limit(&self) -> usize {
125        self.recursion_limit
126    }
127
128    /// Return the nodes that interrupt execution before running.
129    pub fn interrupt_before(&self) -> &[String] {
130        &self.interrupt_before
131    }
132
133    /// Return the nodes that interrupt execution after running.
134    pub fn interrupt_after(&self) -> &[String] {
135        &self.interrupt_after
136    }
137
138    /// Get the last checkpoint state (for interrupt recovery), together with the
139    /// recursion budget consumed up to that checkpoint.
140    ///
141    /// H5: no longer guesses "most recent" from `list()`'s HashMap order + `ids.last()`;
142    /// instead the checkpointer's `last()` returns the truly newest checkpoint by
143    /// (timestamp, seq).
144    pub async fn last_checkpoint_state(&self) -> Option<(S, usize)> {
145        if let Some(ref cp) = self.checkpointer {
146            let guard = cp.lock().await;
147            guard.last().await.ok()?
148        } else {
149            None
150        }
151    }
152
153    /// Create a resume execution context (from the last checkpoint)
154    /// interrupted_node may be "node_name" or "after_node_name"
155    pub async fn create_resume_execution(
156        &self,
157        interrupted_node: &str,
158    ) -> Option<GraphExecution<S>> {
159        let (state, recursion_count) = self.last_checkpoint_state().await?;
160        let current = interrupted_node
161            .strip_prefix("after_")
162            .unwrap_or(interrupted_node);
163        Some(GraphExecution {
164            state,
165            current_node: current.to_string(),
166            steps: Vec::new(),
167            // M6: resume carries over the recursion budget already consumed before the
168            // interrupt, instead of restarting from zero — otherwise repeated
169            // interrupt→resume could bypass recursion_limit indefinitely.
170            recursion_count,
171            interrupted_at: interrupted_node.to_string(),
172        })
173    }
174
175    pub(super) async fn get_node(&self, name: &str) -> GraphResult<Arc<dyn GraphNode<S>>> {
176        if let Some(node) = self.nodes.get(name) {
177            return Ok(node.clone());
178        }
179        if let Some(node) = self.runtime_nodes.read().await.get(name) {
180            return Ok(node.clone());
181        }
182        Err(GraphError::ExecutionError(format!(
183            "Node '{}' not found",
184            name
185        )))
186    }
187
188    /// Submit a new task for dynamic planning mid-execution
189    pub fn submit_task(&self, description: impl Into<String>) {
190        if let Ok(mut inbox) = self.task_inbox.try_lock() {
191            inbox.push(DynamicTask {
192                id: uuid::Uuid::new_v4().to_string(),
193                description: description.into(),
194            });
195        }
196    }
197
198    /// Inject a runtime node directly
199    pub async fn inject_node(&self, name: &str, node: Arc<dyn GraphNode<S>>) -> GraphResult<()> {
200        self.runtime_nodes
201            .write()
202            .await
203            .insert(name.to_string(), node);
204        Ok(())
205    }
206
207    /// Inject a runtime edge directly
208    pub async fn inject_edge(&self, source: &str, target: &str) -> GraphResult<()> {
209        self.runtime_edges
210            .write()
211            .await
212            .push(GraphEdge::fixed(source, target));
213        Ok(())
214    }
215
216    /// Inject a subgraph as a runtime node
217    pub async fn inject_subgraph<SubS: StateSchema + 'static>(
218        &self,
219        name: &str,
220        subgraph: CompiledGraph<SubS>,
221        input_mapper: impl Fn(&S) -> SubS + Send + Sync + 'static,
222        output_mapper: impl Fn(&SubS, &mut S) + Send + Sync + 'static,
223    ) -> GraphResult<()> {
224        let node = SubgraphNode::new(name, subgraph, input_mapper, output_mapper);
225        self.runtime_nodes
226            .write()
227            .await
228            .insert(name.to_string(), Arc::new(node));
229        Ok(())
230    }
231}
lc_langgraph/compiled/graph.rs

lc_langgraph/compiled/
graph.rs
