From f15972f8f2198a861499082338dd79f3167b37a9 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Fri, 11 Sep 2026 09:46:30 -0600 Subject: [PATCH] Keep Ctrl-C owned for the whole native Git command phase owned() polled tokio's ctrl_c() only for the duration of one Git acquisition. On Unix that listener permanently replaces the default SIGINT disposition, so once acquisition finished nothing handled Ctrl-C and fabro create, fabro run --detach, and fabro run before attach silently ignored it while waiting on the server. Introduce an Interruption handle that the command entry points create from the run arguments: it installs a listener only when --workflow-git or --target-git is in play, guards create (and start for fabro run) as one phase, and tracks owned Git tasks so interruption waits for their cleanup before returning. attach keeps installing its own listener. Co-Authored-By: Claude Fable 5.1 --- lib/apps/fabro-cli/Cargo.toml | 2 +- .../fabro-cli/src/commands/run/command.rs | 38 +++-- lib/apps/fabro-cli/src/commands/run/create.rs | 7 + lib/apps/fabro-cli/src/commands/run/mod.rs | 10 +- .../src/commands/run/remote_workflow.rs | 155 ++++++++++++++---- .../fabro-cli/src/commands/run/resolution.rs | 47 ++++-- 6 files changed, 200 insertions(+), 59 deletions(-) diff --git a/lib/apps/fabro-cli/Cargo.toml b/lib/apps/fabro-cli/Cargo.toml index a91a23eb2..e953d0a26 100644 --- a/lib/apps/fabro-cli/Cargo.toml +++ b/lib/apps/fabro-cli/Cargo.toml @@ -100,7 +100,7 @@ object_store.workspace = true bytes.workspace = true tokio-util.workspace = true libc = "0.2" -nix = { version = "0.30", features = ["fs"] } +nix = { version = "0.30", features = ["fs", "signal"] } [target.'cfg(target_os = "macos")'.dependencies] core-foundation = { version = "0.9", optional = true } diff --git a/lib/apps/fabro-cli/src/commands/run/command.rs b/lib/apps/fabro-cli/src/commands/run/command.rs index 2ec25f636..2dfdd235a 100644 --- a/lib/apps/fabro-cli/src/commands/run/command.rs +++ b/lib/apps/fabro-cli/src/commands/run/command.rs @@ -1,6 +1,7 @@ use anyhow::Result; use fabro_util::terminal::Styles; +use super::remote_workflow::Interruption; use crate::args::RunArgs; use crate::command_context::CommandContext; use crate::shared::print_json_pretty; @@ -15,16 +16,32 @@ pub(crate) async fn execute(mut args: RunArgs, base_ctx: &CommandContext) -> Res let quiet = args.detach; let prevent_idle_sleep = ctx.user_settings().cli.exec.prevent_idle_sleep; - let created_run = Box::pin(super::create::create_run(&ctx, &args, styles)).await?; + // Ctrl-C stays owned here through start; `attach` installs its own listener. + let interruption = Interruption::for_run_args(&args); + let (created_run, client) = interruption + .guard(async { + let created_run = Box::pin(super::create::create_run( + &ctx, + &args, + styles, + &interruption, + )) + .await?; - if !quiet { - fabro_util::printerr!( - printer, - " {} {}", - styles.dim.apply_to("Run:"), - styles.dim.apply_to(&created_run.run_id), - ); - } + if !quiet { + fabro_util::printerr!( + printer, + " {} {}", + styles.dim.apply_to("Run:"), + styles.dim.apply_to(&created_run.run_id), + ); + } + + let client = ctx.server().await?; + super::start::start_run_with_client(&client, &created_run.run_id, false).await?; + Ok((created_run, client)) + }) + .await?; #[cfg(feature = "sleep_inhibitor")] let _sleep_guard = sleep_inhibitor::guard(prevent_idle_sleep); @@ -32,9 +49,6 @@ pub(crate) async fn execute(mut args: RunArgs, base_ctx: &CommandContext) -> Res #[cfg(not(feature = "sleep_inhibitor"))] let _ = prevent_idle_sleep; - let client = ctx.server().await?; - super::start::start_run_with_client(&client, &created_run.run_id, false).await?; - let json = ctx.json_output(); if args.detach { if json { diff --git a/lib/apps/fabro-cli/src/commands/run/create.rs b/lib/apps/fabro-cli/src/commands/run/create.rs index c7145ef73..ed69e165c 100644 --- a/lib/apps/fabro-cli/src/commands/run/create.rs +++ b/lib/apps/fabro-cli/src/commands/run/create.rs @@ -7,6 +7,7 @@ use fabro_types::{RunId, RunIntent}; use fabro_util::terminal::Styles; use super::overrides::prepare_intent_overrides; +use super::remote_workflow::Interruption; use super::resolution::ResolvedWorkflow; use super::selection::WorkflowSelection; use super::{resolution, selection}; @@ -23,10 +24,14 @@ pub(crate) struct CreatedRun { /// run from an immutable workflow intent, leaving it in the submitted state. /// /// This does NOT start the workflow — starting is a separate request. +/// +/// Native Git acquisition runs under `interruption`; the caller guards this +/// call (and any later phase before `attach`) with the same handle. pub(crate) async fn create_run( ctx: &CommandContext, args: &RunArgs, styles: &Styles, + interruption: &Interruption, ) -> anyhow::Result { let (workflow_selection, target_selection) = selection::parse(args)?; let canonical_cwd = ctx.cwd().canonicalize().with_context(|| { @@ -41,6 +46,7 @@ pub(crate) async fn create_run( &workflow_selection, &canonical_cwd, Some(&user_workflows_root), + interruption, ) }; // Preserve local lookup diagnostics before contacting the server. Remote @@ -94,6 +100,7 @@ pub(crate) async fn create_run( &target_selection, environment.settings.provider, &canonical_cwd, + interruption, ) .await?; if dirty_worktree { diff --git a/lib/apps/fabro-cli/src/commands/run/mod.rs b/lib/apps/fabro-cli/src/commands/run/mod.rs index 6b9177673..b904549b4 100644 --- a/lib/apps/fabro-cli/src/commands/run/mod.rs +++ b/lib/apps/fabro-cli/src/commands/run/mod.rs @@ -47,7 +47,15 @@ pub(crate) async fn dispatch( RunCommands::Create(args) => { let styles: &'static Styles = Box::leak(Box::new(Styles::detect_stderr())); let ctx = base_ctx.with_target(&args.target)?; - let created_run = Box::pin(create::create_run(&ctx, &args, styles)).await?; + let interruption = remote_workflow::Interruption::for_run_args(&args); + let created_run = interruption + .guard(Box::pin(create::create_run( + &ctx, + &args, + styles, + &interruption, + ))) + .await?; if ctx.json_output() { print_json_pretty(&serde_json::json!({ "run_id": created_run.run_id }))?; } else { diff --git a/lib/apps/fabro-cli/src/commands/run/remote_workflow.rs b/lib/apps/fabro-cli/src/commands/run/remote_workflow.rs index 6a7944c92..015353490 100644 --- a/lib/apps/fabro-cli/src/commands/run/remote_workflow.rs +++ b/lib/apps/fabro-cli/src/commands/run/remote_workflow.rs @@ -11,8 +11,10 @@ use tokio::io::{AsyncRead, AsyncReadExt as _}; use tokio::process::Command; use tokio::{fs, signal as tokio_signal, task, time}; use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; use super::selection::RemoteWorkflowRevision; +use crate::args::RunArgs; const OUTPUT_LIMIT: usize = 64 * 1024; @@ -457,27 +459,76 @@ fn resolve_name(records: &str, branch: &str, tag: &str) -> anyhow::Result( - work: impl FnOnce(CancellationToken) -> TFuture + Send + 'static, -) -> anyhow::Result -where - TFuture: Future> + Send + 'static, -{ - let cancel = CancellationToken::new(); - let _cancel_on_drop = cancel.clone().drop_guard(); - let mut task = tokio::spawn(work(cancel.clone())); - tokio::select! { - result = &mut task => result.context("local Git task failed")?, - signal = tokio_signal::ctrl_c() => { - cancel.cancel(); - let _result = task.await.context("local Git cleanup task failed")?; - signal.context("failed to listen for interruption")?; - Err(RemoteWorkflowError::Cancelled.into()) +/// Cooperative Ctrl-C handling for commands that acquire sources with native +/// Git. +/// +/// Tokio's Ctrl-C listener permanently replaces the default SIGINT disposition +/// for the process, so it is installed only when native Git is in play, and the +/// command keeps it armed for every phase up to the point where `attach` +/// installs its own listener or the process exits. Interruption cancels owned +/// Git tasks and waits for their cleanup; an in-progress blocking collection +/// must finish first. +#[derive(Clone)] +pub(crate) struct Interruption { + cancel: CancellationToken, + tasks: TaskTracker, + listens: bool, +} + +impl Interruption { + /// `native_git` reports whether any selection runs native Git. Without it + /// the default SIGINT disposition is left untouched and `guard` is a + /// pass-through. + pub(crate) fn new(native_git: bool) -> Self { + Self { + cancel: CancellationToken::new(), + tasks: TaskTracker::new(), + listens: native_git, } } + + pub(crate) fn for_run_args(args: &RunArgs) -> Self { + Self::new(args.workflow_git.is_some() || args.target_git.is_some()) + } + + /// Run `work` to completion, or until Ctrl-C cancels it and every owned + /// Git task has cleaned up. + pub(crate) async fn guard( + &self, + work: impl Future>, + ) -> anyhow::Result { + if !self.listens { + return work.await; + } + tokio::select! { + result = work => result, + signal = tokio_signal::ctrl_c() => { + self.cancel.cancel(); + self.tasks.close(); + self.tasks.wait().await; + signal.context("failed to listen for interruption")?; + Err(RemoteWorkflowError::Cancelled.into()) + } + } + } + + /// The task owns its child processes and temporary checkout. Dropping the + /// waiter requests cooperative cleanup, not task abortion; the task keeps + /// running until its Git children are reaped and its files are removed. + pub(super) async fn owned( + &self, + work: impl FnOnce(CancellationToken) -> TFuture + Send + 'static, + ) -> anyhow::Result + where + TFuture: Future> + Send + 'static, + { + let cancel = self.cancel.child_token(); + let _cancel_on_drop = cancel.clone().drop_guard(); + self.tasks + .spawn(work(cancel.clone())) + .await + .context("local Git task failed")? + } } #[cfg(test)] @@ -487,7 +538,10 @@ where )] mod tests { use std::os::unix::fs::PermissionsExt as _; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use nix::sys::signal::{self, Signal}; use nix::sys::stat::Mode; use nix::unistd; @@ -706,16 +760,20 @@ mod tests { let (fake, git) = fake_git("printf '%s' $$ > pid; exec /bin/sleep 60"); let checkout = tempfile::tempdir().unwrap(); let path = checkout.path().to_path_buf(); - let worker = tokio::spawn(owned(move |cancel| async move { - git.collect_checkout( - "acme/workflows".parse().unwrap(), - "review".into(), - "1111111111111111111111111111111111111111".into(), - checkout, - cancel, - ) - .await - })); + let worker = tokio::spawn(async move { + Interruption::new(true) + .owned(move |cancel| async move { + git.collect_checkout( + "acme/workflows".parse().unwrap(), + "review".into(), + "1111111111111111111111111111111111111111".into(), + checkout, + cancel, + ) + .await + }) + .await + }); time::timeout(Duration::from_secs(5), async { while !path.join("pid").exists() { time::sleep(Duration::from_millis(10)).await; @@ -740,6 +798,45 @@ mod tests { drop(fake); } + /// Each nextest test runs in its own process, so raising SIGINT here only + /// reaches the listener `guard` installed before polling `work`. + #[tokio::test] + async fn remote_workflow_guard_cancels_owned_tasks_and_waits_for_cleanup_on_ctrl_c() { + let interruption = Interruption::new(true); + let cleaned = Arc::new(AtomicBool::new(false)); + let result: anyhow::Result<()> = interruption + .guard({ + let interruption = interruption.clone(); + let cleaned = Arc::clone(&cleaned); + async move { + interruption + .owned(move |cancel| async move { + signal::raise(Signal::SIGINT).unwrap(); + cancel.cancelled().await; + // Cleanup after cancellation must finish before + // `guard` reports the interruption. + time::sleep(Duration::from_millis(200)).await; + cleaned.store(true, Ordering::SeqCst); + Ok(()) + }) + .await + } + }) + .await; + assert!(matches!( + result.unwrap_err().downcast_ref::(), + Some(RemoteWorkflowError::Cancelled) + )); + assert!(cleaned.load(Ordering::SeqCst)); + assert_eq!( + Interruption::new(false) + .guard(async { Ok::<_, anyhow::Error>(7) }) + .await + .unwrap(), + 7 + ); + } + #[tokio::test] async fn remote_workflow_resolves_exact_default_branches_tags_and_commits() { let fixture = Fixture::new(); diff --git a/lib/apps/fabro-cli/src/commands/run/resolution.rs b/lib/apps/fabro-cli/src/commands/run/resolution.rs index a6fa1daa3..556e92d1d 100644 --- a/lib/apps/fabro-cli/src/commands/run/resolution.rs +++ b/lib/apps/fabro-cli/src/commands/run/resolution.rs @@ -6,7 +6,7 @@ use fabro_types::settings::run::EnvironmentProvider; use fabro_types::{DirtyStatus, RunTarget}; use tokio::task; -use super::remote_workflow::{self, NativeGit}; +use super::remote_workflow::{Interruption, NativeGit}; use super::selection::{TargetSelection, WorkflowSelection}; /// Owns the canonical collector result without copying its contents. Local @@ -29,6 +29,7 @@ pub(super) async fn workflow( selection: &WorkflowSelection, cwd: &Path, user_workflows: Option<&Path>, + interruption: &Interruption, ) -> anyhow::Result { match selection { WorkflowSelection::Local(path) => { @@ -57,10 +58,11 @@ pub(super) async fn workflow( let git = NativeGit::new(); let (repository, selector, revision) = (repository.clone(), selector.clone(), revision.clone()); - let closure = remote_workflow::owned(move |cancel| async move { - git.collect(repository, selector, revision, cancel).await - }) - .await?; + let closure = interruption + .owned(move |cancel| async move { + git.collect(repository, selector, revision, cancel).await + }) + .await?; Ok(ResolvedWorkflow::Git(closure)) } } @@ -70,6 +72,7 @@ pub(super) async fn target( selection: &TargetSelection, provider: EnvironmentProvider, cwd: &Path, + interruption: &Interruption, ) -> anyhow::Result<(RunTarget, bool)> { let path = match selection { TargetSelection::Path(path) => cwd @@ -82,10 +85,11 @@ pub(super) async fn target( } let git = NativeGit::new(); let (repository, branch) = (repository.clone(), branch.clone()); - let target = remote_workflow::owned(move |cancel| async move { - git.resolve_target(repository, branch, &cancel).await - }) - .await?; + let target = interruption + .owned(move |cancel| async move { + git.resolve_target(repository, branch, &cancel).await + }) + .await?; // Canonical admission retains ownership of provider capabilities. return Ok((RunTarget::Git(target), false)); } @@ -199,8 +203,9 @@ mod tests { write_workflow(&root, ".fabro/workflows/review"); std::fs::create_dir(root.join("target")).unwrap(); let selected = TargetSelection::Path("target".into()); + let interruption = Interruption::new(false); assert_eq!( - target(&selected, EnvironmentProvider::Local, &root) + target(&selected, EnvironmentProvider::Local, &root, &interruption) .await .unwrap() .0, @@ -210,7 +215,10 @@ mod tests { ); for provider in [EnvironmentProvider::Docker, EnvironmentProvider::Daytona] { assert_eq!( - target(&selected, provider, &root).await.unwrap().0, + target(&selected, provider, &root, &interruption) + .await + .unwrap() + .0, RunTarget::None {} ); } @@ -218,7 +226,8 @@ mod tests { target( &TargetSelection::Path(".".into()), EnvironmentProvider::Local, - &root + &root, + &interruption ) .await .unwrap() @@ -234,7 +243,8 @@ mod tests { branch: None, }, EnvironmentProvider::Local, - &root + &root, + &interruption ) .await .unwrap_err() @@ -246,7 +256,8 @@ mod tests { target( &TargetSelection::Path(path.into()), EnvironmentProvider::Local, - &root + &root, + &interruption ) .await .is_err(), @@ -267,13 +278,16 @@ mod tests { std::fs::write(project.join(".fabro/project.toml"), "_version = 1\n").unwrap(); git2::Repository::init(&checkout).unwrap(); let selected = WorkflowSelection::Local("review".into()); + let interruption = Interruption::new(false); for (cwd, expected_root) in [ (checkout.as_path(), checkout.as_path()), (project.as_path(), project.as_path()), (root.path(), user.as_path()), ] { let ResolvedWorkflow::Local(package) = - workflow(&selected, cwd, Some(&user)).await.unwrap() + workflow(&selected, cwd, Some(&user), &interruption) + .await + .unwrap() else { panic!("local package"); }; @@ -283,7 +297,8 @@ mod tests { workflow( &WorkflowSelection::Local("missing.toml".into()), &checkout, - Some(&user) + Some(&user), + &interruption ) .await .is_err()