ironflow_api/entities/

step.rs

//! Step-related DTOs.

use chrono::{DateTime, Utc};
use ironflow_store::models::{Assignee, Step, StepKind, StepStatus};
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use uuid::Uuid;

use super::ArtifactResponse;

/// Step response DTO — public API representation of a step.
///
/// # Examples
///
/// ```
/// use ironflow_store::models::Step;
/// use ironflow_api::entities::StepResponse;
/// ```
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[derive(Debug, Serialize, Deserialize)]
pub struct StepResponse {
    /// Unique step identifier.
    pub id: Uuid,
    /// Deterministic trace ID for log correlation.
    pub trace_id: Uuid,
    /// Parent run ID.
    pub run_id: Uuid,
    /// Step name.
    pub name: String,
    /// Step operation type.
    #[cfg_attr(feature = "openapi", schema(value_type = String))]
    pub kind: StepKind,
    /// Execution order (0-based).
    pub position: u32,
    /// Current status.
    pub status: StepStatus,
    /// Which run attempt produced this step (1-based).
    ///
    /// A run retried twice exposes steps with `attempt` 1, 2 and 3. Steps from
    /// earlier attempts are kept so a failed attempt stays inspectable.
    pub attempt: u32,
    /// Input configuration.
    #[cfg_attr(feature = "openapi", schema(value_type = Option<std::collections::HashMap<String, serde_json::Value>>))]
    pub input: Option<Value>,
    /// Step output.
    #[cfg_attr(feature = "openapi", schema(value_type = Option<std::collections::HashMap<String, serde_json::Value>>))]
    pub output: Option<Value>,
    /// Optional error message.
    pub error: Option<String>,
    /// Execution duration in milliseconds.
    pub duration_ms: u64,
    /// Cost in USD.
    #[cfg_attr(feature = "openapi", schema(value_type = f64))]
    pub cost_usd: Decimal,
    /// Input token count (agent steps).
    pub input_tokens: Option<u64>,
    /// Output token count (agent steps).
    pub output_tokens: Option<u64>,
    /// When created.
    pub created_at: DateTime<Utc>,
    /// When updated.
    pub updated_at: DateTime<Utc>,
    /// When execution started.
    pub started_at: Option<DateTime<Utc>>,
    /// When execution completed.
    pub completed_at: Option<DateTime<Utc>>,
    /// IDs of steps this step depends on (direct dependencies).
    pub dependencies: Vec<Uuid>,
    /// Verbose conversation trace for agent steps (thinking blocks, tool
    /// calls, tool results, per-turn usage). `None` when verbose mode was
    /// off or the step is not an agent step.
    #[cfg_attr(feature = "openapi", schema(value_type = Option<serde_json::Value>))]
    pub debug_messages: Option<Value>,
    /// Files this step produced, downloadable through the artifact route.
    ///
    /// Empty when the step produced none or when artifact storage is not
    /// configured on the server.
    #[serde(default)]
    pub artifacts: Vec<ArtifactResponse>,
    /// When this approval gate expires, if it carries an SLA deadline.
    pub approval_deadline_at: Option<DateTime<Utc>>,
    /// Seconds left before the gate escalates. Clamped at 0, `None` when the
    /// step has no deadline.
    pub approval_seconds_remaining: Option<i64>,
    /// Who the approval is currently assigned to.
    ///
    /// Serialized as a prefixed string: `user:{name}` or `group:{name}`.
    #[cfg_attr(feature = "openapi", schema(value_type = Option<String>))]
    pub approval_assignee: Option<Assignee>,
}

impl StepResponse {
    /// Build a response from a step entity with pre-resolved dependencies.
    ///
    /// Artifacts are left empty; use
    /// [`with_dependencies_and_artifacts`](Self::with_dependencies_and_artifacts)
    /// when they have been fetched.
    pub fn with_dependencies(step: Step, dependencies: Vec<Uuid>) -> Self {
        Self::with_dependencies_and_artifacts(step, dependencies, Vec::new())
    }

    /// Build a response from a step entity with its dependencies and artifacts.
    pub fn with_dependencies_and_artifacts(
        step: Step,
        dependencies: Vec<Uuid>,
        artifacts: Vec<ArtifactResponse>,
    ) -> Self {
        let approval_seconds_remaining = step
            .approval_deadline_at
            .map(|at| (at - Utc::now()).num_seconds().max(0));

        StepResponse {
            id: step.id,
            trace_id: step.trace_id,
            run_id: step.run_id,
            name: step.name,
            kind: step.kind,
            position: step.position,
            status: step.status.state,
            attempt: step.attempt,
            input: step.input,
            output: step.output,
            error: step.error,
            duration_ms: step.duration_ms,
            cost_usd: step.cost_usd,
            input_tokens: step.input_tokens,
            output_tokens: step.output_tokens,
            created_at: step.created_at,
            updated_at: step.updated_at,
            started_at: step.started_at,
            completed_at: step.completed_at,
            dependencies,
            debug_messages: step.debug_messages,
            artifacts,
            approval_deadline_at: step.approval_deadline_at,
            approval_seconds_remaining,
            approval_assignee: step.approval_assignee,
        }
    }
}

