ironflow_api/entities/
step.rs

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

ironflow_api/entities/
step.rs
