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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143

//! Execution context passed to every node invocation.

use tokio_util::sync::CancellationToken;
use uuid::Uuid;

/// Execution context provided to each [`Node::call`](crate::Node::call) invocation.
///
/// Contains runtime metadata for idempotency, observability, budget awareness,
/// and cancellation. Nodes that don't need context can ignore it with `_ctx`.
///
/// # Idempotency
///
/// Use [`execution_id`](Self::execution_id) as an idempotency key for side effects
/// (API calls, database writes, payments) to guard against duplicate execution
/// after crash recovery. Unlike `attempt_id`, `execution_id` is deterministic
/// and stable across retries — it only changes when the logical checkpoint changes.
///
/// # Example
/// ```rust,no_run
/// # use takeln::{Node, NodeContext, NodeOutput, GraphError};
/// # use async_trait::async_trait;
/// # #[derive(Clone, serde::Serialize, serde::Deserialize)] struct S { v: String }
/// struct MyNode;
///
/// #[async_trait]
/// impl Node<S> for MyNode {
///     async fn call(&self, ctx: NodeContext, mut state: S) -> Result<NodeOutput<S>, GraphError> {
///         println!("Node {} attempt {} (exec: {}, attempt: {})", ctx.node_name, ctx.attempt, ctx.execution_id, ctx.attempt_id);
///         Ok(NodeOutput::bare(state))
///     }
/// }
/// ```
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct NodeContext {
    /// The thread/session ID for this execution.
    pub thread_id: String,
    /// The name of the currently executing node.
    pub node_name: String,
    /// Current retry attempt (0-indexed; 0 = first attempt).
    pub attempt: u8,
    /// Stable ID for this logical node execution.
    /// Deterministic: derived from thread_id + node_name + last_checkpoint_id.
    /// Use for external idempotency keys that must survive crash/resume.
    pub execution_id: String,
    /// Unique ID for this specific attempt (changes on every retry).
    /// Use for logging/tracing, NOT for idempotency.
    pub attempt_id: String,
    /// The checkpoint ID from the most recent save, if any.
    pub last_checkpoint_id: Option<String>,
    /// Remaining budget in EUR, if a budget is configured.
    pub budget_remaining_eur: Option<f64>,
    /// Cancellation token — check `ctx.cancellation.as_ref().map(|t| t.is_cancelled())`.
    pub cancellation: Option<CancellationToken>,
    /// Input provided when resuming from a yield (only set on re-entry).
    pub resumed_input: Option<serde_json::Value>,
}

impl NodeContext {
    /// Create a new NodeContext. Used internally by the executor.
    pub(crate) fn new(
        thread_id: String,
        node_name: String,
        attempt: u8,
        last_checkpoint_id: Option<String>,
        budget_remaining_eur: Option<f64>,
        cancellation: Option<CancellationToken>,
        resumed_input: Option<serde_json::Value>,
    ) -> Self {
        // Deterministic UUID v5 from namespace + "{thread_id}:{node_name}:{checkpoint_id}"
        let seed = format!(
            "{}:{}:{}",
            thread_id,
            node_name,
            last_checkpoint_id.as_deref().unwrap_or("initial")
        );
        let execution_id = Uuid::new_v5(&Uuid::NAMESPACE_OID, seed.as_bytes()).to_string();

        Self {
            thread_id,
            node_name,
            attempt,
            execution_id,
            attempt_id: Uuid::new_v4().to_string(),
            last_checkpoint_id,
            budget_remaining_eur,
            cancellation,
            resumed_input,
        }
    }
}

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

    #[test]
    fn execution_id_stable_across_attempts() {
        let ctx1 = NodeContext::new(
            "thread1".into(),
            "nodeA".into(),
            0,
            Some("cp1".into()),
            None,
            None,
            None,
        );
        let ctx2 = NodeContext::new(
            "thread1".into(),
            "nodeA".into(),
            1,
            Some("cp1".into()),
            None,
            None,
            None,
        );
        assert_eq!(ctx1.execution_id, ctx2.execution_id);
        assert_ne!(ctx1.attempt_id, ctx2.attempt_id);
    }

    #[test]
    fn execution_id_changes_with_checkpoint() {
        let ctx1 = NodeContext::new(
            "thread1".into(),
            "nodeA".into(),
            0,
            Some("cp1".into()),
            None,
            None,
            None,
        );
        let ctx2 = NodeContext::new(
            "thread1".into(),
            "nodeA".into(),
            0,
            Some("cp2".into()),
            None,
            None,
            None,
        );
        assert_ne!(ctx1.execution_id, ctx2.execution_id);
    }
}