simple-agents-workflow 0.2.35

Workflow IR and validation for SimpleAgents
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
144
145
146
147
148

use std::sync::{Arc, Mutex};

use thiserror::Error;

use crate::trace::{
    TraceEvent, TraceEventKind, TraceTerminalStatus, WorkflowTrace, WorkflowTraceMetadata,
};

/// Thread-safe in-memory recorder for workflow trace events.
#[derive(Debug, Clone)]
pub struct TraceRecorder {
    state: Arc<Mutex<TraceRecorderState>>,
}

#[derive(Debug, Clone)]
struct TraceRecorderState {
    trace: WorkflowTrace,
    next_seq: u64,
    finalized: bool,
}

impl TraceRecorder {
    /// Creates a new recorder with empty event history.
    pub fn new(metadata: WorkflowTraceMetadata) -> Self {
        Self {
            state: Arc::new(Mutex::new(TraceRecorderState {
                trace: WorkflowTrace {
                    metadata,
                    events: Vec::new(),
                },
                next_seq: 0,
                finalized: false,
            })),
        }
    }

    /// Appends an arbitrary event to the trace.
    pub fn append_event(
        &self,
        timestamp_unix_ms: u64,
        kind: TraceEventKind,
    ) -> Result<TraceEvent, TraceRecordError> {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        if state.finalized {
            return Err(TraceRecordError::AlreadyFinalized);
        }

        let event = TraceEvent {
            seq: state.next_seq,
            timestamp_unix_ms,
            kind,
        };

        state.next_seq = state.next_seq.saturating_add(1);
        state.trace.events.push(event.clone());
        Ok(event)
    }

    /// Appends a node enter event.
    pub fn record_node_enter(
        &self,
        timestamp_unix_ms: u64,
        node_id: impl Into<String>,
    ) -> Result<TraceEvent, TraceRecordError> {
        self.append_event(
            timestamp_unix_ms,
            TraceEventKind::NodeEnter {
                node_id: node_id.into(),
            },
        )
    }

    /// Appends a node exit event.
    pub fn record_node_exit(
        &self,
        timestamp_unix_ms: u64,
        node_id: impl Into<String>,
    ) -> Result<TraceEvent, TraceRecordError> {
        self.append_event(
            timestamp_unix_ms,
            TraceEventKind::NodeExit {
                node_id: node_id.into(),
            },
        )
    }

    /// Appends a node error event.
    pub fn record_node_error(
        &self,
        timestamp_unix_ms: u64,
        node_id: impl Into<String>,
        message: impl Into<String>,
    ) -> Result<TraceEvent, TraceRecordError> {
        self.append_event(
            timestamp_unix_ms,
            TraceEventKind::NodeError {
                node_id: node_id.into(),
                message: message.into(),
            },
        )
    }

    /// Appends a workflow terminal event.
    pub fn record_terminal(
        &self,
        timestamp_unix_ms: u64,
        status: TraceTerminalStatus,
    ) -> Result<TraceEvent, TraceRecordError> {
        self.append_event(timestamp_unix_ms, TraceEventKind::Terminal { status })
    }

    /// Returns a snapshot of the current trace.
    pub fn snapshot(&self) -> WorkflowTrace {
        let state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        state.trace.clone()
    }

    /// Finalizes the recorder and returns the immutable trace.
    pub fn finalize(&self, finished_at_unix_ms: u64) -> Result<WorkflowTrace, TraceRecordError> {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());

        if state.finalized {
            return Err(TraceRecordError::AlreadyFinalized);
        }

        state.trace.metadata.finished_at_unix_ms = Some(finished_at_unix_ms);
        state.finalized = true;
        Ok(state.trace.clone())
    }
}

/// Trace recorder failures.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
pub enum TraceRecordError {
    /// Mutation attempted after recorder finalization.
    #[error("trace recorder is already finalized")]
    AlreadyFinalized,
}