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

ironflow_api/routes/
plan_workflow.rs
