ironflow_api/routes/
plan_workflow.rs

1//! `POST /api/v1/workflows/:name/plan` — Build a workflow execution plan.
2
3use axum::Json;
4use axum::extract::{Path, State};
5use axum::response::IntoResponse;
6use ironflow_auth::extractor::Authenticated;
7use ironflow_engine::error::EngineError;
8use ironflow_engine::plan::{
9    ConditionResult, DEFAULT_ESTIMATE_SAMPLE_RUNS, DEFAULT_PLAN_MAX_DEPTH, DEFAULT_PLAN_TIMEOUT,
10    ExecutionPlan, PlanOptions, PlannedStep,
11};
12use ironflow_store::entities::ApiKeyScope;
13use ironflow_store::entities::StepKind;
14use serde::{Deserialize, Serialize};
15use serde_json::{Value, json};
16
17use crate::authorization::AuthorizeExt;
18use crate::error::ApiError;
19use crate::response::ok;
20use crate::state::AppState;
21
22/// Highest sub-workflow expansion depth the API accepts.
23const MAX_PLAN_DEPTH: u32 = 10;
24
25/// Request body for building a workflow execution plan.
26#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
27#[derive(Debug, Default, Deserialize, Serialize)]
28pub struct PlanWorkflowRequest {
29    /// Input payload the plan is computed for. Defaults to `{}`.
30    #[cfg_attr(feature = "openapi", schema(value_type = Option<std::collections::HashMap<String, serde_json::Value>>))]
31    #[serde(default)]
32    pub payload: Option<Value>,
33    /// How deep sub-workflows are expanded. Defaults to 3, capped at 10.
34    #[serde(default)]
35    pub max_depth: Option<u32>,
36    /// Estimate step durations from run history. Defaults to `true`.
37    #[serde(default)]
38    pub estimate_durations: Option<bool>,
39}
40
41/// Outcome of a branch condition as recorded by the planner.
42#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
43#[derive(Debug, Serialize)]
44pub struct ConditionResponse {
45    /// Condition state: `evaluated`, `skipped` or `unevaluable`.
46    pub state: String,
47    /// Expression the handler declared, when the planner knows one.
48    #[serde(skip_serializing_if = "Option::is_none")]
49    pub expression: Option<String>,
50    /// What the expression evaluated to, for an `evaluated` condition.
51    #[serde(skip_serializing_if = "Option::is_none")]
52    pub value: Option<bool>,
53    /// Why the step is skipped, or why the condition cannot be evaluated.
54    #[serde(skip_serializing_if = "Option::is_none")]
55    pub reason: Option<String>,
56}
57
58/// One step the planner expects the run to create.
59#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
60#[derive(Debug, Serialize)]
61pub struct PlannedStepResponse {
62    /// Step name as the handler declares it.
63    pub name: String,
64    /// Step kind, or the name of a custom operation.
65    pub kind: String,
66    /// Workflow that owns this step.
67    pub workflow: String,
68    /// Sub-workflow nesting depth; `0` for the top-level workflow.
69    pub depth: u32,
70    /// Names of the steps this one runs after.
71    pub depends_on: Vec<String>,
72    /// Branch condition recorded for this step, when the handler declared one.
73    #[serde(skip_serializing_if = "Option::is_none")]
74    pub condition: Option<ConditionResponse>,
75    /// Parallel wave this step belongs to, when it runs concurrently.
76    #[serde(skip_serializing_if = "Option::is_none")]
77    pub parallel_group: Option<String>,
78    /// Average duration of this step in past completed runs, in milliseconds.
79    #[serde(skip_serializing_if = "Option::is_none")]
80    pub estimated_duration_ms: Option<u64>,
81}
82
83/// The execution plan of one workflow for one input payload.
84#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
85#[derive(Debug, Serialize)]
86pub struct ExecutionPlanResponse {
87    /// Workflow the plan was built for.
88    pub workflow: String,
89    /// Steps the run is expected to create, in execution order.
90    pub steps: Vec<PlannedStepResponse>,
91    /// Sum of the step estimates, counting each parallel wave once.
92    #[serde(skip_serializing_if = "Option::is_none")]
93    pub estimated_duration_ms: Option<u64>,
94    /// Sub-workflow expansion depth used for this plan.
95    pub max_depth: u32,
96    /// `true` when the step cap or the depth limit cut the plan short.
97    pub truncated: bool,
98    /// Why the plan stopped early, when it did.
99    #[serde(skip_serializing_if = "Option::is_none")]
100    pub incomplete_reason: Option<String>,
101}
102
103/// Render a [`StepKind`] the way the rest of the API does: a plain string.
104fn kind_label(kind: &StepKind) -> String {
105    match kind {
106        StepKind::Shell => "shell".to_string(),
107        StepKind::Http => "http".to_string(),
108        StepKind::Agent => "agent".to_string(),
109        StepKind::Workflow => "workflow".to_string(),
110        StepKind::Approval => "approval".to_string(),
111        StepKind::Decision => "decision".to_string(),
112        StepKind::HumanInput => "human_input".to_string(),
113        StepKind::Signal => "signal".to_string(),
114        StepKind::Custom(name) => name.clone(),
115    }
116}
117
118impl From<ConditionResult> for ConditionResponse {
119    fn from(condition: ConditionResult) -> Self {
120        match condition {
121            ConditionResult::Evaluated { expression, value } => Self {
122                state: "evaluated".to_string(),
123                expression: Some(expression),
124                value: Some(value),
125                reason: None,
126            },
127            ConditionResult::Skipped { reason } => Self {
128                state: "skipped".to_string(),
129                expression: None,
130                value: None,
131                reason: Some(reason),
132            },
133            ConditionResult::Unevaluable { expression, reason } => Self {
134                state: "unevaluable".to_string(),
135                expression: Some(expression),
136                value: None,
137                reason: Some(reason),
138            },
139        }
140    }
141}
142
143impl From<PlannedStep> for PlannedStepResponse {
144    fn from(step: PlannedStep) -> Self {
145        let estimated_duration_ms = step.estimated_duration_ms();
146        Self {
147            name: step.name,
148            kind: kind_label(&step.kind),
149            workflow: step.workflow,
150            depth: step.depth,
151            depends_on: step.depends_on,
152            condition: step.condition.map(ConditionResponse::from),
153            parallel_group: step.parallel_group,
154            estimated_duration_ms,
155        }
156    }
157}
158
159impl From<ExecutionPlan> for ExecutionPlanResponse {
160    fn from(plan: ExecutionPlan) -> Self {
161        let estimated_duration_ms = plan.estimated_duration_ms();
162        Self {
163            workflow: plan.workflow,
164            steps: plan
165                .steps
166                .into_iter()
167                .map(PlannedStepResponse::from)
168                .collect(),
169            estimated_duration_ms,
170            max_depth: plan.max_depth,
171            truncated: plan.truncated,
172            incomplete_reason: plan.incomplete_reason,
173        }
174    }
175}
176
177/// Build a workflow execution plan without running it.
178///
179/// # Errors
180///
181/// - 400 if `max_depth` is out of range or the payload is not a JSON object
182/// - 404 if the workflow is not registered
183#[cfg_attr(
184    feature = "openapi",
185    utoipa::path(
186        post,
187        path = "/api/v1/workflows/{name}/plan",
188        tags = ["workflows"],
189        params(("name" = String, Path, description = "Workflow name")),
190        request_body(content = PlanWorkflowRequest, description = "Input payload and planning options"),
191        responses(
192            (status = 200, description = "Execution plan", body = ExecutionPlanResponse),
193            (status = 400, description = "Invalid payload or max_depth"),
194            (status = 401, description = "Unauthorized"),
195            (status = 404, description = "Workflow not found")
196        ),
197        security(("Bearer" = []))
198    )
199)]
200pub async fn plan_workflow(
201    auth: Authenticated,
202    State(state): State<AppState>,
203    Path(name): Path<String>,
204    Json(req): Json<PlanWorkflowRequest>,
205) -> Result<impl IntoResponse, ApiError> {
206    auth.require(ApiKeyScope::WorkflowsRead, false)?;
207    if !state.engine.handler_names().contains(&name.as_str()) {
208        return Err(ApiError::WorkflowNotFound(name));
209    }
210
211    let max_depth = req.max_depth.unwrap_or(DEFAULT_PLAN_MAX_DEPTH);
212    if max_depth == 0 || max_depth > MAX_PLAN_DEPTH {
213        return Err(ApiError::BadRequest(format!(
214            "max_depth must be between 1 and {MAX_PLAN_DEPTH}"
215        )));
216    }
217
218    let payload = req.payload.unwrap_or_else(|| json!({}));
219    if !payload.is_object() {
220        return Err(ApiError::BadRequest(
221            "payload must be a JSON object".to_string(),
222        ));
223    }
224
225    let options = PlanOptions {
226        max_depth,
227        estimate_durations: req.estimate_durations.unwrap_or(true),
228        sample_runs: DEFAULT_ESTIMATE_SAMPLE_RUNS,
229        timeout: DEFAULT_PLAN_TIMEOUT,
230    };
231
232    let plan = state
233        .engine
234        .plan_handler(&name, payload, options)
235        .await
236        .map_err(|e| match e {
237            EngineError::InvalidWorkflow(msg) => ApiError::BadRequest(msg),
238            EngineError::Store(err) => ApiError::from(err),
239            other => ApiError::Internal(other.to_string()),
240        })?;
241
242    Ok(ok(ExecutionPlanResponse::from(plan)))
243}
244
245#[cfg(test)]
246mod tests {
247    use std::sync::Arc;
248
249    use axum::Router;
250    use axum::body::Body;
251    use axum::http::{Request, StatusCode};
252    use axum::response::Response;
253    use axum::routing::post;
254    use http_body_util::BodyExt;
255    use ironflow_auth::jwt::JwtConfig;
256    use ironflow_core::providers::claude::ClaudeCodeProvider;
257    use ironflow_engine::config::{ShellConfig, StepConfig};
258    use ironflow_engine::context::WorkflowContext;
259    use ironflow_engine::engine::Engine;
260    use ironflow_engine::handler::{HandlerFuture, WorkflowHandler};
261    use ironflow_engine::notify::Event;
262    use ironflow_store::memory::InMemoryStore;
263    use ironflow_store::models::RunFilter;
264    use serde_json::{Value as JsonValue, from_slice};
265    use tokio::sync::broadcast;
266    use tower::ServiceExt;
267
268    use super::*;
269    use crate::routes::test_helpers::create_user_auth_header;
270
271    struct PlannedWorkflow;
272
273    impl WorkflowHandler for PlannedWorkflow {
274        fn name(&self) -> &str {
275            "planned"
276        }
277
278        fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
279            Box::pin(async move {
280                ctx.shell("build", ShellConfig::new("echo build")).await?;
281                ctx.parallel(
282                    vec![
283                        ("test", StepConfig::Shell(ShellConfig::new("echo test"))),
284                        ("lint", StepConfig::Shell(ShellConfig::new("echo lint"))),
285                    ],
286                    true,
287                )
288                .await?;
289                Ok(())
290            })
291        }
292    }
293
294    #[derive(Deserialize)]
295    struct DeployInput {
296        env: String,
297    }
298
299    struct ConditionalWorkflow;
300
301    impl WorkflowHandler for ConditionalWorkflow {
302        fn name(&self) -> &str {
303            "conditional"
304        }
305
306        fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
307            Box::pin(async move {
308                if ctx
309                    .when("production run", |i: &DeployInput| i.env == "prod")
310                    .await?
311                {
312                    ctx.shell("deploy", ShellConfig::new("echo deploy")).await?;
313                } else {
314                    ctx.skip("deploy", "not prod").await?;
315                }
316                Ok(())
317            })
318        }
319    }
320
321    fn test_state() -> AppState {
322        let store = Arc::new(InMemoryStore::new());
323        let provider = Arc::new(ClaudeCodeProvider::new());
324        let mut engine = Engine::new(store.clone(), provider);
325        engine.register(PlannedWorkflow).unwrap();
326        engine.register(ConditionalWorkflow).unwrap();
327        let jwt_config = Arc::new(JwtConfig {
328            secret: "test-secret".to_string(),
329            access_token_ttl_secs: 900,
330            refresh_token_ttl_secs: 604800,
331            cookie_domain: None,
332            cookie_secure: false,
333        });
334        let (event_sender, _) = broadcast::channel::<Event>(1);
335        AppState::new(
336            store,
337            Arc::new(engine),
338            jwt_config,
339            "test-worker-token".to_string(),
340            event_sender,
341        )
342    }
343
344    fn app(state: AppState) -> Router {
345        Router::new()
346            .route("/api/v1/workflows/{name}/plan", post(plan_workflow))
347            .with_state(state)
348    }
349
350    fn plan_request(name: &str, auth: Option<&str>, body: JsonValue) -> Request<Body> {
351        let mut builder = Request::builder()
352            .method("POST")
353            .uri(format!("/api/v1/workflows/{name}/plan"))
354            .header("content-type", "application/json");
355        if let Some(header) = auth {
356            builder = builder.header("authorization", header);
357        }
358        builder.body(Body::from(body.to_string())).unwrap()
359    }
360
361    async fn body_json(response: Response) -> JsonValue {
362        let bytes = response.into_body().collect().await.unwrap().to_bytes();
363        from_slice(&bytes).unwrap()
364    }
365
366    #[tokio::test]
367    async fn plan_returns_steps_for_registered_workflow() {
368        let state = test_state();
369        let auth = create_user_auth_header(&state, "testuser", false).await;
370        let response = app(state)
371            .oneshot(plan_request("planned", Some(&auth), json!({})))
372            .await
373            .unwrap();
374
375        assert_eq!(response.status(), StatusCode::OK);
376        let body = body_json(response).await;
377        assert_eq!(body["data"]["workflow"], "planned");
378        let steps = body["data"]["steps"].as_array().unwrap();
379        assert_eq!(steps.len(), 3);
380        assert_eq!(steps[0]["name"], "build");
381        assert_eq!(steps[0]["kind"], "shell");
382        assert_eq!(steps[1]["name"], "test");
383        assert_eq!(steps[2]["name"], "lint");
384        assert_eq!(body["data"]["truncated"], false);
385    }
386
387    #[tokio::test]
388    async fn plan_marks_parallel_group() {
389        let state = test_state();
390        let auth = create_user_auth_header(&state, "testuser", false).await;
391        let response = app(state)
392            .oneshot(plan_request("planned", Some(&auth), json!({})))
393            .await
394            .unwrap();
395
396        let body = body_json(response).await;
397        let steps = body["data"]["steps"].as_array().unwrap();
398        assert!(steps[0].get("parallel_group").is_none());
399        assert_eq!(steps[1]["parallel_group"], "parallel-1");
400        assert_eq!(steps[2]["parallel_group"], "parallel-1");
401    }
402
403    #[tokio::test]
404    async fn plan_evaluates_condition_from_payload() {
405        let state = test_state();
406        let auth = create_user_auth_header(&state, "testuser", false).await;
407        let router = app(state);
408
409        let response = router
410            .clone()
411            .oneshot(plan_request(
412                "conditional",
413                Some(&auth),
414                json!({"payload": {"env": "prod"}}),
415            ))
416            .await
417            .unwrap();
418        let body = body_json(response).await;
419        let step = &body["data"]["steps"][0];
420        assert_eq!(step["kind"], "shell");
421        assert_eq!(step["condition"]["state"], "evaluated");
422        assert_eq!(step["condition"]["expression"], "production run");
423        assert_eq!(step["condition"]["value"], true);
424
425        let response = router
426            .oneshot(plan_request(
427                "conditional",
428                Some(&auth),
429                json!({"payload": {"env": "dev"}}),
430            ))
431            .await
432            .unwrap();
433        let body = body_json(response).await;
434        let step = &body["data"]["steps"][0];
435        assert_eq!(step["kind"], "skip");
436        assert_eq!(step["condition"]["state"], "skipped");
437        assert_eq!(step["condition"]["reason"], "not prod");
438    }
439
440    #[tokio::test]
441    async fn plan_unknown_workflow_returns_404() {
442        let state = test_state();
443        let auth = create_user_auth_header(&state, "testuser", false).await;
444        let response = app(state)
445            .oneshot(plan_request("nonexistent", Some(&auth), json!({})))
446            .await
447            .unwrap();
448
449        assert_eq!(response.status(), StatusCode::NOT_FOUND);
450    }
451
452    #[tokio::test]
453    async fn plan_rejects_zero_and_excessive_max_depth() {
454        let state = test_state();
455        let auth = create_user_auth_header(&state, "testuser", false).await;
456        let router = app(state);
457
458        let response = router
459            .clone()
460            .oneshot(plan_request(
461                "planned",
462                Some(&auth),
463                json!({"max_depth": 0}),
464            ))
465            .await
466            .unwrap();
467        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
468
469        let response = router
470            .oneshot(plan_request(
471                "planned",
472                Some(&auth),
473                json!({"max_depth": 11}),
474            ))
475            .await
476            .unwrap();
477        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
478    }
479
480    #[tokio::test]
481    async fn plan_rejects_non_object_payload() {
482        let state = test_state();
483        let auth = create_user_auth_header(&state, "testuser", false).await;
484        let response = app(state)
485            .oneshot(plan_request(
486                "planned",
487                Some(&auth),
488                json!({"payload": [1, 2, 3]}),
489            ))
490            .await
491            .unwrap();
492
493        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
494    }
495
496    #[tokio::test]
497    async fn plan_requires_authentication() {
498        let state = test_state();
499        let response = app(state)
500            .oneshot(plan_request("planned", None, json!({})))
501            .await
502            .unwrap();
503
504        assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
505    }
506
507    #[tokio::test]
508    async fn plan_does_not_create_a_run() {
509        let state = test_state();
510        let store = state.store.clone();
511        let auth = create_user_auth_header(&state, "testuser", false).await;
512        let response = app(state)
513            .oneshot(plan_request("planned", Some(&auth), json!({})))
514            .await
515            .unwrap();
516        assert_eq!(response.status(), StatusCode::OK);
517
518        let runs = store.list_runs(RunFilter::default(), 1, 10).await.unwrap();
519        assert!(runs.items.is_empty());
520    }
521}
ironflow_api/routes/plan_workflow.rs

ironflow_api/routes/
plan_workflow.rs
