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

ironflow_store/entities/
step.rs
