ironflow_api/entities/
step.rs

1//! Step-related DTOs.
2
3use chrono::{DateTime, Utc};
4use ironflow_store::models::{
5    ApprovalRequirement, Assignee, Step, StepApproval, StepKind, StepStatus,
6};
7use rust_decimal::Decimal;
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10use uuid::Uuid;
11
12use super::ArtifactResponse;
13
14/// Step response DTO — public API representation of a step.
15///
16/// # Examples
17///
18/// ```
19/// use ironflow_store::models::Step;
20/// use ironflow_api::entities::StepResponse;
21/// ```
22#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
23#[derive(Debug, Serialize, Deserialize)]
24pub struct StepResponse {
25    /// Unique step identifier.
26    pub id: Uuid,
27    /// Deterministic trace ID for log correlation.
28    pub trace_id: Uuid,
29    /// Parent run ID.
30    pub run_id: Uuid,
31    /// Step name.
32    pub name: String,
33    /// Step operation type.
34    #[cfg_attr(feature = "openapi", schema(value_type = String))]
35    pub kind: StepKind,
36    /// Execution order (0-based).
37    pub position: u32,
38    /// Current status.
39    pub status: StepStatus,
40    /// Which run attempt produced this step (1-based).
41    ///
42    /// A run retried twice exposes steps with `attempt` 1, 2 and 3. Steps from
43    /// earlier attempts are kept so a failed attempt stays inspectable.
44    pub attempt: u32,
45    /// Input configuration.
46    #[cfg_attr(feature = "openapi", schema(value_type = Option<std::collections::HashMap<String, serde_json::Value>>))]
47    pub input: Option<Value>,
48    /// Step output.
49    #[cfg_attr(feature = "openapi", schema(value_type = Option<std::collections::HashMap<String, serde_json::Value>>))]
50    pub output: Option<Value>,
51    /// Optional error message.
52    pub error: Option<String>,
53    /// Execution duration in milliseconds.
54    pub duration_ms: u64,
55    /// Cost in USD.
56    #[cfg_attr(feature = "openapi", schema(value_type = f64))]
57    pub cost_usd: Decimal,
58    /// Uncached input token count (agent steps).
59    pub input_tokens: Option<u64>,
60    /// Input tokens served from the prompt cache (agent steps).
61    pub cache_read_input_tokens: Option<u64>,
62    /// Input tokens written to the prompt cache (agent steps).
63    pub cache_creation_input_tokens: Option<u64>,
64    /// Output token count (agent steps).
65    pub output_tokens: Option<u64>,
66    /// When created.
67    pub created_at: DateTime<Utc>,
68    /// When updated.
69    pub updated_at: DateTime<Utc>,
70    /// When execution started.
71    pub started_at: Option<DateTime<Utc>>,
72    /// When execution completed.
73    pub completed_at: Option<DateTime<Utc>>,
74    /// IDs of steps this step depends on (direct dependencies).
75    pub dependencies: Vec<Uuid>,
76    /// Verbose conversation trace for agent steps (thinking blocks, tool
77    /// calls, tool results, per-turn usage). `None` when verbose mode was
78    /// off or the step is not an agent step.
79    #[cfg_attr(feature = "openapi", schema(value_type = Option<serde_json::Value>))]
80    pub debug_messages: Option<Value>,
81    /// Files this step produced, downloadable through the artifact route.
82    ///
83    /// Empty when the step produced none or when artifact storage is not
84    /// configured on the server.
85    #[serde(default)]
86    pub artifacts: Vec<ArtifactResponse>,
87    /// When this approval gate expires, if it carries an SLA deadline.
88    pub approval_deadline_at: Option<DateTime<Utc>>,
89    /// Seconds left before the gate escalates. Clamped at 0, `None` when the
90    /// step has no deadline.
91    pub approval_seconds_remaining: Option<i64>,
92    /// Who the approval is currently assigned to.
93    ///
94    /// Serialized as a prefixed string: `user:{name}` or `group:{name}`.
95    #[cfg_attr(feature = "openapi", schema(value_type = Option<String>))]
96    pub approval_assignee: Option<Assignee>,
97    /// Approvers the workflow handler required when the gate opened. `None`
98    /// for a gate opened without approvers.
99    #[serde(default)]
100    pub approval_requirement: Option<ApprovalRequirement>,
101    /// Votes cast on the approval gate so far, at most one per user.
102    #[serde(default)]
103    pub approvals: Vec<StepApproval>,
104    /// Distinct approvals the gate needs: the requirement's count, `1` for an
105    /// approval step without rules, `None` for any other step kind.
106    #[serde(default)]
107    pub approvals_required: Option<u32>,
108    /// Provider Account the agent step ran under, if any.
109    #[serde(default)]
110    pub account_id: Option<Uuid>,
111}
112
113impl StepResponse {
114    /// Build a response from a step entity with pre-resolved dependencies.
115    ///
116    /// Artifacts are left empty; use
117    /// [`with_dependencies_and_artifacts`](Self::with_dependencies_and_artifacts)
118    /// when they have been fetched.
119    pub fn with_dependencies(step: Step, dependencies: Vec<Uuid>) -> Self {
120        Self::with_dependencies_and_artifacts(step, dependencies, Vec::new())
121    }
122
123    /// Build a response from a step entity with its dependencies and artifacts.
124    pub fn with_dependencies_and_artifacts(
125        step: Step,
126        dependencies: Vec<Uuid>,
127        artifacts: Vec<ArtifactResponse>,
128    ) -> Self {
129        let approval_seconds_remaining = step
130            .approval_deadline_at
131            .map(|at| (at - Utc::now()).num_seconds().max(0));
132        let approvals_required = match (&step.kind, &step.approval_requirement) {
133            // The first valid answer resolves a human input: its requirement
134            // only restricts who may answer.
135            (StepKind::HumanInput, _) => None,
136            (_, Some(requirement)) => Some(requirement.required_approvers),
137            (StepKind::Approval, None) => Some(1),
138            _ => None,
139        };
140
141        StepResponse {
142            id: step.id,
143            trace_id: step.trace_id,
144            run_id: step.run_id,
145            name: step.name,
146            kind: step.kind,
147            position: step.position,
148            status: step.status.state,
149            attempt: step.attempt,
150            input: step.input,
151            output: step.output,
152            error: step.error,
153            duration_ms: step.duration_ms,
154            cost_usd: step.cost_usd,
155            input_tokens: step.input_tokens,
156            cache_read_input_tokens: step.cache_read_input_tokens,
157            cache_creation_input_tokens: step.cache_creation_input_tokens,
158            output_tokens: step.output_tokens,
159            created_at: step.created_at,
160            updated_at: step.updated_at,
161            started_at: step.started_at,
162            completed_at: step.completed_at,
163            dependencies,
164            debug_messages: step.debug_messages,
165            artifacts,
166            approval_deadline_at: step.approval_deadline_at,
167            approval_seconds_remaining,
168            approval_assignee: step.approval_assignee,
169            approval_requirement: step.approval_requirement,
170            approvals: step.approvals,
171            approvals_required,
172            account_id: step.account_id,
173        }
174    }
175}
176
177impl From<Step> for StepResponse {
178    fn from(step: Step) -> Self {
179        Self::with_dependencies(step, Vec::new())
180    }
181}
182
183#[cfg(test)]
184mod tests {
185    use std::collections::HashMap;
186
187    use chrono::TimeDelta;
188    use ironflow_store::memory::InMemoryStore;
189    use ironflow_store::models::{NewRun, NewStep, StepUpdate, TriggerKind, step_trace_id};
190    use ironflow_store::store::RunStore;
191    use serde_json::json;
192
193    use super::*;
194
195    /// A persisted step -- [`Step`] is `#[non_exhaustive]`, so it can only be
196    /// obtained from a store.
197    async fn step() -> Step {
198        let store = InMemoryStore::new();
199        let run = store
200            .create_run(NewRun {
201                created_by: None,
202                workflow_name: "test".to_string(),
203                trigger: TriggerKind::Manual,
204                payload: json!({}),
205                max_retries: 0,
206                handler_version: None,
207                labels: HashMap::new(),
208                scheduled_at: None,
209                idempotency_key: None,
210                max_cost_usd: None,
211            })
212            .await
213            .expect("create run")
214            .into_run();
215
216        store
217            .create_step(NewStep {
218                run_id: run.id,
219                trace_id: step_trace_id(run.id, "build", 0),
220                name: "build".to_string(),
221                kind: StepKind::Shell,
222                position: 0,
223                input: None,
224                is_error_handler: false,
225            })
226            .await
227            .expect("create step")
228    }
229
230    #[tokio::test]
231    async fn a_step_without_artifacts_exposes_an_empty_list() {
232        let response = StepResponse::from(step().await);
233        assert!(response.artifacts.is_empty());
234    }
235
236    #[tokio::test]
237    async fn artifacts_are_carried_through() {
238        let step = step().await;
239        let artifact = ArtifactResponse {
240            id: Uuid::now_v7(),
241            step_id: step.id,
242            name: "report.html".to_string(),
243            content_type: "text/html".to_string(),
244            size_bytes: 1,
245            sha256: "0".repeat(64),
246            created_at: Utc::now(),
247        };
248
249        let response =
250            StepResponse::with_dependencies_and_artifacts(step, Vec::new(), vec![artifact]);
251
252        assert_eq!(response.artifacts.len(), 1);
253        assert_eq!(response.artifacts[0].name, "report.html");
254    }
255
256    #[tokio::test]
257    async fn artifacts_serialize_as_a_json_array() {
258        let body = serde_json::to_value(StepResponse::from(step().await)).expect("serialize");
259        assert!(body["artifacts"].is_array());
260    }
261
262    /// An approval step whose deadline is `offset_secs` from now.
263    async fn gate_with_deadline(offset_secs: i64) -> Step {
264        let store = InMemoryStore::new();
265        let run = store
266            .create_run(NewRun {
267                created_by: None,
268                workflow_name: "test".to_string(),
269                trigger: TriggerKind::Manual,
270                payload: json!({}),
271                max_retries: 0,
272                handler_version: None,
273                labels: HashMap::new(),
274                scheduled_at: None,
275                idempotency_key: None,
276                max_cost_usd: None,
277            })
278            .await
279            .expect("create run")
280            .into_run();
281
282        let step = store
283            .create_step(NewStep {
284                run_id: run.id,
285                trace_id: step_trace_id(run.id, "prod-gate", 0),
286                name: "prod-gate".to_string(),
287                kind: StepKind::Approval,
288                position: 0,
289                input: None,
290                is_error_handler: false,
291            })
292            .await
293            .expect("create step");
294
295        store
296            .update_step(
297                step.id,
298                StepUpdate {
299                    status: Some(StepStatus::Running),
300                    ..StepUpdate::default()
301                },
302            )
303            .await
304            .expect("to running");
305        store
306            .update_step(
307                step.id,
308                StepUpdate {
309                    status: Some(StepStatus::AwaitingApproval),
310                    approval_deadline_at: Some(Utc::now() + TimeDelta::seconds(offset_secs)),
311                    approval_assignee: Some(Assignee::group("release-managers")),
312                    ..StepUpdate::default()
313                },
314            )
315            .await
316            .expect("arm timer");
317
318        store.get_step(step.id).await.expect("get").expect("exists")
319    }
320
321    #[tokio::test]
322    async fn a_step_without_a_deadline_reports_no_sla() {
323        let response = StepResponse::from(step().await);
324        assert!(response.approval_deadline_at.is_none());
325        assert!(response.approval_seconds_remaining.is_none());
326        assert!(response.approval_assignee.is_none());
327        assert!(response.approval_requirement.is_none());
328        assert!(response.approvals.is_empty());
329    }
330
331    #[tokio::test]
332    async fn a_shell_step_requires_no_approvals() {
333        let response = StepResponse::from(step().await);
334        assert_eq!(response.approvals_required, None);
335    }
336
337    #[tokio::test]
338    async fn a_rule_less_approval_step_requires_one_approval() {
339        let response = StepResponse::from(gate_with_deadline(3600).await);
340        assert!(response.approval_requirement.is_none());
341        assert_eq!(response.approvals_required, Some(1));
342    }
343
344    #[tokio::test]
345    async fn an_approval_requirement_sets_the_required_count() {
346        let mut gate = gate_with_deadline(3600).await;
347        let requirement = ApprovalRequirement {
348            reason: Some("amount > 100k".to_string()),
349            required_approvers: 3,
350            approver_groups: vec!["finance".to_string()],
351        };
352        gate.approval_requirement = Some(requirement.clone());
353        gate.approvals = vec![StepApproval {
354            user_id: Uuid::now_v7(),
355            approved_by: "alice".to_string(),
356            at: Utc::now(),
357        }];
358
359        let response = StepResponse::from(gate);
360
361        assert_eq!(response.approvals_required, Some(3));
362        assert_eq!(response.approval_requirement, Some(requirement));
363        assert_eq!(response.approvals.len(), 1);
364        assert_eq!(response.approvals[0].approved_by, "alice");
365    }
366
367    #[tokio::test]
368    async fn a_human_input_step_requires_no_approval_count() {
369        let mut step = gate_with_deadline(3600).await;
370        step.kind = StepKind::HumanInput;
371        step.approval_requirement = Some(ApprovalRequirement {
372            reason: None,
373            required_approvers: 2,
374            approver_groups: vec!["product".to_string()],
375        });
376
377        let response = StepResponse::from(step);
378
379        assert_eq!(response.approvals_required, None);
380        assert!(response.approval_requirement.is_some());
381    }
382
383    #[tokio::test]
384    async fn a_future_deadline_reports_the_remaining_seconds() {
385        let response = StepResponse::from(gate_with_deadline(3600).await);
386
387        assert!(response.approval_deadline_at.is_some());
388        let remaining = response
389            .approval_seconds_remaining
390            .expect("a deadline yields a countdown");
391        assert!(remaining > 0 && remaining <= 3600, "got {remaining}");
392        assert_eq!(
393            response.approval_assignee,
394            Some(Assignee::group("release-managers"))
395        );
396    }
397
398    #[tokio::test]
399    async fn a_past_deadline_clamps_the_countdown_at_zero() {
400        let response = StepResponse::from(gate_with_deadline(-3600).await);
401        assert_eq!(response.approval_seconds_remaining, Some(0));
402    }
403
404    #[tokio::test]
405    async fn trace_id_is_exposed_in_step_response() {
406        let s = step().await;
407        let expected_trace_id = s.trace_id;
408        let response = StepResponse::from(s);
409
410        assert_eq!(response.trace_id, expected_trace_id);
411
412        let body = serde_json::to_value(&response).expect("serialize");
413        assert!(body["trace_id"].is_string());
414    }
415}
ironflow_api/entities/step.rs

ironflow_api/entities/
step.rs
