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

ironflow_api/routes/
human_input.rs
