ironflow_store/entities/
step.rs

1//! [`Step`] entity and related request/update types.
2
3use chrono::{DateTime, Utc};
4use rust_decimal::Decimal;
5use serde::{Deserialize, Serialize};
6use serde_json::Value;
7use uuid::Uuid;
8
9use super::{FsmState, StepKind, StepStatus};
10
11/// Attempt number assigned to steps deserialized from payloads predating the
12/// `attempt` field.
13fn default_attempt() -> u32 {
14    1
15}
16
17/// Generate a deterministic trace ID for a step.
18///
19/// Uses UUIDv5 with `NAMESPACE_OID` and the input
20/// `"{run_id}:{name}:{position}"`, so the same run replayed with the same
21/// steps always produces the same trace IDs.
22///
23/// # Examples
24///
25/// ```
26/// use ironflow_store::entities::step_trace_id;
27/// use uuid::Uuid;
28///
29/// let run_id = Uuid::nil();
30/// let id1 = step_trace_id(run_id, "build", 0);
31/// let id2 = step_trace_id(run_id, "build", 0);
32/// assert_eq!(id1, id2);
33///
34/// let id3 = step_trace_id(run_id, "test", 1);
35/// assert_ne!(id1, id3);
36/// ```
37pub fn step_trace_id(run_id: Uuid, name: &str, position: u32) -> Uuid {
38    Uuid::new_v5(
39        &Uuid::NAMESPACE_OID,
40        format!("{run_id}:{name}:{position}").as_bytes(),
41    )
42}
43
44/// A single operation within a run.
45///
46/// Steps are executed sequentially in order of [`position`](Step::position).
47///
48/// # Examples
49///
50/// ```
51/// use ironflow_store::entities::Step;
52///
53/// // Steps are created by RunStore::create_step, not directly.
54/// ```
55#[derive(Debug, Clone, Serialize, Deserialize)]
56#[non_exhaustive]
57pub struct Step {
58    /// Unique identifier (UUIDv7).
59    pub id: Uuid,
60    /// Deterministic trace ID for log correlation.
61    ///
62    /// Generated as `UUIDv5(NAMESPACE_OID, "{run_id}:{name}:{position}")` so
63    /// the same run replayed with the same steps always produces the same IDs.
64    pub trace_id: Uuid,
65    /// The run this step belongs to.
66    pub run_id: Uuid,
67    /// Human-readable step name (e.g. "build", "test", "review").
68    pub name: String,
69    /// The type of operation.
70    pub kind: StepKind,
71    /// Execution wave within the run (0-based).
72    ///
73    /// In linear flows, this strictly increases (0, 1, 2, ...).
74    /// In DAGs with parallel execution, steps at the same wave share
75    /// the same position and execute concurrently. Use
76    /// `step_dependencies` to determine the actual execution order.
77    pub position: u32,
78    /// Current FSM status — embeds state + state_machine_id for SQL-side transitions.
79    pub status: FsmState<StepStatus>,
80    /// Which run attempt produced this step (1-based).
81    ///
82    /// A run retried twice holds steps with `attempt` 1, 2 and 3. Steps from
83    /// earlier attempts are kept for inspection and are never replayed.
84    /// Derived by the store from `Run::retry_count` at creation time.
85    #[serde(default = "default_attempt")]
86    pub attempt: u32,
87    /// Serialized operation configuration.
88    pub input: Option<Value>,
89    /// Serialized operation output.
90    pub output: Option<Value>,
91    /// Error message if the step failed.
92    pub error: Option<String>,
93    /// Wall-clock execution duration in milliseconds.
94    pub duration_ms: u64,
95    /// Cost in USD (agent steps only).
96    pub cost_usd: Decimal,
97    /// Input token count (agent steps only).
98    pub input_tokens: Option<u64>,
99    /// Output token count (agent steps only).
100    pub output_tokens: Option<u64>,
101    /// When the step was created.
102    pub created_at: DateTime<Utc>,
103    /// When the step record was last updated.
104    pub updated_at: DateTime<Utc>,
105    /// When step execution started.
106    pub started_at: Option<DateTime<Utc>>,
107    /// When step execution finished.
108    pub completed_at: Option<DateTime<Utc>>,
109    /// Debug messages (verbose conversation trace), stored as JSON.
110    pub debug_messages: Option<Value>,
111    /// Whether this step is an error handler (`on_error`) rather than a normal step.
112    #[serde(default)]
113    pub is_error_handler: bool,
114}
115
116/// Request to create a new step.
117///
118/// # Examples
119///
120/// ```
121/// use ironflow_store::entities::{NewStep, StepKind, step_trace_id};
122/// use serde_json::json;
123/// use uuid::Uuid;
124///
125/// let run_id = Uuid::nil();
126/// let req = NewStep {
127///     run_id,
128///     trace_id: step_trace_id(run_id, "build", 0),
129///     name: "build".to_string(),
130///     kind: StepKind::Shell,
131///     position: 0,
132///     input: Some(json!({"command": "cargo build"})),
133///     is_error_handler: false,
134/// };
135/// ```
136#[derive(Debug, Clone, Serialize, Deserialize)]
137pub struct NewStep {
138    /// The run this step belongs to.
139    pub run_id: Uuid,
140    /// Deterministic trace ID for log correlation.
141    pub trace_id: Uuid,
142    /// Step name.
143    pub name: String,
144    /// Operation type.
145    pub kind: StepKind,
146    /// Execution order (0-based).
147    pub position: u32,
148    /// Serialized operation configuration.
149    pub input: Option<Value>,
150    /// Whether this step is an error handler (`on_error`).
151    #[serde(default)]
152    pub is_error_handler: bool,
153}
154
155/// Partial update for a step after execution.
156///
157/// Only `Some` fields are applied; `None` fields are left unchanged.
158///
159/// # Examples
160///
161/// ```
162/// use ironflow_store::entities::{StepUpdate, StepStatus};
163/// use serde_json::json;
164///
165/// let update = StepUpdate {
166///     status: Some(StepStatus::Completed),
167///     output: Some(json!({"stdout": "ok"})),
168///     ..StepUpdate::default()
169/// };
170/// ```
171#[derive(Debug, Clone, Default, Serialize, Deserialize)]
172pub struct StepUpdate {
173    /// New status.
174    pub status: Option<StepStatus>,
175    /// Operation output.
176    pub output: Option<Value>,
177    /// Error message.
178    pub error: Option<String>,
179    /// Execution duration.
180    pub duration_ms: Option<u64>,
181    /// Cost in USD.
182    pub cost_usd: Option<Decimal>,
183    /// Input token count.
184    pub input_tokens: Option<u64>,
185    /// Output token count.
186    pub output_tokens: Option<u64>,
187    /// When execution started.
188    pub started_at: Option<DateTime<Utc>>,
189    /// When execution completed.
190    pub completed_at: Option<DateTime<Utc>>,
191    /// Debug messages (verbose conversation trace), stored as JSON.
192    pub debug_messages: Option<Value>,
193}
194
195#[cfg(test)]
196mod tests {
197    use super::*;
198    use serde_json::json;
199
200    #[test]
201    fn newstep_serde_roundtrip() {
202        let new_step = NewStep {
203            run_id: Uuid::nil(),
204            trace_id: step_trace_id(Uuid::nil(), "build", 0),
205            name: "build".to_string(),
206            kind: StepKind::Shell,
207            position: 0,
208            input: Some(json!({"command": "cargo build"})),
209            is_error_handler: false,
210        };
211
212        let json = serde_json::to_string(&new_step).expect("serialize");
213        let back: NewStep = serde_json::from_str(&json).expect("deserialize");
214
215        assert_eq!(back.run_id, new_step.run_id);
216        assert_eq!(back.name, new_step.name);
217        assert_eq!(back.kind, new_step.kind);
218        assert_eq!(back.position, new_step.position);
219        assert_eq!(back.input, new_step.input);
220    }
221
222    #[test]
223    fn step_serde_preserves_all_fields() {
224        use crate::entities::FsmState;
225        use chrono::Utc;
226
227        let now = Utc::now();
228        let run_id = Uuid::now_v7();
229        let step = Step {
230            id: Uuid::now_v7(),
231            trace_id: step_trace_id(run_id, "test-step", 1),
232            run_id,
233            name: "test-step".to_string(),
234            kind: StepKind::Agent,
235            position: 1,
236            status: FsmState::new(StepStatus::Completed, Uuid::now_v7()),
237            attempt: 2,
238            input: Some(json!({"input": "data"})),
239            output: Some(json!({"output": "result"})),
240            error: None,
241            duration_ms: 2500,
242            cost_usd: Decimal::new(150, 2),
243            input_tokens: Some(100),
244            output_tokens: Some(200),
245            created_at: now,
246            updated_at: now,
247            started_at: Some(now),
248            completed_at: Some(now),
249            debug_messages: None,
250            is_error_handler: false,
251        };
252
253        let json = serde_json::to_string(&step).expect("serialize");
254        let back: Step = serde_json::from_str(&json).expect("deserialize");
255
256        assert_eq!(back.id, step.id);
257        assert_eq!(back.run_id, step.run_id);
258        assert_eq!(back.name, step.name);
259        assert_eq!(back.kind, step.kind);
260        assert_eq!(back.position, step.position);
261        assert_eq!(back.status.state, step.status.state);
262        assert_eq!(back.attempt, step.attempt);
263        assert_eq!(back.input, step.input);
264        assert_eq!(back.output, step.output);
265        assert_eq!(back.error, step.error);
266        assert_eq!(back.duration_ms, step.duration_ms);
267        assert_eq!(back.cost_usd, step.cost_usd);
268        assert_eq!(back.input_tokens, step.input_tokens);
269        assert_eq!(back.output_tokens, step.output_tokens);
270    }
271
272    #[test]
273    fn stepupdate_default_is_no_changes() {
274        let update = StepUpdate::default();
275        assert!(update.status.is_none());
276        assert!(update.output.is_none());
277        assert!(update.error.is_none());
278        assert!(update.duration_ms.is_none());
279        assert!(update.cost_usd.is_none());
280        assert!(update.input_tokens.is_none());
281        assert!(update.output_tokens.is_none());
282        assert!(update.started_at.is_none());
283        assert!(update.completed_at.is_none());
284        assert!(update.debug_messages.is_none());
285    }
286
287    #[test]
288    fn stepupdate_serde_roundtrip() {
289        let update = StepUpdate {
290            status: Some(StepStatus::Completed),
291            output: Some(json!({"result": "ok"})),
292            error: None,
293            duration_ms: Some(1000),
294            cost_usd: Some(Decimal::new(50, 2)),
295            input_tokens: Some(50),
296            output_tokens: Some(75),
297            started_at: None,
298            completed_at: None,
299            debug_messages: None,
300        };
301
302        let json = serde_json::to_string(&update).expect("serialize");
303        let back: StepUpdate = serde_json::from_str(&json).expect("deserialize");
304
305        assert_eq!(back.status, update.status);
306        assert_eq!(back.output, update.output);
307        assert_eq!(back.duration_ms, update.duration_ms);
308        assert_eq!(back.cost_usd, update.cost_usd);
309        assert_eq!(back.input_tokens, update.input_tokens);
310        assert_eq!(back.output_tokens, update.output_tokens);
311    }
312
313    #[test]
314    fn trace_id_is_deterministic() {
315        let run_id = Uuid::nil();
316        let id1 = step_trace_id(run_id, "build", 0);
317        let id2 = step_trace_id(run_id, "build", 0);
318        assert_eq!(id1, id2);
319    }
320
321    #[test]
322    fn trace_id_differs_for_different_inputs() {
323        let run_id = Uuid::nil();
324        let a = step_trace_id(run_id, "build", 0);
325        let b = step_trace_id(run_id, "test", 0);
326        let c = step_trace_id(run_id, "build", 1);
327        let d = step_trace_id(Uuid::max(), "build", 0);
328
329        assert_ne!(a, b);
330        assert_ne!(a, c);
331        assert_ne!(a, d);
332    }
333
334    #[test]
335    fn trace_id_is_uuid_v5() {
336        let id = step_trace_id(Uuid::nil(), "build", 0);
337        assert_eq!(id.get_version_num(), 5);
338    }
339}
ironflow_store/entities/step.rs

ironflow_store/entities/
step.rs
