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

ironflow_api/routes/
human_input.rs
