From 497281cdd7388634c6f2066cda08244a2e47f6ba Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 17 Sep 2026 22:31:35 -0400 Subject: [PATCH] Cover human gates in Petri runs through the questions API The server scenario answers a gate in the in-process run through the questions API and checks the branch it routed and the cleared pending question. The CLI scenarios drive the real worker: a gate answered through the API over the worker's control channel, two parallel gates each bound to their own answer, and an unanswered gate that expires with its default and records `interview.timeout`. Co-Authored-By: Claude Fable 5.1 --- lib/apps/fabro-cli/tests/it/scenario/petri.rs | 267 ++++++++++++++++-- .../fabro-server/tests/it/scenario/petri.rs | 142 ++++++++++ 2 files changed, 392 insertions(+), 17 deletions(-) diff --git a/lib/apps/fabro-cli/tests/it/scenario/petri.rs b/lib/apps/fabro-cli/tests/it/scenario/petri.rs index ec95672b1..a3ea4903c 100644 --- a/lib/apps/fabro-cli/tests/it/scenario/petri.rs +++ b/lib/apps/fabro-cli/tests/it/scenario/petri.rs @@ -251,17 +251,22 @@ async fn wait_for_http_ready(base_url: &str, child: &mut Child) { /// A workspace holding a command-only bundle whose `workflow.toml` names /// Petri, with the given stage script. fn write_petri_workspace(context: &fabro_test::TestContext, script: &str) -> PathBuf { - let workspace = context.temp_dir.join("petri-workspace"); - std::fs::create_dir_all(&workspace).expect("the workspace creates"); - std::fs::write( - workspace.join("workflow.fabro"), - format!( + write_petri_workflow( + context, + &format!( "digraph Command {{\n graph [goal=\"Run one command\", default_max_retries=0]\n start \ [shape=Mdiamond]\n exit [shape=Msquare]\n say [shape=parallelogram, \ script=\"{script}\", max_retries=0]\n start -> say -> exit\n}}\n" ), ) - .expect("the workflow writes"); +} + +/// A workspace holding the given workflow with a `workflow.toml` that names +/// Petri. +fn write_petri_workflow(context: &fabro_test::TestContext, dot: &str) -> PathBuf { + let workspace = context.temp_dir.join("petri-workspace"); + std::fs::create_dir_all(&workspace).expect("the workspace creates"); + std::fs::write(workspace.join("workflow.fabro"), dot).expect("the workflow writes"); std::fs::write( workspace.join("workflow.toml"), "_version = 1\n\n[workflow]\ngraph = \"workflow.fabro\"\nengine = \"petri\"\n\n[run]\ngoal \ @@ -271,12 +276,22 @@ fn write_petri_workspace(context: &fabro_test::TestContext, script: &str) -> Pat workspace } -/// `fabro run --detach` against the server: the run is created and started, -/// and its id comes back. +/// `fabro run --detach --auto-approve` against the server: the run is +/// created and started, and its id comes back. fn run_detached( context: &fabro_test::TestContext, server: &RunningServer, workspace: &Path, +) -> String { + run_detached_with(context, server, workspace, &["--auto-approve"]) +} + +/// `fabro run --detach` against the server with extra arguments. +fn run_detached_with( + context: &fabro_test::TestContext, + server: &RunningServer, + workspace: &Path, + extra: &[&str], ) -> String { let target = server.target(); seed_dev_token_auth( @@ -287,15 +302,9 @@ fn run_detached( let output = context .run_cmd() .current_dir(workspace) - .args([ - "--server", - &target, - "--detach", - "--auto-approve", - "--environment", - "local", - "workflow.toml", - ]) + .args(["--server", &target, "--detach"]) + .args(extra) + .args(["--environment", "local", "workflow.toml"]) .output() .expect("the detached run executes"); assert!( @@ -566,3 +575,227 @@ async fn run_status_offline(server: &RunningServer) -> Option { .ok() .map(|response| response.status().to_string()) } + +/// The run's pending questions, as the API lists them. +async fn questions(server: &RunningServer, run_id: &str) -> Vec { + run_json(server, &format!("runs/{run_id}/questions")).await["data"] + .as_array() + .cloned() + .expect("the questions list is an array") +} + +/// Wait until `count` questions are pending at once. +async fn wait_for_questions( + server: &RunningServer, + run_id: &str, + count: usize, +) -> Vec { + let deadline = Instant::now() + RUN_TIMEOUT; + loop { + let pending = questions(server, run_id).await; + if pending.len() >= count { + return pending; + } + assert!( + Instant::now() < deadline, + "run {run_id} did not ask {count} question(s); pending: {pending:?}" + ); + tokio::time::sleep(POLL).await; + } +} + +/// Answer a question through the API, as the web app and the CLI do. +async fn answer(server: &RunningServer, run_id: &str, question_id: &str, body: serde_json::Value) { + let response = fabro_test::test_http_client() + .post(format!( + "{}/api/v1/runs/{run_id}/questions/{question_id}/answer", + server.api_base_url + )) + .bearer_auth(TEST_DEV_TOKEN) + .json(&body) + .send() + .await + .expect("the answer sends"); + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + assert_eq!( + status, + fabro_http::StatusCode::NO_CONTENT, + "POST /api/v1/runs/{run_id}/questions/{question_id}/answer: {body}" + ); +} + +/// A yes/no gate whose branches each leave a marker file. +fn gate_dot(markers: &Path, gate_attrs: &str) -> String { + format!( + "digraph Gate {{\n graph [goal=\"Ask before running\"]\n start [shape=Mdiamond]\n \ + exit [shape=Msquare]\n gate [shape=hexagon, label=\"Go?\", \ + question_type=\"yes_no\"{gate_attrs}]\n yes [shape=parallelogram, script=\"touch \ + {dir}/yes\"]\n no [shape=parallelogram, script=\"touch {dir}/no\"]\n start -> gate\n \ + gate -> yes [label=\"[Y] Yes\"]\n gate -> no [label=\"[N] No\"]\n yes -> exit\n no \ + -> exit\n}}\n", + dir = markers.display() + ) +} + +/// Two gates as the branches of one parallel node; the join's results are +/// written out, so each gate's answer is read from its branch result. +fn two_gates_dot(markers: &Path) -> String { + format!( + "digraph Gates {{\n graph [goal=\"Ask twice at once\"]\n start [shape=Mdiamond]\n \ + exit [shape=Msquare]\n fan [shape=component]\n a [shape=hexagon, label=\"A?\", \ + question_type=\"yes_no\"]\n b [shape=hexagon, label=\"B?\", \ + question_type=\"yes_no\"]\n join [shape=tripleoctagon]\n report \ + [shape=parallelogram, script=\"cat > {dir}/results.json\", \ + stdin_source=\"context.parallel.results\"]\n start -> fan\n fan -> a\n fan -> b\n \ + a -> join [label=\"[Y] Yes\"]\n a -> join [label=\"[N] No\"]\n b -> join [label=\"[Y] \ + Yes\"]\n b -> join [label=\"[N] No\"]\n join -> report -> exit\n}}\n", + dir = markers.display() + ) +} + +/// A human gate in the worker asks through the server: the question is +/// listed by the questions API with the gate's stage and options, the +/// answer reaches the worker over its control channel and routes the gate, +/// and the run's stream records the interview. +#[tokio::test(flavor = "multi_thread")] +async fn a_human_gate_in_the_worker_is_answered_through_the_api() { + if host_plugin().is_none() { + return; + } + let context = test_context!(); + let server = RunningServer::start().await; + let markers = context.temp_dir.join("markers"); + std::fs::create_dir_all(&markers).expect("the marker dir creates"); + let workspace = write_petri_workflow(&context, &gate_dot(&markers, "")); + let run_id = run_detached_with(&context, &server, &workspace, &[]); + + let pending = wait_for_questions(&server, &run_id, 1).await; + let question = &pending[0]; + assert_eq!(question["stage"], "gate", "{question}"); + assert_eq!(question["question_type"], "yes_no", "{question}"); + let question_id = question["id"].as_str().expect("an id").to_string(); + answer( + &server, + &run_id, + &question_id, + serde_json::json!({ "kind": "no" }), + ) + .await; + + let status = wait_for_status(&server, &run_id, &["succeeded", "failed"]).await; + assert_eq!( + status, + "succeeded", + "server stderr:\n{}", + server.stderr_text() + ); + assert!( + markers.join("no").exists() && !markers.join("yes").exists(), + "the no branch ran" + ); + let names = run_events(&server, &run_id).await; + let names = event_names(&names); + assert!( + names.contains(&"interview.started") && names.contains(&"interview.completed"), + "{names:?}" + ); + assert!(questions(&server, &run_id).await.is_empty()); + let store = server.petri_store().await; + let outcome = engine::outcome_of(&store, &run_id) + .await + .expect("the run's Petri record inspects"); + assert_eq!(outcome.status, RunStatus::Success, "{outcome:?}"); + server.shutdown(); +} + +/// Two branches of a parallel node ask at once; each answer, given through +/// the API in the other order, binds to its own branch. +#[tokio::test(flavor = "multi_thread")] +async fn two_parallel_gates_in_the_worker_each_bind_their_own_answer() { + if host_plugin().is_none() { + return; + } + let context = test_context!(); + let server = RunningServer::start().await; + let markers = context.temp_dir.join("markers"); + std::fs::create_dir_all(&markers).expect("the marker dir creates"); + let workspace = write_petri_workflow(&context, &two_gates_dot(&markers)); + let run_id = run_detached_with(&context, &server, &workspace, &[]); + + let pending = wait_for_questions(&server, &run_id, 2).await; + let id_of = |stage: &str| { + pending + .iter() + .find(|question| question["stage"] == stage) + .and_then(|question| question["id"].as_str()) + .unwrap_or_else(|| panic!("`{stage}` is pending: {pending:?}")) + .to_string() + }; + let (a, b) = (id_of("a"), id_of("b")); + assert_ne!(a, b); + for question in &pending { + assert_eq!(question["question_type"], "yes_no", "{question}"); + } + // A yes/no question takes `yes` or `no`, as the API validates it. + answer(&server, &run_id, &b, serde_json::json!({ "kind": "yes" })).await; + answer(&server, &run_id, &a, serde_json::json!({ "kind": "no" })).await; + + let status = wait_for_status(&server, &run_id, &["succeeded", "failed"]).await; + assert_eq!( + status, + "succeeded", + "server stderr:\n{}", + server.stderr_text() + ); + let results: serde_json::Value = serde_json::from_str( + &std::fs::read_to_string(markers.join("results.json")).expect("the join wrote its results"), + ) + .expect("the results parse"); + let results = results.as_array().expect("a list of branch results"); + assert_eq!(results.len(), 2, "{results:?}"); + assert_eq!(results[0]["id"], "a"); + assert_eq!(results[0]["context_updates"]["human.gate.selected"], "N"); + assert_eq!(results[1]["id"], "b"); + assert_eq!(results[1]["context_updates"]["human.gate.selected"], "Y"); + server.shutdown(); +} + +/// A gate nobody answers expires on its own deadline: the run takes the +/// gate's default, the stream records the timeout, and nothing stays +/// pending. +#[tokio::test(flavor = "multi_thread")] +async fn an_unanswered_gate_in_the_worker_expires_with_its_default() { + if host_plugin().is_none() { + return; + } + let context = test_context!(); + let server = RunningServer::start().await; + let markers = context.temp_dir.join("markers"); + std::fs::create_dir_all(&markers).expect("the marker dir creates"); + let workspace = write_petri_workflow( + &context, + &gate_dot(&markers, ", timeout=\"2s\", human.default_choice=\"no\""), + ); + let run_id = run_detached_with(&context, &server, &workspace, &[]); + + let pending = wait_for_questions(&server, &run_id, 1).await; + assert_eq!(pending[0]["timeout_seconds"], 2.0, "{}", pending[0]); + + let status = wait_for_status(&server, &run_id, &["succeeded", "failed"]).await; + assert_eq!( + status, + "succeeded", + "server stderr:\n{}", + server.stderr_text() + ); + assert!( + markers.join("no").exists() && !markers.join("yes").exists(), + "the default ran" + ); + let events = run_events(&server, &run_id).await; + let names = event_names(&events); + assert!(names.contains(&"interview.timeout"), "{names:?}"); + assert!(questions(&server, &run_id).await.is_empty()); + server.shutdown(); +} diff --git a/lib/apps/fabro-server/tests/it/scenario/petri.rs b/lib/apps/fabro-server/tests/it/scenario/petri.rs index ae2dae66c..728eaee22 100644 --- a/lib/apps/fabro-server/tests/it/scenario/petri.rs +++ b/lib/apps/fabro-server/tests/it/scenario/petri.rs @@ -390,3 +390,145 @@ async fn an_unknown_model_is_refused_at_create_with_attractor_model_unknown() { "expected the admission diagnostic in the detail, got {body}" ); } + +/// A yes/no gate whose branches each leave a marker file. +fn gate_dot(markers: &std::path::Path) -> String { + format!( + r#"digraph Gate {{ + graph [goal="Ask before running"] + start [shape=Mdiamond] + exit [shape=Msquare] + gate [shape=hexagon, label="Go?", question_type="yes_no"] + yes [shape=parallelogram, script="touch {dir}/yes"] + no [shape=parallelogram, script="touch {dir}/no"] + start -> gate + gate -> yes [label="[Y] Yes"] + gate -> no [label="[N] No"] + yes -> exit + no -> exit +}}"#, + dir = markers.display() + ) +} + +/// The run's first pending question, once one is listed. +async fn wait_for_question(app: &axum::Router, run_id: &str) -> serde_json::Value { + for _ in 0..600 { + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/questions"))) + .body(Body::empty()) + .expect("questions request should build"); + let response = app + .clone() + .oneshot(req) + .await + .expect("questions request routes"); + let body = response_json( + response, + StatusCode::OK, + format!("GET /api/v1/runs/{run_id}/questions"), + ) + .await; + if let Some(question) = body["data"].as_array().and_then(|items| items.first()) { + return question.clone(); + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + panic!("run {run_id} never asked a question"); +} + +/// A human gate in a Petri run asks through the questions API and is +/// answered through it: the question is listed with the gate's stage and +/// options, the answer routes the gate, and the run's stream records the +/// interview as a legacy stage's would. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_human_gate_is_answered_through_the_questions_api() { + if host_plugin().is_none() { + return; + } + let workspace = tempfile::tempdir().expect("workspace tempdir"); + let markers = tempfile::tempdir().expect("marker tempdir"); + let settings = settings_from_toml( + "_version = 1\n\n[run.environment]\nid = \"local\"\n\n[server.execution]\nengine = \ + \"petri\"\n", + ); + let state = test_app_state_with_options(settings, 5); + let app = test_app_with_scheduler(Arc::clone(&state)); + + let dot = gate_dot(markers.path()); + let version_id = register_version(&app, &[ + ("workflow.fabro", &dot), + ("workflow.toml", PLAIN_SETTINGS), + ]) + .await; + let run_id = + create_and_start_run_from_intent(&app, intent(&version_id, workspace.path())).await; + + let question = wait_for_question(&app, &run_id).await; + assert_eq!(question["stage"], "gate", "{question}"); + assert_eq!(question["text"], "Go?", "{question}"); + assert_eq!(question["question_type"], "yes_no", "{question}"); + let keys: Vec<&str> = question["options"] + .as_array() + .expect("options") + .iter() + .filter_map(|option| option["key"].as_str()) + .collect(); + assert_eq!(keys, vec!["Y", "N"], "{question}"); + let question_id = question["id"].as_str().expect("an id").to_string(); + assert!(question_id.starts_with("gate."), "{question_id}"); + + let req = Request::builder() + .method("POST") + .uri(api(&format!( + "/runs/{run_id}/questions/{question_id}/answer" + ))) + .header("content-type", "application/json") + .body(Body::from(r#"{"kind":"no"}"#)) + .expect("answer request should build"); + let response = app + .clone() + .oneshot(req) + .await + .expect("answer request routes"); + crate::helpers::response_status( + response, + StatusCode::NO_CONTENT, + format!("POST /api/v1/runs/{run_id}/questions/{question_id}/answer"), + ) + .await; + + let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await; + let run = run_json(&app, &run_id).await; + assert_eq!(status, "succeeded", "run: {run}"); + assert!( + markers.path().join("no").exists() && !markers.path().join("yes").exists(), + "the no branch ran" + ); + let outcome = petri_outcome(&state, &run_id).await; + assert_eq!(outcome.status, RunStatus::Success, "{outcome:?}"); + let state_body = { + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/state"))) + .body(Body::empty()) + .expect("state request should build"); + response_json( + app.clone() + .oneshot(req) + .await + .expect("state request routes"), + StatusCode::OK, + format!("GET /api/v1/runs/{run_id}/state"), + ) + .await + }; + assert!( + state_body["pending_interviews"] + .as_object() + .is_some_and(serde_json::Map::is_empty), + "the answered question is no longer pending: {}", + state_body["pending_interviews"] + ); +}