From 57ec693a903987314fd5447b428951bfd7d1867e Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Wed, 8 Apr 2026 03:47:35 -0400 Subject: [PATCH] test(cli): replace slow active-run coverage Replace the slow CLI integration tests that waited on worker shutdown grace periods with focused coverage that still checks the important behavior. The attach JSON test now finishes the gated run cleanly, the rm force test uses a mocked server contract, and the Ctrl-C cancel path is covered at the attach layer instead of through a full live run. Add a cooperative subprocess cancel control message so cancel and delete can abort pending interviews without relying only on the 5 second hard kill fallback. --- .../fabro-cli/src/commands/run/attach.rs | 65 ++++++++++++++++ .../fabro-cli/src/commands/run/runner.rs | 57 ++++++++++++-- lib/crates/fabro-cli/src/server_client.rs | 5 +- lib/crates/fabro-cli/tests/it/cmd/attach.rs | 78 ++++++++++++++++++- lib/crates/fabro-cli/tests/it/cmd/rm.rs | 67 ++++++++-------- lib/crates/fabro-cli/tests/it/cmd/run.rs | 69 +--------------- lib/crates/fabro-cli/tests/it/cmd/support.rs | 23 ------ .../fabro-interview/src/control_protocol.rs | 20 +++++ lib/crates/fabro-server/src/server.rs | 55 ++++++++++++- 9 files changed, 305 insertions(+), 134 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index 6a53cfa5b..9ebb1eb9c 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -449,11 +449,47 @@ mod tests { use super::*; use fabro_interview::{Answer, AnswerValue}; use fabro_util::terminal::Styles; + use httpmock::MockServer; fn no_color_styles() -> &'static Styles { Box::leak(Box::new(Styles::new(false))) } + fn terminal_run_state_response() -> serde_json::Value { + serde_json::json!({ + "run": null, + "graph_source": null, + "start": null, + "status": { + "status": "failed", + "reason": "cancelled", + "updated_at": "2026-04-05T12:00:02Z" + }, + "checkpoint": null, + "checkpoints": [], + "conclusion": null, + "retro": null, + "retro_prompt": null, + "retro_response": null, + "sandbox": null, + "final_patch": null, + "pull_request": null, + "nodes": {} + }) + } + + fn cancel_run_response(run_id: RunId) -> serde_json::Value { + serde_json::json!({ + "id": run_id, + "status": "cancelled", + "error": null, + "queue_position": null, + "status_reason": "cancelled", + "pending_control": "cancel", + "created_at": "2026-04-05T12:00:00Z" + }) + } + #[tokio::test] async fn attach_errors_without_store_context() { let dir = tempfile::tempdir().unwrap(); @@ -525,4 +561,33 @@ mod tests { fn json_pending_interview_does_not_require_manual_input_when_auto_approve_is_enabled() { assert!(!json_pending_interview_requires_manual_input(true, true)); } + + #[tokio::test] + async fn handle_detach_signal_with_kill_on_detach_cancels_active_run_via_server() { + let run_id = fabro_types::fixtures::RUN_1; + let server = MockServer::start(); + let cancel_mock = server.mock(|when, then| { + when.method("POST") + .path(format!("/api/v1/runs/{run_id}/cancel")); + then.status(200) + .header("Content-Type", "application/json") + .body(cancel_run_response(run_id).to_string()); + }); + let state_mock = server.mock(|when, then| { + when.method("GET") + .path(format!("/api/v1/runs/{run_id}/state")); + then.status(200) + .header("Content-Type", "application/json") + .body(terminal_run_state_response().to_string()); + }); + let client = + server_client::connect_server_target_direct(&format!("{}/api/v1", server.base_url())) + .await + .unwrap(); + + handle_detach_signal(&client, &run_id, true).await; + + cancel_mock.assert(); + state_mock.assert(); + } } diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index 5321062ad..467670e80 100644 --- a/lib/crates/fabro-cli/src/commands/run/runner.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -66,12 +66,13 @@ pub(crate) async fn execute( let artifact_uploader = build_artifact_uploader(run_id, client.clone_for_reuse(), artifact_upload_token); let interviewer = Arc::new(ControlInterviewer::new()); + let cancel_token = Arc::new(AtomicBool::new(false)); tokio::spawn(read_worker_control_stream( io::stdin(), Arc::clone(&interviewer), + Arc::clone(&cancel_token), )); let run_control = RunControlState::new(); - let cancel_token = Arc::new(AtomicBool::new(false)); install_signal_handlers(Arc::clone(&run_control), Arc::clone(&cancel_token))?; let github_app = maybe_build_github_app_credentials(&run_record.settings)?; let services = StartServices { @@ -106,15 +107,18 @@ pub(crate) async fn execute( Ok(()) } -async fn read_worker_control_stream(reader: R, interviewer: Arc) -where +async fn read_worker_control_stream( + reader: R, + interviewer: Arc, + cancel_token: Arc, +) where R: AsyncRead + Unpin, { let mut lines = BufReader::new(reader).lines(); loop { match lines.next_line().await { Ok(Some(line)) => { - apply_worker_control_line(&interviewer, &line).await; + apply_worker_control_line(&interviewer, &cancel_token, &line).await; } Ok(None) | Err(_) => { interviewer.abort_all().await; @@ -124,7 +128,11 @@ where } } -async fn apply_worker_control_line(interviewer: &ControlInterviewer, line: &str) { +async fn apply_worker_control_line( + interviewer: &ControlInterviewer, + cancel_token: &AtomicBool, + line: &str, +) { if line.trim().is_empty() { return; } @@ -137,6 +145,10 @@ async fn apply_worker_control_line(interviewer: &ControlInterviewer, line: &str) WorkerControlMessage::InterviewAnswer { qid, answer } => { let _ = interviewer.submit(&qid, answer.into()).await; } + WorkerControlMessage::RunCancel => { + cancel_token.store(true, Ordering::SeqCst); + interviewer.abort_all().await; + } } } @@ -470,6 +482,7 @@ fn install_signal_handlers( #[cfg(test)] mod tests { use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; use httpmock::MockServer; use serde_json::json; @@ -680,6 +693,7 @@ mod tests { #[tokio::test] async fn worker_control_line_routes_answer_by_question_id() { let interviewer = Arc::new(ControlInterviewer::new()); + let cancel_token = Arc::new(AtomicBool::new(false)); let mut question = Question::new("Approve?", QuestionType::YesNo); question.id = "q-1".to_string(); let ask_interviewer = Arc::clone(&interviewer); @@ -687,25 +701,56 @@ mod tests { apply_worker_control_line( &interviewer, + &cancel_token, r#"{"v":1,"type":"interview.answer","qid":"q-1","answer":{"kind":"yes"}}"#, ) .await; let answer: fabro_interview::Answer = answer_task.await.unwrap(); assert_eq!(answer.value, AnswerValue::Yes); + assert!(!cancel_token.load(Ordering::SeqCst)); + } + + #[tokio::test] + async fn worker_control_line_cancel_sets_cancel_token_and_aborts_pending_interviews() { + let interviewer = Arc::new(ControlInterviewer::new()); + let cancel_token = Arc::new(AtomicBool::new(false)); + let mut question = Question::new("Approve?", QuestionType::YesNo); + question.id = "q-1".to_string(); + let ask_interviewer = Arc::clone(&interviewer); + let answer_task = tokio::spawn(async move { ask_interviewer.ask(question).await }); + tokio::task::yield_now().await; + + apply_worker_control_line( + &interviewer, + &cancel_token, + r#"{"v":1,"type":"run.cancel"}"#, + ) + .await; + + let answer: fabro_interview::Answer = answer_task.await.unwrap(); + assert_eq!(answer.value, AnswerValue::Aborted); + assert!(cancel_token.load(Ordering::SeqCst)); } #[tokio::test] async fn worker_control_stream_eof_aborts_pending_interviews() { let interviewer = Arc::new(ControlInterviewer::new()); + let cancel_token = Arc::new(AtomicBool::new(false)); let mut question = Question::new("Approve?", QuestionType::YesNo); question.id = "q-1".to_string(); let ask_interviewer = Arc::clone(&interviewer); let answer_task = tokio::spawn(async move { ask_interviewer.ask(question).await }); - read_worker_control_stream(tokio::io::empty(), Arc::clone(&interviewer)).await; + read_worker_control_stream( + tokio::io::empty(), + Arc::clone(&interviewer), + Arc::clone(&cancel_token), + ) + .await; let answer: fabro_interview::Answer = answer_task.await.unwrap(); assert_eq!(answer.value, AnswerValue::Aborted); + assert!(!cancel_token.load(Ordering::SeqCst)); } } diff --git a/lib/crates/fabro-cli/src/server_client.rs b/lib/crates/fabro-cli/src/server_client.rs index 1fe3f6549..68d194ff2 100644 --- a/lib/crates/fabro-cli/src/server_client.rs +++ b/lib/crates/fabro-cli/src/server_client.rs @@ -972,7 +972,10 @@ mod tests { .await; let client = test_client(&server.url("/api/v1")); - let events = client.list_run_events(&run_id, None, Some(1)).await.unwrap(); + let events = client + .list_run_events(&run_id, None, Some(1)) + .await + .unwrap(); first_page.assert_async().await; assert_eq!(events.len(), 1); diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index e7f303382..4233ba8d3 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -1,4 +1,5 @@ use std::io::{BufRead, BufReader, Read}; +use std::path::Path; use std::process::{Output, Stdio}; use std::sync::mpsc; use std::time::{Duration, Instant}; @@ -8,10 +9,63 @@ use serde_json::Value; use crate::support::{example_fixture, fabro_json_snapshot, run_output_filters, unique_run_id}; -use super::support::{output_stdout, resolve_run, wait_for_status, write_gated_workflow}; +use super::support::{ + output_stdout, resolve_run, server_target, wait_for_status, write_gated_workflow, +}; const SHARED_DAEMON_TIMEOUT: Duration = Duration::from_secs(30); +fn server_endpoint(storage_dir: &Path) -> (reqwest::Client, String) { + let target = server_target(storage_dir); + if target.starts_with('/') { + ( + reqwest::ClientBuilder::new() + .unix_socket(target) + .no_proxy() + .build() + .expect("test Unix-socket HTTP client should build"), + "http://fabro".to_string(), + ) + } else { + ( + reqwest::ClientBuilder::new() + .no_proxy() + .build() + .expect("test TCP HTTP client should build"), + target, + ) + } +} + +async fn wait_for_server_question(client: &reqwest::Client, base_url: &str, run_id: &str) -> Value { + let deadline = std::time::Instant::now() + SHARED_DAEMON_TIMEOUT; + loop { + let response = client + .get(format!("{base_url}/api/v1/runs/{run_id}/questions")) + .query(&[("page[limit]", "100"), ("page[offset]", "0")]) + .send() + .await + .expect("question request should succeed"); + assert!( + response.status().is_success(), + "question request failed: {}", + response.status() + ); + let body: Value = response + .json() + .await + .expect("question response should parse"); + if let Some(question) = body["data"].as_array().and_then(|items| items.first()) { + return question.clone(); + } + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for a pending question" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } +} + fn format_output_snapshot(output: &Output, filters: &[(String, String)]) -> String { let stdout = apply_filters(&String::from_utf8_lossy(&output.stdout), filters); let stderr = apply_filters(&String::from_utf8_lossy(&output.stderr), filters); @@ -713,4 +767,26 @@ fn attach_json_errors_without_prompting_for_human_input() { } ] "#); + + let run = resolve_run(&context, &run_id); + tokio::runtime::Runtime::new() + .expect("test runtime should build") + .block_on(async { + let (client, base_url) = server_endpoint(&context.storage_dir); + let question = wait_for_server_question(&client, &base_url, &run_id).await; + let question_id = question["id"] + .as_str() + .expect("question id should be present"); + + let response = client + .post(format!( + "{base_url}/api/v1/runs/{run_id}/questions/{question_id}/answer" + )) + .json(&serde_json::json!({ "selected_option_key": "A" })) + .send() + .await + .expect("answer submission should succeed"); + assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT); + }); + wait_for_status(&run.run_dir, &["succeeded"]); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/rm.rs b/lib/crates/fabro-cli/tests/it/cmd/rm.rs index 3452baef3..7930e8506 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/rm.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/rm.rs @@ -5,8 +5,7 @@ use serde_json::Value; use crate::support::unique_run_id; use super::support::{ - output_stdout, resolve_run, setup_completed_fast_dry_run, setup_created_fast_dry_run, - setup_local_sandbox_run, wait_for_no_process_match, wait_for_status, write_gated_workflow, + setup_completed_fast_dry_run, setup_created_fast_dry_run, setup_local_sandbox_run, }; #[test] @@ -150,35 +149,42 @@ fn rm_force_deletes_run_without_sandbox_json_when_store_has_sandbox() { } #[test] -fn rm_force_terminates_active_run_worker() { +fn rm_force_removes_active_run() { let context = test_context!(); - let _gate = write_gated_workflow(&context.temp_dir.join("slow.fabro"), "slow", "Run slowly"); - - let output = context - .run_cmd() - .env("OPENAI_API_KEY", "test") - .args([ - "--detach", - "--provider", - "openai", - "--sandbox", - "local", - "--no-retro", - "slow.fabro", - ]) - .output() - .expect("run --detach should execute"); - assert!( - output.status.success(), - "run --detach failed:\nstdout:\n{}\nstderr:\n{}", - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) + let run_id = unique_run_id(); + let server = MockServer::start(); + let list_mock = server.mock(|when, then| { + when.method("GET").path("/api/v1/runs"); + then.status(200) + .header("Content-Type", "application/json") + .body( + serde_json::json!([ + { + "run_id": run_id, + "workflow_name": "Active Workflow", + "workflow_slug": "active-workflow", + "goal": "Active goal", + "labels": {}, + "host_repo_path": null, + "start_time": "2026-04-05T12:00:00Z", + "status": "running", + "status_reason": null, + "duration_ms": 123, + "total_usd_micros": null + } + ]) + .to_string(), + ); + }); + let delete_mock = server.mock(|when, then| { + when.method("DELETE").path(format!("/api/v1/runs/{run_id}")); + then.status(204); + }); + context.write_home( + ".fabro/settings.toml", + format!("[server]\ntarget = \"{}/api/v1\"\n", server.base_url()), ); - let run_id = output_stdout(&output).trim().to_string(); - let run = resolve_run(&context, &run_id); - wait_for_status(&run.run_dir, &["running"]); - let mut filters = context.filters(); filters.push(( r"\b[0-9A-HJKMNP-TV-Z]{12}\b".to_string(), @@ -193,9 +199,8 @@ fn rm_force_terminates_active_run_worker() { ----- stderr ----- [ULID] "); - - assert!(!run.run_dir.exists(), "run directory should be deleted"); - wait_for_no_process_match(&format!("fabro {} ", &run_id[..12])); + list_mock.assert(); + delete_mock.assert(); } #[test] diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index d250b5d6a..bcfb93b66 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -1,12 +1,8 @@ use fabro_test::{fabro_snapshot, test_context}; -use fabro_types::StatusReason; use httpmock::MockServer; use serde_json::Value; -use super::support::{ - output_stderr, resolve_run, run_state, wait_for_event_names, wait_for_no_process_match, - wait_for_status, write_gated_workflow, -}; +use super::support::{output_stderr, wait_for_event_names}; use crate::support::{example_fixture, fabro_json_snapshot, run_output_filters, unique_run_id}; fn run_status_response(run_id: &str, status: &str) -> serde_json::Value { @@ -1395,66 +1391,3 @@ fn detach_creates_run_dir_with_detach_log() { "# ); } - -#[test] -fn ctrl_c_cancels_active_run_via_server() { - let context = test_context!(); - let gate = write_gated_workflow(&context.temp_dir.join("slow.fabro"), "slow", "Run slowly"); - let run_id = unique_run_id(); - - let mut run_cmd = std::process::Command::new(env!("CARGO_BIN_EXE_fabro")); - run_cmd.current_dir(&context.temp_dir); - run_cmd.env("NO_COLOR", "1"); - run_cmd.env("HOME", &context.home_dir); - run_cmd.env("FABRO_NO_UPGRADE_CHECK", "true"); - run_cmd.env("FABRO_TEST_IN_MEMORY_STORE", "1"); - run_cmd.env("FABRO_STORAGE_DIR", &context.storage_dir); - run_cmd.env("FABRO_SERVER_MAX_CONCURRENT_RUNS", "64"); - run_cmd.env("OPENAI_API_KEY", "test"); - run_cmd.args([ - "run", - "--run-id", - run_id.as_str(), - "--label", - &context.test_run_label(), - "--label", - &context.test_case_label(), - "--provider", - "openai", - "--sandbox", - "local", - "--no-retro", - "slow.fabro", - ]); - let child = run_cmd.spawn().expect("run should spawn"); - - let run = resolve_run(&context, &run_id); - wait_for_status(&run.run_dir, &["running"]); - - let kill_status = std::process::Command::new("kill") - .args(["-INT", &child.id().to_string()]) - .status() - .expect("kill should execute"); - assert!(kill_status.success(), "kill -INT should succeed"); - - let output = child - .wait_with_output() - .expect("run should exit after SIGINT"); - assert!( - !output.status.success(), - "run should exit non-zero after cancellation\nstdout:\n{}\nstderr:\n{}", - String::from_utf8_lossy(&output.stdout), - output_stderr(&output) - ); - - let final_status = wait_for_status(&run.run_dir, &["failed"]); - assert_eq!(final_status, "failed"); - assert_eq!( - run_state(&run.run_dir) - .status - .and_then(|record| record.reason), - Some(StatusReason::Cancelled) - ); - let gate_pattern = gate.gate_path().to_string_lossy().into_owned(); - wait_for_no_process_match(&gate_pattern); -} diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index 922b4846e..94f5379ca 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -281,34 +281,11 @@ pub(crate) fn setup_project_fixture(context: &TestContext) -> ProjectFixture { } impl WorkflowGate { - pub(crate) fn gate_path(&self) -> &Path { - &self.gate_path - } - pub(crate) fn release(&self) { write_text_file(&self.gate_path, "open\n"); } } -pub(crate) fn wait_for_no_process_match(pattern: &str) { - let deadline = Instant::now() + COMMAND_TIMEOUT; - loop { - let output = std::process::Command::new("pgrep") - .args(["-f", pattern]) - .output() - .expect("pgrep should execute"); - if !output.status.success() { - return; - } - assert!( - Instant::now() < deadline, - "timed out waiting for processes matching {pattern:?} to exit: {}", - String::from_utf8_lossy(&output.stdout) - ); - std::thread::sleep(Duration::from_millis(50)); - } -} - pub(crate) fn setup_artifact_run(context: &TestContext) -> WorkspaceRunSetup { let workspace_dir = context.temp_dir.join("artifact-run"); std::fs::create_dir_all(&workspace_dir) diff --git a/lib/crates/fabro-interview/src/control_protocol.rs b/lib/crates/fabro-interview/src/control_protocol.rs index 5fa8105b5..db7511844 100644 --- a/lib/crates/fabro-interview/src/control_protocol.rs +++ b/lib/crates/fabro-interview/src/control_protocol.rs @@ -22,6 +22,14 @@ impl WorkerControlEnvelope { }, } } + + #[must_use] + pub fn cancel_run() -> Self { + Self { + v: WORKER_CONTROL_PROTOCOL_VERSION, + message: WorkerControlMessage::RunCancel, + } + } } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -32,6 +40,8 @@ pub enum WorkerControlMessage { qid: String, answer: WorkerControlAnswer, }, + #[serde(rename = "run.cancel")] + RunCancel, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -97,4 +107,14 @@ mod tests { let parsed: WorkerControlEnvelope = serde_json::from_str(&json).unwrap(); assert_eq!(parsed, envelope); } + + #[test] + fn cancel_run_round_trips_through_json() { + let envelope = WorkerControlEnvelope::cancel_run(); + let json = serde_json::to_string(&envelope).unwrap(); + assert_eq!(json, r#"{"v":1,"type":"run.cancel"}"#); + + let parsed: WorkerControlEnvelope = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed, envelope); + } } diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 1a731dcf0..ad0d0ffb4 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -342,9 +342,19 @@ impl RunAnswerTransport { } } - async fn abort_pending(&self) { - if let Self::InProcess { interviewer } = self { - interviewer.abort_all().await; + async fn cancel_run(&self) -> Result<(), AnswerTransportError> { + match self { + Self::Subprocess { control_tx } => { + let message = WorkerControlEnvelope::cancel_run(); + timeout(WORKER_CONTROL_ENQUEUE_TIMEOUT, control_tx.send(message)) + .await + .map_err(|_| AnswerTransportError::Timeout)? + .map_err(|_| AnswerTransportError::Closed) + } + Self::InProcess { interviewer } => { + interviewer.abort_all().await; + Ok(()) + } } } } @@ -2113,7 +2123,7 @@ async fn delete_run_internal(state: &Arc, id: RunId) -> Result<(), Res token.store(true, Ordering::SeqCst); } if let Some(answer_transport) = managed_run.answer_transport.clone() { - answer_transport.abort_pending().await; + let _ = answer_transport.cancel_run().await; } if let Some(cancel_tx) = managed_run.cancel_tx.take() { let _ = cancel_tx.send(()); @@ -5121,6 +5131,7 @@ async fn cancel_run( created_at, response_status, persist_cancelled_status, + answer_transport, cancel_token, cancel_tx, worker_pid, @@ -5145,6 +5156,7 @@ async fn cancel_run( managed_run.created_at, response_status, persist_cancelled_status, + managed_run.answer_transport.clone(), managed_run.cancel_token.clone(), managed_run.cancel_tx.take(), managed_run.worker_pid, @@ -5173,6 +5185,9 @@ async fn cancel_run( if let Some(cancel_tx) = cancel_tx { let _ = cancel_tx.send(()); } + if let Some(answer_transport) = answer_transport { + let _ = answer_transport.cancel_run().await; + } if let Some(worker_pid) = worker_pid { #[cfg(unix)] fabro_proc::sigterm(worker_pid); @@ -5771,6 +5786,7 @@ mod tests { use fabro_config::server::{ AuthProvider, AuthSettings, GitAuthorSettings, GitProvider, GitSettings, WebSettings, }; + use fabro_interview::{AnswerValue, ControlInterviewer, Interviewer, Question, QuestionType}; use fabro_types::{InterviewQuestionRecord, InterviewQuestionType, RunBlobId, RunId, fixtures}; #[cfg(unix)] use std::process::Stdio; @@ -5809,6 +5825,37 @@ mod tests { format!("/api/v1{path}") } + #[tokio::test] + async fn subprocess_answer_transport_cancel_run_enqueues_cancel_message() { + let (control_tx, mut control_rx) = tokio::sync::mpsc::channel(1); + let transport = RunAnswerTransport::Subprocess { control_tx }; + + transport.cancel_run().await.unwrap(); + + assert_eq!( + control_rx.recv().await, + Some(WorkerControlEnvelope::cancel_run()) + ); + } + + #[tokio::test] + async fn in_process_answer_transport_cancel_run_aborts_pending_interviews() { + let interviewer = Arc::new(ControlInterviewer::new()); + let transport = RunAnswerTransport::InProcess { + interviewer: Arc::clone(&interviewer), + }; + let mut question = Question::new("Approve?", QuestionType::YesNo); + question.id = "q-1".to_string(); + let ask_interviewer = Arc::clone(&interviewer); + let answer_task = tokio::spawn(async move { ask_interviewer.ask(question).await }); + tokio::task::yield_now().await; + + transport.cancel_run().await.unwrap(); + + let answer = answer_task.await.unwrap(); + assert_eq!(answer.value, AnswerValue::Aborted); + } + fn minimal_manifest_json(dot_source: &str) -> serde_json::Value { serde_json::json!({ "version": 1,