impl From<Step> for StepResponse {
    fn from(step: Step) -> Self {
        Self::with_dependencies(step, Vec::new())
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;

    use chrono::TimeDelta;
    use ironflow_store::memory::InMemoryStore;
    use ironflow_store::models::{NewRun, NewStep, StepUpdate, TriggerKind, step_trace_id};
    use ironflow_store::store::RunStore;
    use serde_json::json;

    use super::*;

    /// A persisted step -- [`Step`] is `#[non_exhaustive]`, so it can only be
    /// obtained from a store.
    async fn step() -> Step {
        let store = InMemoryStore::new();
        let run = store
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                max_cost_usd: None,
            })
            .await
            .expect("create run")
            .into_run();

        store
            .create_step(NewStep {
                run_id: run.id,
                trace_id: step_trace_id(run.id, "build", 0),
                name: "build".to_string(),
                kind: StepKind::Shell,
                position: 0,
                input: None,
                is_error_handler: false,
            })
            .await
            .expect("create step")
    }

    #[tokio::test]
    async fn a_step_without_artifacts_exposes_an_empty_list() {
        let response = StepResponse::from(step().await);
        assert!(response.artifacts.is_empty());
    }

    #[tokio::test]
    async fn artifacts_are_carried_through() {
        let step = step().await;
        let artifact = ArtifactResponse {
            id: Uuid::now_v7(),
            step_id: step.id,
            name: "report.html".to_string(),
            content_type: "text/html".to_string(),
            size_bytes: 1,
            sha256: "0".repeat(64),
            created_at: Utc::now(),
        };

        let response =
            StepResponse::with_dependencies_and_artifacts(step, Vec::new(), vec![artifact]);

        assert_eq!(response.artifacts.len(), 1);
        assert_eq!(response.artifacts[0].name, "report.html");
    }

    #[tokio::test]
    async fn artifacts_serialize_as_a_json_array() {
        let body = serde_json::to_value(StepResponse::from(step().await)).expect("serialize");
        assert!(body["artifacts"].is_array());
    }

    /// An approval step whose deadline is `offset_secs` from now.
    async fn gate_with_deadline(offset_secs: i64) -> Step {
        let store = InMemoryStore::new();
        let run = store
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                max_cost_usd: None,
            })
            .await
            .expect("create run")
            .into_run();

        let step = store
            .create_step(NewStep {
                run_id: run.id,
                trace_id: step_trace_id(run.id, "prod-gate", 0),
                name: "prod-gate".to_string(),
                kind: StepKind::Approval,
                position: 0,
                input: None,
                is_error_handler: false,
            })
            .await
            .expect("create step");

        store
            .update_step(
                step.id,
                StepUpdate {
                    status: Some(StepStatus::Running),
                    ..StepUpdate::default()
                },
            )
            .await
            .expect("to running");
        store
            .update_step(
                step.id,
                StepUpdate {
                    status: Some(StepStatus::AwaitingApproval),
                    approval_deadline_at: Some(Utc::now() + TimeDelta::seconds(offset_secs)),
                    approval_assignee: Some(Assignee::group("release-managers")),
                    ..StepUpdate::default()
                },
            )
            .await
            .expect("arm timer");

        store.get_step(step.id).await.expect("get").expect("exists")
    }

    #[tokio::test]
    async fn a_step_without_a_deadline_reports_no_sla() {
        let response = StepResponse::from(step().await);
        assert!(response.approval_deadline_at.is_none());
        assert!(response.approval_seconds_remaining.is_none());
        assert!(response.approval_assignee.is_none());
    }

    #[tokio::test]
    async fn a_future_deadline_reports_the_remaining_seconds() {
        let response = StepResponse::from(gate_with_deadline(3600).await);

        assert!(response.approval_deadline_at.is_some());
        let remaining = response
            .approval_seconds_remaining
            .expect("a deadline yields a countdown");
        assert!(remaining > 0 && remaining <= 3600, "got {remaining}");
        assert_eq!(
            response.approval_assignee,
            Some(Assignee::group("release-managers"))
        );
    }

    #[tokio::test]
    async fn a_past_deadline_clamps_the_countdown_at_zero() {
        let response = StepResponse::from(gate_with_deadline(-3600).await);
        assert_eq!(response.approval_seconds_remaining, Some(0));
    }

    #[tokio::test]
    async fn trace_id_is_exposed_in_step_response() {
        let s = step().await;
        let expected_trace_id = s.trace_id;
        let response = StepResponse::from(s);

        assert_eq!(response.trace_id, expected_trace_id);

        let body = serde_json::to_value(&response).expect("serialize");
        assert!(body["trace_id"].is_string());
    }
}