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

ironflow_api/routes/
human_input.rs
