takeln 0.11.0

Typed Rust runtime for durable DAG-based agent workflows
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

use async_trait::async_trait;
use takeln::{Graph, GraphError, InMemoryCheckpointer, Node, NodeContext, NodeOutput, YieldRequest};

#[derive(Clone, serde::Serialize, serde::Deserialize, Default, Debug)]
struct ResumableState {
    step_completed: Option<String>,
}

struct NodeA;

#[async_trait]
impl Node<ResumableState> for NodeA {
    async fn call(
        &self,
        _ctx: NodeContext,
        mut state: ResumableState,
    ) -> Result<NodeOutput<ResumableState>, GraphError> {
        println!("NodeA completed successfully.");
        state.step_completed = Some("NodeA".to_string());
        Ok(NodeOutput::bare(state))
    }
}

struct YieldingNodeB;

#[async_trait]
impl Node<ResumableState> for YieldingNodeB {
    async fn call(&self, _ctx: NodeContext, _state: ResumableState) -> Result<NodeOutput<ResumableState>, GraphError> {
        println!("NodeB encountered an interruption (e.g. API down, need manual intervention). Yielding...");
        Err(GraphError::Yield(YieldRequest::simple("Interrupted")))
    }
}

struct SuccessfulNodeB;

#[async_trait]
impl Node<ResumableState> for SuccessfulNodeB {
    async fn call(
        &self,
        _ctx: NodeContext,
        mut state: ResumableState,
    ) -> Result<NodeOutput<ResumableState>, GraphError> {
        println!("NodeB executed successfully after resumption!");
        state.step_completed = Some("NodeB".to_string());
        Ok(NodeOutput::bare(state))
    }
}

#[tokio::main]
async fn main() {
    // 1. Initial graph setup where Node B will yield/fail
    let mut initial_graph = Graph::new();
    initial_graph.add_node("NodeA", NodeA);
    initial_graph.add_node("NodeB", YieldingNodeB);
    initial_graph.add_edge("NodeA", "NodeB");
    initial_graph.add_edge("NodeB", "__END__");

    let checkpointer = InMemoryCheckpointer::new();

    println!("--- Initial Run (will yield at NodeB) ---");
    let state = ResumableState::default();
    initial_graph
        .run("thread_chk", state, "NodeA", &checkpointer, None)
        .await
        .unwrap();

    // 2. Resume run using a graph where Node B is now healthy/successful
    let mut resume_graph = Graph::new();
    resume_graph.add_node("NodeB", SuccessfulNodeB);
    resume_graph.add_edge("NodeB", "__END__");

    println!("\n--- Resuming Run (from NodeB) using first-class resume API ---");
    let final_state = resume_graph
        .resume("thread_chk", &checkpointer, None)
        .await
        .unwrap()
        .unwrap();
    println!("Final State: {:?}", final_state);
}