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.

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();
+    }
 }

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<R>(reader: R, interviewer: Arc<ControlInterviewer>)
-where
+async fn read_worker_control_stream<R>(
+    reader: R,
+    interviewer: Arc<ControlInterviewer>,
+    cancel_token: Arc<AtomicBool>,
+) 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));
     }
 }

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);

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"]);
 }

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]

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);
-}

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)

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);
+    }
 }

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<AppState>, 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,