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,