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::{ApprovalRequirement, Assignee, FsmState, StepApproval, 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    /// Uncached input token count (agent steps only).
98    pub input_tokens: Option<u64>,
99    /// Input tokens served from the prompt cache (agent steps only).
100    #[serde(default)]
101    pub cache_read_input_tokens: Option<u64>,
102    /// Input tokens written to the prompt cache (agent steps only).
103    #[serde(default)]
104    pub cache_creation_input_tokens: Option<u64>,
105    /// Output token count (agent steps only).
106    pub output_tokens: Option<u64>,
107    /// When the step was created.
108    pub created_at: DateTime<Utc>,
109    /// When the step record was last updated.
110    pub updated_at: DateTime<Utc>,
111    /// When step execution started.
112    pub started_at: Option<DateTime<Utc>>,
113    /// When step execution finished.
114    pub completed_at: Option<DateTime<Utc>>,
115    /// Debug messages (verbose conversation trace), stored as JSON.
116    pub debug_messages: Option<Value>,
117    /// Whether this step is an error handler (`on_error`) rather than a normal step.
118    #[serde(default)]
119    pub is_error_handler: bool,
120    /// When the approval gate on this step expires, if it carries an SLA deadline.
121    ///
122    /// Only ever set while the step is [`StepStatus::AwaitingApproval`]. Cleared
123    /// the moment the gate resolves (approved, rejected, or escalated).
124    #[serde(default)]
125    pub approval_deadline_at: Option<DateTime<Utc>>,
126    /// Index of the next escalation policy to apply when the deadline fires.
127    #[serde(default)]
128    pub approval_stage: u32,
129    /// User or group the approval is currently assigned to, after reassignment.
130    #[serde(default)]
131    pub approval_assignee: Option<Assignee>,
132    /// Approvers the handler required when the gate opened. `None` for a gate
133    /// opened without approvers: one approval resolves it.
134    #[serde(default)]
135    pub approval_requirement: Option<ApprovalRequirement>,
136    /// Votes cast on the approval gate so far, at most one per user.
137    #[serde(default)]
138    pub approvals: Vec<StepApproval>,
139    /// Provider Account the agent step ran under, if any.
140    #[serde(default)]
141    pub account_id: Option<Uuid>,
142}
143
144/// Request to create a new step.
145///
146/// # Examples
147///
148/// ```
149/// use ironflow_store::entities::{NewStep, StepKind, step_trace_id};
150/// use serde_json::json;
151/// use uuid::Uuid;
152///
153/// let run_id = Uuid::nil();
154/// let req = NewStep {
155///     run_id,
156///     trace_id: step_trace_id(run_id, "build", 0),
157///     name: "build".to_string(),
158///     kind: StepKind::Shell,
159///     position: 0,
160///     input: Some(json!({"command": "cargo build"})),
161///     is_error_handler: false,
162/// };
163/// ```
164#[derive(Debug, Clone, Serialize, Deserialize)]
165pub struct NewStep {
166    /// The run this step belongs to.
167    pub run_id: Uuid,
168    /// Deterministic trace ID for log correlation.
169    pub trace_id: Uuid,
170    /// Step name.
171    pub name: String,
172    /// Operation type.
173    pub kind: StepKind,
174    /// Execution order (0-based).
175    pub position: u32,
176    /// Serialized operation configuration.
177    pub input: Option<Value>,
178    /// Whether this step is an error handler (`on_error`).
179    #[serde(default)]
180    pub is_error_handler: bool,
181}
182
183/// Partial update for a step after execution.
184///
185/// Only `Some` fields are applied; `None` fields are left unchanged.
186///
187/// # Examples
188///
189/// ```
190/// use ironflow_store::entities::{StepUpdate, StepStatus};
191/// use serde_json::json;
192///
193/// let update = StepUpdate {
194///     status: Some(StepStatus::Completed),
195///     output: Some(json!({"stdout": "ok"})),
196///     ..StepUpdate::default()
197/// };
198/// ```
199#[derive(Debug, Clone, Default, Serialize, Deserialize)]
200pub struct StepUpdate {
201    /// New status.
202    pub status: Option<StepStatus>,
203    /// Operation output.
204    pub output: Option<Value>,
205    /// Error message.
206    pub error: Option<String>,
207    /// Execution duration.
208    pub duration_ms: Option<u64>,
209    /// Cost in USD.
210    pub cost_usd: Option<Decimal>,
211    /// Uncached input token count.
212    pub input_tokens: Option<u64>,
213    /// Input tokens served from the prompt cache.
214    #[serde(default)]
215    pub cache_read_input_tokens: Option<u64>,
216    /// Input tokens written to the prompt cache.
217    #[serde(default)]
218    pub cache_creation_input_tokens: Option<u64>,
219    /// Output token count.
220    pub output_tokens: Option<u64>,
221    /// When execution started.
222    pub started_at: Option<DateTime<Utc>>,
223    /// When execution completed.
224    pub completed_at: Option<DateTime<Utc>>,
225    /// Debug messages (verbose conversation trace), stored as JSON.
226    pub debug_messages: Option<Value>,
227    /// New approval deadline. `None` leaves it unchanged — use
228    /// `clear_approval_deadline` to remove it.
229    #[serde(default)]
230    pub approval_deadline_at: Option<DateTime<Utc>>,
231    /// New escalation stage index.
232    #[serde(default)]
233    pub approval_stage: Option<u32>,
234    /// New approval assignee.
235    #[serde(default)]
236    pub approval_assignee: Option<Assignee>,
237    /// Approval requirement recorded when the gate opened.
238    #[serde(default)]
239    pub approval_requirement: Option<ApprovalRequirement>,
240    /// Clear the approval deadline (sets it to `NULL`). Wins over
241    /// `approval_deadline_at` when both are set.
242    #[serde(default)]
243    pub clear_approval_deadline: bool,
244    /// Provider Account the agent step ran under.
245    #[serde(default, skip_serializing_if = "Option::is_none")]
246    pub account_id: Option<Uuid>,
247}
248
249#[cfg(test)]
250mod tests {
251    use super::*;
252    use serde_json::json;
253
254    #[test]
255    fn newstep_serde_roundtrip() {
256        let new_step = NewStep {
257            run_id: Uuid::nil(),
258            trace_id: step_trace_id(Uuid::nil(), "build", 0),
259            name: "build".to_string(),
260            kind: StepKind::Shell,
261            position: 0,
262            input: Some(json!({"command": "cargo build"})),
263            is_error_handler: false,
264        };
265
266        let json = serde_json::to_string(&new_step).expect("serialize");
267        let back: NewStep = serde_json::from_str(&json).expect("deserialize");
268
269        assert_eq!(back.run_id, new_step.run_id);
270        assert_eq!(back.name, new_step.name);
271        assert_eq!(back.kind, new_step.kind);
272        assert_eq!(back.position, new_step.position);
273        assert_eq!(back.input, new_step.input);
274    }
275
276    #[test]
277    fn step_serde_preserves_all_fields() {
278        use crate::entities::FsmState;
279        use chrono::Utc;
280
281        let now = Utc::now();
282        let run_id = Uuid::now_v7();
283        let step = Step {
284            id: Uuid::now_v7(),
285            trace_id: step_trace_id(run_id, "test-step", 1),
286            run_id,
287            name: "test-step".to_string(),
288            kind: StepKind::Agent,
289            position: 1,
290            status: FsmState::new(StepStatus::Completed, Uuid::now_v7()),
291            attempt: 2,
292            input: Some(json!({"input": "data"})),
293            output: Some(json!({"output": "result"})),
294            error: None,
295            duration_ms: 2500,
296            cost_usd: Decimal::new(150, 2),
297            input_tokens: Some(100),
298            cache_read_input_tokens: Some(4000),
299            cache_creation_input_tokens: Some(300),
300            output_tokens: Some(200),
301            created_at: now,
302            updated_at: now,
303            started_at: Some(now),
304            completed_at: Some(now),
305            debug_messages: None,
306            is_error_handler: false,
307            approval_deadline_at: Some(now),
308            approval_stage: 2,
309            approval_assignee: Some(Assignee::group("sre-oncall")),
310            approval_requirement: Some(ApprovalRequirement {
311                reason: Some("amount > 10k".to_string()),
312                required_approvers: 2,
313                approver_groups: vec!["finance".to_string()],
314            }),
315            approvals: vec![StepApproval {
316                user_id: Uuid::now_v7(),
317                approved_by: "alice".to_string(),
318                at: now,
319            }],
320            account_id: Some(Uuid::now_v7()),
321        };
322
323        let json = serde_json::to_string(&step).expect("serialize");
324        let back: Step = serde_json::from_str(&json).expect("deserialize");
325
326        assert_eq!(back.id, step.id);
327        assert_eq!(back.run_id, step.run_id);
328        assert_eq!(back.name, step.name);
329        assert_eq!(back.kind, step.kind);
330        assert_eq!(back.position, step.position);
331        assert_eq!(back.status.state, step.status.state);
332        assert_eq!(back.attempt, step.attempt);
333        assert_eq!(back.input, step.input);
334        assert_eq!(back.output, step.output);
335        assert_eq!(back.error, step.error);
336        assert_eq!(back.duration_ms, step.duration_ms);
337        assert_eq!(back.cost_usd, step.cost_usd);
338        assert_eq!(back.input_tokens, step.input_tokens);
339        assert_eq!(back.cache_read_input_tokens, step.cache_read_input_tokens);
340        assert_eq!(
341            back.cache_creation_input_tokens,
342            step.cache_creation_input_tokens
343        );
344        assert_eq!(back.output_tokens, step.output_tokens);
345        assert_eq!(back.account_id, step.account_id);
346        assert_eq!(back.approval_deadline_at, step.approval_deadline_at);
347        assert_eq!(back.approval_stage, step.approval_stage);
348        assert_eq!(back.approval_assignee, step.approval_assignee);
349        assert_eq!(back.approval_requirement, step.approval_requirement);
350        assert_eq!(back.approvals, step.approvals);
351    }
352
353    #[test]
354    fn step_serde_defaults_approval_fields_when_absent() {
355        let run_id = Uuid::now_v7();
356        let payload = json!({
357            "id": Uuid::now_v7(),
358            "trace_id": step_trace_id(run_id, "legacy", 0),
359            "run_id": run_id,
360            "name": "legacy",
361            "kind": "shell",
362            "position": 0,
363            "status": {"state": "pending", "state_machine_id": Uuid::now_v7()},
364            "input": null,
365            "output": null,
366            "error": null,
367            "duration_ms": 0,
368            "cost_usd": 0.0,
369            "input_tokens": null,
370            "cache_read_input_tokens": null,
371            "cache_creation_input_tokens": null,
372            "output_tokens": null,
373            "created_at": "2026-09-21T12:00:00Z",
374            "updated_at": "2026-09-21T12:00:00Z",
375            "started_at": null,
376            "completed_at": null,
377            "debug_messages": null
378        });
379
380        let step: Step = serde_json::from_value(payload).expect("deserialize");
381
382        assert_eq!(step.approval_stage, 0);
383        assert!(step.cache_read_input_tokens.is_none());
384        assert!(step.cache_creation_input_tokens.is_none());
385        assert!(step.approval_deadline_at.is_none());
386        assert!(step.approval_assignee.is_none());
387        assert!(step.approval_requirement.is_none());
388        assert!(step.approvals.is_empty());
389    }
390
391    #[test]
392    fn stepupdate_default_is_no_changes() {
393        let update = StepUpdate::default();
394        assert!(update.status.is_none());
395        assert!(update.output.is_none());
396        assert!(update.error.is_none());
397        assert!(update.duration_ms.is_none());
398        assert!(update.cost_usd.is_none());
399        assert!(update.input_tokens.is_none());
400        assert!(update.cache_read_input_tokens.is_none());
401        assert!(update.cache_creation_input_tokens.is_none());
402        assert!(update.output_tokens.is_none());
403        assert!(update.started_at.is_none());
404        assert!(update.completed_at.is_none());
405        assert!(update.debug_messages.is_none());
406        assert!(update.approval_deadline_at.is_none());
407        assert!(update.approval_stage.is_none());
408        assert!(update.approval_assignee.is_none());
409        assert!(update.approval_requirement.is_none());
410        assert!(!update.clear_approval_deadline);
411    }
412
413    #[test]
414    fn stepupdate_serde_roundtrip() {
415        let update = StepUpdate {
416            status: Some(StepStatus::Completed),
417            output: Some(json!({"result": "ok"})),
418            error: None,
419            duration_ms: Some(1000),
420            cost_usd: Some(Decimal::new(50, 2)),
421            input_tokens: Some(50),
422            cache_read_input_tokens: Some(2000),
423            cache_creation_input_tokens: Some(150),
424            output_tokens: Some(75),
425            started_at: None,
426            completed_at: None,
427            debug_messages: None,
428            approval_deadline_at: Some(Utc::now()),
429            approval_stage: Some(1),
430            approval_assignee: Some(Assignee::group("sre-oncall")),
431            approval_requirement: Some(ApprovalRequirement::default()),
432            clear_approval_deadline: false,
433            account_id: Some(Uuid::now_v7()),
434        };
435
436        let json = serde_json::to_string(&update).expect("serialize");
437        let back: StepUpdate = serde_json::from_str(&json).expect("deserialize");
438
439        assert_eq!(back.status, update.status);
440        assert_eq!(back.output, update.output);
441        assert_eq!(back.duration_ms, update.duration_ms);
442        assert_eq!(back.account_id, update.account_id);
443        assert_eq!(back.cost_usd, update.cost_usd);
444        assert_eq!(back.input_tokens, update.input_tokens);
445        assert_eq!(back.cache_read_input_tokens, update.cache_read_input_tokens);
446        assert_eq!(
447            back.cache_creation_input_tokens,
448            update.cache_creation_input_tokens
449        );
450        assert_eq!(back.output_tokens, update.output_tokens);
451        assert_eq!(back.approval_deadline_at, update.approval_deadline_at);
452        assert_eq!(back.approval_stage, update.approval_stage);
453        assert_eq!(back.approval_assignee, update.approval_assignee);
454        assert_eq!(back.approval_requirement, update.approval_requirement);
455        assert_eq!(back.clear_approval_deadline, update.clear_approval_deadline);
456    }
457
458    #[test]
459    fn stepupdate_deserializes_without_cache_fields() {
460        let payload = json!({
461            "status": "completed",
462            "output": null,
463            "error": null,
464            "duration_ms": 10,
465            "cost_usd": null,
466            "input_tokens": 12,
467            "output_tokens": 3,
468            "started_at": null,
469            "completed_at": null,
470            "debug_messages": null
471        });
472
473        let update: StepUpdate = serde_json::from_value(payload).expect("deserialize");
474
475        assert_eq!(update.input_tokens, Some(12));
476        assert!(update.cache_read_input_tokens.is_none());
477        assert!(update.cache_creation_input_tokens.is_none());
478    }
479
480    #[test]
481    fn trace_id_is_deterministic() {
482        let run_id = Uuid::nil();
483        let id1 = step_trace_id(run_id, "build", 0);
484        let id2 = step_trace_id(run_id, "build", 0);
485        assert_eq!(id1, id2);
486    }
487
488    #[test]
489    fn trace_id_differs_for_different_inputs() {
490        let run_id = Uuid::nil();
491        let a = step_trace_id(run_id, "build", 0);
492        let b = step_trace_id(run_id, "test", 0);
493        let c = step_trace_id(run_id, "build", 1);
494        let d = step_trace_id(Uuid::max(), "build", 0);
495
496        assert_ne!(a, b);
497        assert_ne!(a, c);
498        assert_ne!(a, d);
499    }
500
501    #[test]
502    fn trace_id_is_uuid_v5() {
503        let id = step_trace_id(Uuid::nil(), "build", 0);
504        assert_eq!(id.get_version_num(), 5);
505    }
506}
ironflow_store/entities/step.rs

ironflow_store/entities/
step.rs
