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                concurrency_key: None,
211                max_cost_usd: None,
212            })
213            .await
214            .expect("create run")
215            .into_run();
216
217        store
218            .create_step(NewStep {
219                run_id: run.id,
220                trace_id: step_trace_id(run.id, "build", 0),
221                name: "build".to_string(),
222                kind: StepKind::Shell,
223                position: 0,
224                input: None,
225                is_error_handler: false,
226            })
227            .await
228            .expect("create step")
229    }
230
231    #[tokio::test]
232    async fn a_step_without_artifacts_exposes_an_empty_list() {
233        let response = StepResponse::from(step().await);
234        assert!(response.artifacts.is_empty());
235    }
236
237    #[tokio::test]
238    async fn artifacts_are_carried_through() {
239        let step = step().await;
240        let artifact = ArtifactResponse {
241            id: Uuid::now_v7(),
242            step_id: step.id,
243            name: "report.html".to_string(),
244            content_type: "text/html".to_string(),
245            size_bytes: 1,
246            sha256: "0".repeat(64),
247            created_at: Utc::now(),
248        };
249
250        let response =
251            StepResponse::with_dependencies_and_artifacts(step, Vec::new(), vec![artifact]);
252
253        assert_eq!(response.artifacts.len(), 1);
254        assert_eq!(response.artifacts[0].name, "report.html");
255    }
256
257    #[tokio::test]
258    async fn artifacts_serialize_as_a_json_array() {
259        let body = serde_json::to_value(StepResponse::from(step().await)).expect("serialize");
260        assert!(body["artifacts"].is_array());
261    }
262
263    /// An approval step whose deadline is `offset_secs` from now.
264    async fn gate_with_deadline(offset_secs: i64) -> Step {
265        let store = InMemoryStore::new();
266        let run = store
267            .create_run(NewRun {
268                created_by: None,
269                workflow_name: "test".to_string(),
270                trigger: TriggerKind::Manual,
271                payload: json!({}),
272                max_retries: 0,
273                handler_version: None,
274                labels: HashMap::new(),
275                scheduled_at: None,
276                idempotency_key: None,
277                concurrency_key: None,
278                max_cost_usd: None,
279            })
280            .await
281            .expect("create run")
282            .into_run();
283
284        let step = store
285            .create_step(NewStep {
286                run_id: run.id,
287                trace_id: step_trace_id(run.id, "prod-gate", 0),
288                name: "prod-gate".to_string(),
289                kind: StepKind::Approval,
290                position: 0,
291                input: None,
292                is_error_handler: false,
293            })
294            .await
295            .expect("create step");
296
297        store
298            .update_step(
299                step.id,
300                StepUpdate {
301                    status: Some(StepStatus::Running),
302                    ..StepUpdate::default()
303                },
304            )
305            .await
306            .expect("to running");
307        store
308            .update_step(
309                step.id,
310                StepUpdate {
311                    status: Some(StepStatus::AwaitingApproval),
312                    approval_deadline_at: Some(Utc::now() + TimeDelta::seconds(offset_secs)),
313                    approval_assignee: Some(Assignee::group("release-managers")),
314                    ..StepUpdate::default()
315                },
316            )
317            .await
318            .expect("arm timer");
319
320        store.get_step(step.id).await.expect("get").expect("exists")
321    }
322
323    #[tokio::test]
324    async fn a_step_without_a_deadline_reports_no_sla() {
325        let response = StepResponse::from(step().await);
326        assert!(response.approval_deadline_at.is_none());
327        assert!(response.approval_seconds_remaining.is_none());
328        assert!(response.approval_assignee.is_none());
329        assert!(response.approval_requirement.is_none());
330        assert!(response.approvals.is_empty());
331    }
332
333    #[tokio::test]
334    async fn a_shell_step_requires_no_approvals() {
335        let response = StepResponse::from(step().await);
336        assert_eq!(response.approvals_required, None);
337    }
338
339    #[tokio::test]
340    async fn a_rule_less_approval_step_requires_one_approval() {
341        let response = StepResponse::from(gate_with_deadline(3600).await);
342        assert!(response.approval_requirement.is_none());
343        assert_eq!(response.approvals_required, Some(1));
344    }
345
346    #[tokio::test]
347    async fn an_approval_requirement_sets_the_required_count() {
348        let mut gate = gate_with_deadline(3600).await;
349        let requirement = ApprovalRequirement {
350            reason: Some("amount > 100k".to_string()),
351            required_approvers: 3,
352            approver_groups: vec!["finance".to_string()],
353        };
354        gate.approval_requirement = Some(requirement.clone());
355        gate.approvals = vec![StepApproval {
356            user_id: Uuid::now_v7(),
357            approved_by: "alice".to_string(),
358            at: Utc::now(),
359        }];
360
361        let response = StepResponse::from(gate);
362
363        assert_eq!(response.approvals_required, Some(3));
364        assert_eq!(response.approval_requirement, Some(requirement));
365        assert_eq!(response.approvals.len(), 1);
366        assert_eq!(response.approvals[0].approved_by, "alice");
367    }
368
369    #[tokio::test]
370    async fn a_human_input_step_requires_no_approval_count() {
371        let mut step = gate_with_deadline(3600).await;
372        step.kind = StepKind::HumanInput;
373        step.approval_requirement = Some(ApprovalRequirement {
374            reason: None,
375            required_approvers: 2,
376            approver_groups: vec!["product".to_string()],
377        });
378
379        let response = StepResponse::from(step);
380
381        assert_eq!(response.approvals_required, None);
382        assert!(response.approval_requirement.is_some());
383    }
384
385    #[tokio::test]
386    async fn a_future_deadline_reports_the_remaining_seconds() {
387        let response = StepResponse::from(gate_with_deadline(3600).await);
388
389        assert!(response.approval_deadline_at.is_some());
390        let remaining = response
391            .approval_seconds_remaining
392            .expect("a deadline yields a countdown");
393        assert!(remaining > 0 && remaining <= 3600, "got {remaining}");
394        assert_eq!(
395            response.approval_assignee,
396            Some(Assignee::group("release-managers"))
397        );
398    }
399
400    #[tokio::test]
401    async fn a_past_deadline_clamps_the_countdown_at_zero() {
402        let response = StepResponse::from(gate_with_deadline(-3600).await);
403        assert_eq!(response.approval_seconds_remaining, Some(0));
404    }
405
406    #[tokio::test]
407    async fn trace_id_is_exposed_in_step_response() {
408        let s = step().await;
409        let expected_trace_id = s.trace_id;
410        let response = StepResponse::from(s);
411
412        assert_eq!(response.trace_id, expected_trace_id);
413
414        let body = serde_json::to_value(&response).expect("serialize");
415        assert!(body["trace_id"].is_string());
416    }
417}
ironflow_api/entities/step.rs

ironflow_api/entities/
step.rs
