ironflow_api/routes/
human_input.rs

1//! `POST /api/v1/runs/:id/steps/:step_id/input` -- Answer a human input step.
2//!
3//! `POST /api/v1/runs/:id/steps/:step_id/reject` -- Reject a human input step.
4
5use axum::Json;
6use axum::extract::{Path, State};
7use axum::response::IntoResponse;
8use chrono::Utc;
9use ironflow_auth::extractor::Authenticated;
10use ironflow_engine::config::HUMAN_INPUT_SCHEMA_KEY;
11use ironflow_engine::engine::ExecutionMode;
12use ironflow_store::entities::ApiKeyScope;
13use ironflow_store::models::{
14    Run, RunStatus, Step, StepApproval, StepKind, StepStatus, StepUpdate,
15};
16use jsonschema::validator_for;
17use serde::{Deserialize, Serialize};
18use serde_json::Value;
19use tokio::spawn;
20use tracing::error;
21use uuid::Uuid;
22
23use crate::authorization::AuthorizeExt;
24use crate::entities::RunResponse;
25use crate::error::ApiError;
26use crate::response::ok;
27use crate::routes::approve_run::{authorize_gate, awaits_gate, resume_paused_run_later};
28use crate::state::AppState;
29
30/// Request body for rejecting a human input step.
31#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
32#[derive(Debug, Default, Deserialize, Serialize)]
33pub struct RejectHumanInputRequest {
34    /// Why the input is refused. Passed to the handler.
35    #[serde(default)]
36    pub reason: Option<String>,
37}
38
39/// Answer a human input step with a value matching its JSON schema.
40///
41/// The answer is validated against the schema stored on the step. The caller
42/// must be allowed to resolve the step like an approval gate (admin, approver
43/// groups, assignee or delegation). The first valid answer wins: the step
44/// completes, the run moves from `AwaitingApproval` back to `Running` and
45/// resumes, and the handler receives the typed answer.
46///
47/// # Errors
48///
49/// - 400 if the step is not a human input or is not awaiting input
50/// - 403 if the caller may not answer the step
51/// - 404 if the run or the step does not exist
52/// - 409 if the input was already answered or rejected
53/// - 422 if the answer does not match the schema
54#[cfg_attr(
55    feature = "openapi",
56    utoipa::path(
57        post,
58        path = "/api/v1/runs/{id}/steps/{step_id}/input",
59        tags = ["runs"],
60        params(
61            ("id" = Uuid, Path, description = "Run ID"),
62            ("step_id" = Uuid, Path, description = "Step ID")
63        ),
64        request_body(content = Value, content_type = "application/json", description = "Answer matching the JSON schema stored on the step"),
65        responses(
66            (status = 200, description = "Answer recorded, the run resumes", body = RunResponse),
67            (status = 400, description = "Step is not a human input awaiting input"),
68            (status = 401, description = "Unauthorized"),
69            (status = 403, description = "Forbidden"),
70            (status = 404, description = "Run or step not found"),
71            (status = 409, description = "Input already answered or rejected"),
72            (status = 422, description = "Answer does not match the schema")
73        ),
74        security(("Bearer" = []))
75    )
76)]
77pub async fn submit_human_input(
78    auth: Authenticated,
79    State(state): State<AppState>,
80    Path((id, step_id)): Path<(Uuid, Uuid)>,
81    Json(answer): Json<Value>,
82) -> Result<impl IntoResponse, ApiError> {
83    auth.require(ApiKeyScope::RunsManage, false)?;
84    let (run, step) = open_input_step(&state, id, step_id).await?;
85    let actor = authorize_gate(&auth, &state, &run, &step).await?;
86    validate_answer(step.input.as_ref(), &answer)?;
87
88    // Like approve, two concurrent answers are not serialized: both may pass
89    // the status check above before either is written. This is an accepted
90    // limitation of the gate routes.
91    let now = Utc::now();
92    state
93        .store
94        .record_step_approval(
95            step.id,
96            StepApproval {
97                user_id: auth.user_id,
98                approved_by: actor,
99                at: now,
100            },
101        )
102        .await?;
103    state
104        .store
105        .update_step(
106            step.id,
107            StepUpdate {
108                status: Some(StepStatus::Completed),
109                output: Some(answer),
110                completed_at: Some(now),
111                clear_approval_deadline: true,
112                ..StepUpdate::default()
113            },
114        )
115        .await?;
116    release_run(&state, &run).await?;
117
118    Ok(ok(RunResponse::from(state.get_run_or_404(id).await?)))
119}
120
121/// Reject a human input step.
122///
123/// The step is marked `Rejected` with the reason, the run moves from
124/// `AwaitingApproval` back to `Running` and resumes: the handler receives a
125/// `HumanInputRejected` error and decides what happens next. The body is
126/// optional; without a reason, one naming the caller is recorded.
127///
128/// # Errors
129///
130/// - 400 if the step is not a human input or is not awaiting input
131/// - 403 if the caller may not resolve the step
132/// - 404 if the run or the step does not exist
133/// - 409 if the input was already answered or rejected
134#[cfg_attr(
135    feature = "openapi",
136    utoipa::path(
137        post,
138        path = "/api/v1/runs/{id}/steps/{step_id}/reject",
139        tags = ["runs"],
140        params(
141            ("id" = Uuid, Path, description = "Run ID"),
142            ("step_id" = Uuid, Path, description = "Step ID")
143        ),
144        request_body(content = RejectHumanInputRequest, description = "Optional rejection reason"),
145        responses(
146            (status = 200, description = "Input rejected, the run resumes", body = RunResponse),
147            (status = 400, description = "Step is not a human input awaiting input"),
148            (status = 401, description = "Unauthorized"),
149            (status = 403, description = "Forbidden"),
150            (status = 404, description = "Run or step not found"),
151            (status = 409, description = "Input already answered or rejected")
152        ),
153        security(("Bearer" = []))
154    )
155)]
156pub async fn reject_human_input(
157    auth: Authenticated,
158    State(state): State<AppState>,
159    Path((id, step_id)): Path<(Uuid, Uuid)>,
160    body: Option<Json<RejectHumanInputRequest>>,
161) -> Result<impl IntoResponse, ApiError> {
162    auth.require(ApiKeyScope::RunsManage, false)?;
163    let (run, step) = open_input_step(&state, id, step_id).await?;
164    let actor = authorize_gate(&auth, &state, &run, &step).await?;
165
166    let reason = body
167        .and_then(|Json(b)| b.reason)
168        .filter(|r| !r.trim().is_empty())
169        .unwrap_or_else(|| format!("input rejected by {actor}"));
170
171    state
172        .store
173        .update_step(
174            step.id,
175            StepUpdate {
176                status: Some(StepStatus::Rejected),
177                error: Some(reason),
178                completed_at: Some(Utc::now()),
179                clear_approval_deadline: true,
180                ..StepUpdate::default()
181            },
182        )
183        .await?;
184    release_run(&state, &run).await?;
185
186    Ok(ok(RunResponse::from(state.get_run_or_404(id).await?)))
187}
188
189/// Load the run and the human input step, and check the step can be resolved.
190///
191/// A resolved step returns 409 whatever the run status, so replaying an
192/// accepted request is reported as a conflict rather than a bad request.
193async fn open_input_step(
194    state: &AppState,
195    run_id: Uuid,
196    step_id: Uuid,
197) -> Result<(Run, Step), ApiError> {
198    let run = state.get_run_or_404(run_id).await?;
199    let step = state
200        .store
201        .get_step(step_id)
202        .await?
203        .filter(|s| s.run_id == run_id)
204        .ok_or(ApiError::StepNotFound(step_id))?;
205
206    if step.kind != StepKind::HumanInput {
207        return Err(ApiError::BadRequest(
208            "step does not wait for input".to_string(),
209        ));
210    }
211
212    match step.status.state {
213        StepStatus::Completed => {
214            return Err(ApiError::Conflict("input already provided".to_string()));
215        }
216        StepStatus::Rejected => {
217            return Err(ApiError::Conflict("input already rejected".to_string()));
218        }
219        _ => {}
220    }
221
222    if step.status.state != StepStatus::AwaitingApproval || !awaits_gate(&run) {
223        return Err(ApiError::BadRequest(
224            "step is not awaiting input".to_string(),
225        ));
226    }
227
228    Ok((run, step))
229}
230
231/// Validate `answer` against the JSON schema stored in the step input.
232fn validate_answer(input: Option<&Value>, answer: &Value) -> Result<(), ApiError> {
233    let schema = input
234        .and_then(|i| i.get(HUMAN_INPUT_SCHEMA_KEY))
235        .ok_or_else(|| ApiError::Internal("input step has no stored schema".to_string()))?;
236    let validator = validator_for(schema)
237        .map_err(|e| ApiError::Internal(format!("input step has an invalid schema: {e}")))?;
238
239    let errors: Vec<String> = validator
240        .iter_errors(answer)
241        .map(|e| e.to_string())
242        .collect();
243    if errors.is_empty() {
244        Ok(())
245    } else {
246        Err(ApiError::InvalidInput(errors))
247    }
248}
249
250/// Let the run go on once its input is answered or rejected.
251///
252/// A run an operator paused while it waited stays paused: it resumes to
253/// `Pending` once the operator resumes it.
254async fn release_run(state: &AppState, run: &Run) -> Result<(), ApiError> {
255    if run.status.state == RunStatus::Paused {
256        return resume_paused_run_later(state, run.id).await;
257    }
258    state
259        .store
260        .update_run_status(run.id, resume_status(state))
261        .await?;
262    resume_in_background(state, run.id);
263    Ok(())
264}
265
266/// The status a run moves to once its input is answered or rejected:
267/// `Running` to resume it here, `Pending` to requeue it for a worker.
268fn resume_status(state: &AppState) -> RunStatus {
269    match state.engine.execution_mode() {
270        ExecutionMode::Local => RunStatus::Running,
271        ExecutionMode::Workers => RunStatus::Pending,
272    }
273}
274
275/// Under [`ExecutionMode::Local`], resume the run in the background; the
276/// handler replays up to the input. Under [`ExecutionMode::Workers`], do
277/// nothing: the run is already `Pending` and a worker picks it up.
278fn resume_in_background(state: &AppState, id: Uuid) {
279    if !matches!(state.engine.execution_mode(), ExecutionMode::Local) {
280        return;
281    }
282    let engine = state.engine.clone();
283    spawn(async move {
284        if let Err(err) = engine.resume_run(id).await {
285            error!(run_id = %id, error = %err, "failed to resume run after human input");
286        }
287    });
288}
289
290#[cfg(test)]
291mod tests {
292    use std::collections::HashMap;
293    use std::sync::Arc;
294
295    use axum::Router;
296    use axum::body::Body;
297    use axum::http::{Request, StatusCode};
298    use axum::routing::post;
299    use http_body_util::BodyExt;
300    use ironflow_auth::jwt::{AccessToken, JwtConfig};
301    use ironflow_auth::password;
302    use ironflow_core::providers::claude::ClaudeCodeProvider;
303    use ironflow_engine::engine::Engine;
304    use ironflow_engine::handler::input_schema_for;
305    use ironflow_engine::notify::Event;
306    use ironflow_store::memory::InMemoryStore;
307    use ironflow_store::models::{
308        Assignee, NewRun, NewStep, NewUser, TriggerKind, User, step_trace_id,
309    };
310    use ironflow_store::store::RunStore;
311    use ironflow_store::user_store::UserStore;
312    use schemars::JsonSchema;
313    use serde_json::{from_slice, json};
314    use tokio::sync::broadcast;
315    use tower::ServiceExt;
316
317    use super::*;
318    use crate::routes::test_helpers::create_user_auth_header;
319
320    /// The answer type the test steps ask for.
321    #[allow(dead_code)]
322    #[derive(Deserialize, JsonSchema)]
323    struct Answers {
324        answers: Vec<String>,
325    }
326
327    /// An answer type whose schema carries a `$defs` reference.
328    #[allow(dead_code)]
329    #[derive(Deserialize, JsonSchema)]
330    struct Nested {
331        target: Target,
332    }
333
334    #[allow(dead_code)]
335    #[derive(Deserialize, JsonSchema)]
336    struct Target {
337        env: String,
338    }
339
340    fn test_state(store: Arc<InMemoryStore>) -> AppState {
341        let provider = Arc::new(ClaudeCodeProvider::new());
342        let engine = Arc::new(Engine::new(store.clone(), provider));
343        let jwt_config = Arc::new(JwtConfig {
344            secret: "test-secret".to_string(),
345            access_token_ttl_secs: 900,
346            refresh_token_ttl_secs: 604800,
347            cookie_domain: None,
348            cookie_secure: false,
349        });
350        let (event_sender, _) = broadcast::channel::<Event>(1);
351        AppState::new(
352            store,
353            engine,
354            jwt_config,
355            "test-worker-token".to_string(),
356            event_sender,
357        )
358    }
359
360    async fn member(store: &Arc<InMemoryStore>, username: &str) -> User {
361        let password_hash = password::hash("password123").expect("hash");
362        store
363            .create_user(NewUser {
364                email: format!("{username}@example.com"),
365                username: username.to_string(),
366                password_hash,
367                // The first user would otherwise become an implicit admin.
368                is_admin: Some(false),
369            })
370            .await
371            .expect("create user")
372    }
373
374    fn member_header(user: &User, state: &AppState) -> String {
375        let token = AccessToken::for_user(user.id, &user.username, false, &state.jwt_config)
376            .expect("token");
377        format!("Bearer {}", token.0)
378    }
379
380    async fn awaiting_run(store: &Arc<InMemoryStore>) -> Uuid {
381        let run = store
382            .create_run(NewRun {
383                created_by: None,
384                workflow_name: "test".to_string(),
385                trigger: TriggerKind::Manual,
386                payload: json!({}),
387                max_retries: 0,
388                handler_version: None,
389                labels: HashMap::new(),
390                scheduled_at: None,
391                idempotency_key: None,
392                concurrency_key: None,
393                priority: 0,
394                concurrency_limits: Vec::new(),
395                max_cost_usd: None,
396                worker_tags: Vec::new(),
397            })
398            .await
399            .unwrap()
400            .into_run();
401        store
402            .update_run_status(run.id, RunStatus::Running)
403            .await
404            .unwrap();
405        store
406            .update_run_status(run.id, RunStatus::AwaitingApproval)
407            .await
408            .unwrap();
409        run.id
410    }
411
412    /// A step of `kind` awaiting approval on `run_id`.
413    async fn awaiting_step(
414        store: &Arc<InMemoryStore>,
415        run_id: Uuid,
416        kind: StepKind,
417        assignee: Option<Assignee>,
418    ) -> Uuid {
419        let step = store
420            .create_step(NewStep {
421                run_id,
422                trace_id: step_trace_id(run_id, "clarify", 0),
423                name: "clarify".to_string(),
424                kind,
425                position: 0,
426                input: Some(json!({
427                    "message": "Answer the questions",
428                    HUMAN_INPUT_SCHEMA_KEY: input_schema_for::<Answers>(),
429                })),
430                is_error_handler: false,
431            })
432            .await
433            .unwrap();
434        store
435            .update_step(
436                step.id,
437                StepUpdate {
438                    status: Some(StepStatus::Running),
439                    ..StepUpdate::default()
440                },
441            )
442            .await
443            .unwrap();
444        store
445            .update_step(
446                step.id,
447                StepUpdate {
448                    status: Some(StepStatus::AwaitingApproval),
449                    approval_assignee: assignee,
450                    ..StepUpdate::default()
451                },
452            )
453            .await
454            .unwrap();
455        step.id
456    }
457
458    /// A run awaiting a human input step, with an admin header.
459    async fn setup() -> (Arc<InMemoryStore>, AppState, Uuid, Uuid) {
460        let store = Arc::new(InMemoryStore::new());
461        let run_id = awaiting_run(&store).await;
462        let step_id = awaiting_step(&store, run_id, StepKind::HumanInput, None).await;
463        let state = test_state(store.clone());
464        (store, state, run_id, step_id)
465    }
466
467    fn app(state: AppState) -> Router {
468        Router::new()
469            .route("/{id}/steps/{step_id}/input", post(submit_human_input))
470            .route("/{id}/steps/{step_id}/reject", post(reject_human_input))
471            .with_state(state)
472    }
473
474    /// POST `body` (or nothing) to `/{run_id}/steps/{step_id}/{verb}`.
475    async fn call(
476        state: &AppState,
477        auth: &str,
478        run_id: Uuid,
479        step_id: Uuid,
480        verb: &str,
481        body: Option<Value>,
482    ) -> (StatusCode, Value) {
483        let builder = Request::builder()
484            .method("POST")
485            .uri(format!("/{run_id}/steps/{step_id}/{verb}"))
486            .header("authorization", auth);
487        let req = match body {
488            Some(body) => builder
489                .header("content-type", "application/json")
490                .body(Body::from(body.to_string()))
491                .unwrap(),
492            None => builder.body(Body::empty()).unwrap(),
493        };
494
495        let resp = app(state.clone()).oneshot(req).await.unwrap();
496        let status = resp.status();
497        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
498        let json_val = if bytes.is_empty() {
499            Value::Null
500        } else {
501            from_slice(&bytes).unwrap()
502        };
503        (status, json_val)
504    }
505
506    #[tokio::test]
507    async fn human_input_submit_valid_answer_resumes_the_run() {
508        let (store, state, run_id, step_id) = setup().await;
509        let auth = create_user_auth_header(&state, "admin", true).await;
510        let answer = json!({"answers": ["staging"]});
511
512        let (status, body) = call(
513            &state,
514            &auth,
515            run_id,
516            step_id,
517            "input",
518            Some(answer.clone()),
519        )
520        .await;
521
522        assert_eq!(status, StatusCode::OK);
523        assert_eq!(body["data"]["status"], "running");
524        let step = store.get_step(step_id).await.unwrap().unwrap();
525        assert_eq!(step.status.state, StepStatus::Completed);
526        assert_eq!(step.output, Some(answer));
527        assert_eq!(step.approvals.len(), 1);
528        assert_eq!(step.approvals[0].approved_by, "admin");
529    }
530
531    #[tokio::test]
532    async fn human_input_submit_invalid_answer_returns_422() {
533        let (store, state, run_id, step_id) = setup().await;
534        let auth = create_user_auth_header(&state, "admin", true).await;
535
536        for answer in [json!({"answers": 3}), json!({})] {
537            let (status, body) = call(&state, &auth, run_id, step_id, "input", Some(answer)).await;
538
539            assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
540            assert_eq!(body["error"]["code"], "INVALID_INPUT");
541            let errors = body["error"]["details"]["errors"]
542                .as_array()
543                .expect("errors array");
544            assert!(!errors.is_empty());
545        }
546
547        let step = store.get_step(step_id).await.unwrap().unwrap();
548        assert_eq!(step.status.state, StepStatus::AwaitingApproval);
549        let run = store.get_run(run_id).await.unwrap().unwrap();
550        assert_eq!(run.status.state, RunStatus::AwaitingApproval);
551    }
552
553    #[tokio::test]
554    async fn human_input_submit_twice_returns_409() {
555        let (_store, state, run_id, step_id) = setup().await;
556        let auth = create_user_auth_header(&state, "admin", true).await;
557        let answer = json!({"answers": ["yes"]});
558
559        let (first, _) = call(
560            &state,
561            &auth,
562            run_id,
563            step_id,
564            "input",
565            Some(answer.clone()),
566        )
567        .await;
568        assert_eq!(first, StatusCode::OK);
569
570        let (second, body) = call(&state, &auth, run_id, step_id, "input", Some(answer)).await;
571        assert_eq!(second, StatusCode::CONFLICT);
572        assert_eq!(body["error"]["message"], "input already provided");
573    }
574
575    #[tokio::test]
576    async fn human_input_submit_on_an_approval_step_returns_400() {
577        let store = Arc::new(InMemoryStore::new());
578        let run_id = awaiting_run(&store).await;
579        let step_id = awaiting_step(&store, run_id, StepKind::Approval, None).await;
580        let state = test_state(store);
581        let auth = create_user_auth_header(&state, "admin", true).await;
582
583        let answer = json!({"answers": []});
584        let (status, _) = call(&state, &auth, run_id, step_id, "input", Some(answer)).await;
585        assert_eq!(status, StatusCode::BAD_REQUEST);
586    }
587
588    #[tokio::test]
589    async fn human_input_submit_on_a_run_not_awaiting_returns_400() {
590        let (store, state, run_id, step_id) = setup().await;
591        store
592            .update_run_status(run_id, RunStatus::Running)
593            .await
594            .unwrap();
595        let auth = create_user_auth_header(&state, "admin", true).await;
596
597        let answer = json!({"answers": []});
598        let (status, _) = call(&state, &auth, run_id, step_id, "input", Some(answer)).await;
599        assert_eq!(status, StatusCode::BAD_REQUEST);
600    }
601
602    #[tokio::test]
603    async fn human_input_unknown_step_or_step_of_another_run_returns_404() {
604        let (store, state, run_id, step_id) = setup().await;
605        let auth = create_user_auth_header(&state, "admin", true).await;
606        let answer = json!({"answers": []});
607
608        let unknown = Uuid::now_v7();
609        let (status, body) = call(
610            &state,
611            &auth,
612            run_id,
613            unknown,
614            "input",
615            Some(answer.clone()),
616        )
617        .await;
618        assert_eq!(status, StatusCode::NOT_FOUND);
619        assert_eq!(body["error"]["code"], "STEP_NOT_FOUND");
620
621        let other_run = awaiting_run(&store).await;
622        let (status, _) = call(&state, &auth, other_run, step_id, "input", Some(answer)).await;
623        assert_eq!(status, StatusCode::NOT_FOUND);
624    }
625
626    #[tokio::test]
627    async fn human_input_unknown_run_returns_404() {
628        let (_store, state, _run_id, step_id) = setup().await;
629        let auth = create_user_auth_header(&state, "admin", true).await;
630
631        let answer = json!({"answers": []});
632        let (status, body) = call(
633            &state,
634            &auth,
635            Uuid::now_v7(),
636            step_id,
637            "input",
638            Some(answer),
639        )
640        .await;
641        assert_eq!(status, StatusCode::NOT_FOUND);
642        assert_eq!(body["error"]["code"], "RUN_NOT_FOUND");
643    }
644
645    #[tokio::test]
646    async fn human_input_non_admin_non_assignee_is_forbidden() {
647        let store = Arc::new(InMemoryStore::new());
648        let alice = member(&store, "alice").await;
649        let bob = member(&store, "bob").await;
650        let run_id = awaiting_run(&store).await;
651        let assignee = Some(Assignee::user(&alice.username));
652        let step_id = awaiting_step(&store, run_id, StepKind::HumanInput, assignee).await;
653        let state = test_state(store.clone());
654
655        let answer = json!({"answers": ["yes"]});
656        let bob_auth = member_header(&bob, &state);
657        let (status, _) = call(
658            &state,
659            &bob_auth,
660            run_id,
661            step_id,
662            "input",
663            Some(answer.clone()),
664        )
665        .await;
666        assert_eq!(status, StatusCode::FORBIDDEN);
667
668        let alice_auth = member_header(&alice, &state);
669        let (status, _) = call(&state, &alice_auth, run_id, step_id, "input", Some(answer)).await;
670        assert_eq!(status, StatusCode::OK);
671        let step = store.get_step(step_id).await.unwrap().unwrap();
672        assert_eq!(step.approvals[0].user_id, alice.id);
673    }
674
675    #[tokio::test]
676    async fn human_input_submit_on_a_paused_run_keeps_it_paused() {
677        let (store, state, run_id, step_id) = setup().await;
678        store
679            .update_run_status(run_id, RunStatus::Paused)
680            .await
681            .unwrap();
682        let auth = create_user_auth_header(&state, "admin", true).await;
683        let answer = json!({"answers": ["staging"]});
684
685        let (status, body) = call(&state, &auth, run_id, step_id, "input", Some(answer)).await;
686
687        assert_eq!(status, StatusCode::OK);
688        assert_eq!(body["data"]["status"], "paused");
689        assert_eq!(body["data"]["resume_status"], "pending");
690        let step = store.get_step(step_id).await.unwrap().unwrap();
691        assert_eq!(step.status.state, StepStatus::Completed);
692        let run = store.get_run(run_id).await.unwrap().unwrap();
693        assert_eq!(run.status.state, RunStatus::Paused);
694        assert_eq!(run.resume_status, Some(RunStatus::Pending));
695    }
696
697    #[tokio::test]
698    async fn human_input_reject_marks_the_step_and_resumes_the_run() {
699        let (store, state, run_id, step_id) = setup().await;
700        let auth = create_user_auth_header(&state, "admin", true).await;
701
702        let body = json!({"reason": "out of scope"});
703        let (status, resp) = call(&state, &auth, run_id, step_id, "reject", Some(body)).await;
704
705        assert_eq!(status, StatusCode::OK);
706        assert_eq!(resp["data"]["status"], "running");
707        let step = store.get_step(step_id).await.unwrap().unwrap();
708        assert_eq!(step.status.state, StepStatus::Rejected);
709        assert_eq!(step.error.as_deref(), Some("out of scope"));
710        assert!(step.approval_deadline_at.is_none());
711    }
712
713    /// A run awaiting a human input assigned to a member, on an engine that
714    /// requeues resumed runs for a worker.
715    async fn workers_setup() -> (Arc<InMemoryStore>, AppState, Uuid, Uuid, String) {
716        let store = Arc::new(InMemoryStore::new());
717        let alice = member(&store, "alice").await;
718        let run_id = awaiting_run(&store).await;
719        let assignee = Some(Assignee::user(&alice.username));
720        let step_id = awaiting_step(&store, run_id, StepKind::HumanInput, assignee).await;
721        let provider = Arc::new(ClaudeCodeProvider::new());
722        let engine =
723            Engine::new(store.clone(), provider).with_execution_mode(ExecutionMode::Workers);
724        let jwt_config = Arc::new(JwtConfig {
725            secret: "test-secret".to_string(),
726            access_token_ttl_secs: 900,
727            refresh_token_ttl_secs: 604800,
728            cookie_domain: None,
729            cookie_secure: false,
730        });
731        let (event_sender, _) = broadcast::channel::<Event>(1);
732        let state = AppState::new(
733            store.clone(),
734            Arc::new(engine),
735            jwt_config,
736            "test-worker-token".to_string(),
737            event_sender,
738        );
739        let auth = member_header(&alice, &state);
740        (store, state, run_id, step_id, auth)
741    }
742
743    #[tokio::test]
744    async fn human_input_submit_in_workers_mode_requeues_the_run() {
745        let (store, state, run_id, step_id, auth) = workers_setup().await;
746
747        let answer = json!({"answers": ["ok"]});
748        let (status, body) = call(&state, &auth, run_id, step_id, "input", Some(answer)).await;
749
750        assert_eq!(status, StatusCode::OK);
751        assert_eq!(body["data"]["status"], "pending");
752        let run = store.get_run(run_id).await.unwrap().unwrap();
753        assert_eq!(run.status.state, RunStatus::Pending);
754        let step = store.get_step(step_id).await.unwrap().unwrap();
755        assert_eq!(step.status.state, StepStatus::Completed);
756    }
757
758    #[tokio::test]
759    async fn human_input_reject_in_workers_mode_requeues_the_run() {
760        let (store, state, run_id, step_id, auth) = workers_setup().await;
761
762        let body = json!({"reason": "out of scope"});
763        let (status, resp) = call(&state, &auth, run_id, step_id, "reject", Some(body)).await;
764
765        assert_eq!(status, StatusCode::OK);
766        assert_eq!(resp["data"]["status"], "pending");
767        let run = store.get_run(run_id).await.unwrap().unwrap();
768        assert_eq!(run.status.state, RunStatus::Pending);
769        let step = store.get_step(step_id).await.unwrap().unwrap();
770        assert_eq!(step.status.state, StepStatus::Rejected);
771    }
772
773    #[tokio::test]
774    async fn human_input_reject_without_body_uses_the_default_reason() {
775        let (store, state, run_id, step_id) = setup().await;
776        let auth = create_user_auth_header(&state, "admin", true).await;
777
778        let (status, _) = call(&state, &auth, run_id, step_id, "reject", None).await;
779
780        assert_eq!(status, StatusCode::OK);
781        let step = store.get_step(step_id).await.unwrap().unwrap();
782        assert_eq!(step.error.as_deref(), Some("input rejected by admin"));
783    }
784
785    #[tokio::test]
786    async fn human_input_reject_with_a_blank_reason_uses_the_default_reason() {
787        let (store, state, run_id, step_id) = setup().await;
788        let auth = create_user_auth_header(&state, "admin", true).await;
789
790        let body = json!({"reason": "   "});
791        let (status, _) = call(&state, &auth, run_id, step_id, "reject", Some(body)).await;
792
793        assert_eq!(status, StatusCode::OK);
794        let step = store.get_step(step_id).await.unwrap().unwrap();
795        assert_eq!(step.error.as_deref(), Some("input rejected by admin"));
796    }
797
798    #[tokio::test]
799    async fn human_input_reject_after_submit_returns_409() {
800        let (_store, state, run_id, step_id) = setup().await;
801        let auth = create_user_auth_header(&state, "admin", true).await;
802
803        let answer = json!({"answers": ["yes"]});
804        let (first, _) = call(&state, &auth, run_id, step_id, "input", Some(answer)).await;
805        assert_eq!(first, StatusCode::OK);
806
807        let (second, _) = call(&state, &auth, run_id, step_id, "reject", None).await;
808        assert_eq!(second, StatusCode::CONFLICT);
809    }
810
811    #[tokio::test]
812    async fn human_input_submit_after_reject_returns_409() {
813        let (_store, state, run_id, step_id) = setup().await;
814        let auth = create_user_auth_header(&state, "admin", true).await;
815
816        let (first, _) = call(&state, &auth, run_id, step_id, "reject", None).await;
817        assert_eq!(first, StatusCode::OK);
818
819        let answer = json!({"answers": ["yes"]});
820        let (second, body) = call(&state, &auth, run_id, step_id, "input", Some(answer)).await;
821        assert_eq!(second, StatusCode::CONFLICT);
822        assert_eq!(body["error"]["message"], "input already rejected");
823    }
824
825    #[test]
826    fn human_input_validate_answer_without_a_schema_is_internal() {
827        let err = validate_answer(Some(&json!({"message": "Answer?"})), &json!({}))
828            .expect_err("no schema");
829        assert!(matches!(err, ApiError::Internal(_)), "got {err:?}");
830
831        let err = validate_answer(None, &json!({})).expect_err("no input");
832        assert!(matches!(err, ApiError::Internal(_)), "got {err:?}");
833    }
834
835    #[test]
836    fn human_input_validate_answer_resolves_defs_references() {
837        let schema = input_schema_for::<Nested>();
838        assert!(schema.get("$defs").is_some(), "schema: {schema}");
839        let input = json!({ HUMAN_INPUT_SCHEMA_KEY: schema });
840
841        let valid = validate_answer(Some(&input), &json!({"target": {"env": "prod"}}));
842        assert!(valid.is_ok(), "got {valid:?}");
843
844        let err = validate_answer(Some(&input), &json!({"target": {"env": 1}}))
845            .expect_err("wrong nested type");
846        match err {
847            ApiError::InvalidInput(errors) => assert_eq!(errors.len(), 1, "got {errors:?}"),
848            other => panic!("expected InvalidInput, got {other:?}"),
849        }
850    }
851}
ironflow_api/routes/human_input.rs

ironflow_api/routes/
human_input.rs
