From e2f884df23ed2d5607e4d5077e7c779cf3dbe599 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 21 Mar 2026 13:07:58 -0400 Subject: [PATCH] Fix resume command bugs, remove --ssh flag, simplify EventEmitter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix three bugs from PR review in the new `fabro resume` command: 1. RunArgs.workflow lost its required constraint when --run-branch was removed — add #[arg(required = true)] so clap enforces it. 2. run_resumed ignored --preserve-sandbox (hardcoded false), --verbose (no ProgressUI), and --ssh (no listener). Wire preserve_sandbox through resolve_preserve_sandbox, create a ProgressUI registered on the emitter, and handle SSH access. 3. prepare_from_checkpoint unconditionally created a LocalSandbox, ignoring --sandbox. Add the same sandbox resolution logic used by prepare_from_branch (Local, Docker, Ssh, Exe, Daytona). Remove --ssh from `fabro run` and `fabro resume` since `fabro ssh` is the dedicated command now. Remove the flag from RunArgs, ResumeArgs, RunSpec, and all docs/changelogs. Simplify EventEmitter: change on_event to take &self (via Mutex>>) instead of &mut self, removing the need for the late_listeners workaround. emit() snapshots the listener list before dispatching to prevent deadlocks from reentrant emit calls. Co-Authored-By: Claude Opus 4.6 (1M context) --- docs/changelog/2026-03-04.mdx | 6 +- docs/changelog/2026-03-09.mdx | 2 +- docs/execution/environments.mdx | 6 +- docs/human-tools/ssh-access.mdx | 25 +----- docs/human-tools/vs-code.mdx | 13 ++- docs/integrations/daytona.mdx | 10 +-- docs/integrations/exe-dev.mdx | 2 +- docs/reference/cli.mdx | 2 - lib/crates/fabro-api/src/server.rs | 2 +- lib/crates/fabro-cli/src/commands/create.rs | 1 - lib/crates/fabro-cli/src/commands/resume.rs | 86 +++++++++++++++++-- lib/crates/fabro-cli/src/commands/run.rs | 37 +------- .../fabro-cli/src/commands/run_progress.rs | 4 +- lib/crates/fabro-cli/src/main.rs | 1 - lib/crates/fabro-cli/tests/cli.rs | 4 +- .../src/devcontainer_bridge.rs | 6 +- lib/crates/fabro-workflows/src/engine.rs | 12 +-- lib/crates/fabro-workflows/src/event.rs | 33 ++++--- lib/crates/fabro-workflows/src/run_spec.rs | 2 - .../tests/daytona_integration.rs | 4 +- .../fabro-workflows/tests/integration.rs | 44 +++++----- 21 files changed, 160 insertions(+), 142 deletions(-) diff --git a/docs/changelog/2026-03-04.mdx b/docs/changelog/2026-03-04.mdx index bcf148209..a44dcca43 100644 --- a/docs/changelog/2026-03-04.mdx +++ b/docs/changelog/2026-03-04.mdx @@ -11,11 +11,7 @@ Previously, starting too many runs at once could overwhelm the machine. Now exce ## SSH access to running sandboxes -Use `--ssh` to get SSH access into running Daytona sandboxes for live debugging while the workflow executes. When something goes wrong mid-run, you can drop into the sandbox, inspect the filesystem, and understand the problem without waiting for the run to finish. - -```bash -fabro run start --ssh my-workflow.fabro -``` +Use `fabro ssh ` to get SSH access into running Daytona sandboxes for live debugging while the workflow executes. When something goes wrong mid-run, you can drop into the sandbox, inspect the filesystem, and understand the problem without waiting for the run to finish. Use `--preserve-sandbox` to keep sandboxes alive after a run completes for post-mortem inspection. diff --git a/docs/changelog/2026-03-09.mdx b/docs/changelog/2026-03-09.mdx index 68b64b27f..322a9d4da 100644 --- a/docs/changelog/2026-03-09.mdx +++ b/docs/changelog/2026-03-09.mdx @@ -35,7 +35,7 @@ image = "my-custom-image:latest" ``` ```bash -fabro run --ssh my-workflow.fabro +fabro ssh ``` ## `fabro cp` — copy files to and from sandboxes diff --git a/docs/execution/environments.mdx b/docs/execution/environments.mdx index c216e56fe..e8581f659 100644 --- a/docs/execution/environments.mdx +++ b/docs/execution/environments.mdx @@ -170,10 +170,10 @@ When using server defaults, labels are merged — run config labels override def Connect to a running Daytona sandbox via SSH for live debugging: ```bash -fabro run workflow.fabro --sandbox daytona --ssh +fabro ssh ``` -This creates temporary SSH credentials (valid for 60 minutes) and prints the connection command. +This creates temporary SSH credentials (valid for 60 minutes) and connects directly. ### Preserving the sandbox @@ -306,7 +306,7 @@ image = "my-custom-image:latest" Connect to a running exe.dev sandbox via SSH for live debugging: ```bash -fabro run workflow.fabro --sandbox exe --ssh +fabro ssh ``` This prints the SSH connection command so you can connect to the VM while the workflow runs. diff --git a/docs/human-tools/ssh-access.mdx b/docs/human-tools/ssh-access.mdx index b466ff5fa..11142931e 100644 --- a/docs/human-tools/ssh-access.mdx +++ b/docs/human-tools/ssh-access.mdx @@ -24,29 +24,12 @@ fabro ssh --print fabro ssh --ttl 120 ``` -## Enabling SSH access during `fabro run` - -Pass the `--ssh` flag to `fabro run` to create SSH credentials at the start of the run: - -```bash -fabro run workflow.fabro --sandbox daytona --ssh -``` - -After the sandbox is created, Fabro generates temporary SSH credentials (valid for 60 minutes) and prints the connection command: - -``` - Sandbox: daytona (fabro-20260307-143022-a3f2) - ssh daytona@fabro-20260307-143022-a3f2.ssh.daytona.io -``` - -Copy and run the `ssh` command in a separate terminal to connect. - ## Keeping the sandbox alive -By default, Daytona sandboxes are destroyed when the workflow finishes. To keep the sandbox running after the workflow completes — so you can continue debugging — combine `--ssh` with `--preserve-sandbox`: +By default, Daytona sandboxes are destroyed when the workflow finishes. To keep the sandbox running after the workflow completes — so you can continue debugging — pass `--preserve-sandbox`: ```bash -fabro run workflow.fabro --sandbox daytona --ssh --preserve-sandbox +fabro run workflow.fabro --sandbox daytona --preserve-sandbox ``` Without `--preserve-sandbox`, the SSH session is terminated when the run ends and the sandbox is cleaned up. @@ -73,9 +56,9 @@ Once connected, you have a full shell inside the sandbox VM: ## Credential lifetime -SSH credentials are temporary and expire after **60 minutes** by default. With `fabro ssh`, you can set a custom TTL with `--ttl `. If your session expires, run `fabro ssh` again or start a new run with `--ssh` to get fresh credentials. +SSH credentials are temporary and expire after **60 minutes** by default. With `fabro ssh`, you can set a custom TTL with `--ttl `. If your session expires, run `fabro ssh` again to get fresh credentials. ## Limitations -- SSH access is **Daytona-only**. Passing `--ssh` with other sandbox providers prints a warning and is ignored. +- SSH access is **Daytona-only**. - SSH access is currently available only from the **CLI**. The API server and web UI do not yet expose an SSH endpoint. diff --git a/docs/human-tools/vs-code.mdx b/docs/human-tools/vs-code.mdx index 533b63141..6d6f467af 100644 --- a/docs/human-tools/vs-code.mdx +++ b/docs/human-tools/vs-code.mdx @@ -16,17 +16,16 @@ VS Code remote access requires [SSH access](/human-tools/ssh-access), which is o ## Connecting to a sandbox -1. Start a workflow with SSH access and a preserved sandbox: +1. Start a workflow with a preserved Daytona sandbox: ```bash - fabro run workflow.fabro --sandbox daytona --ssh --preserve-sandbox + fabro run workflow.fabro --sandbox daytona --preserve-sandbox ``` -2. Fabro prints the SSH connection command: +2. Use `fabro ssh` to get the connection command: - ``` - Sandbox: daytona (fabro-20260307-143022-a3f2) - ssh daytona@fabro-20260307-143022-a3f2.ssh.daytona.io + ```bash + fabro ssh --print ``` 3. In VS Code, open the Command Palette (`Cmd+Shift+P` / `Ctrl+Shift+P`) and run **Remote-SSH: Connect to Host...** @@ -51,4 +50,4 @@ Once connected, VS Code operates as if the sandbox filesystem were local: - **Use `--preserve-sandbox`** — Without it, the sandbox is destroyed when the workflow finishes and your VS Code session disconnects. Combine with `auto_stop_interval` in your [run config](/execution/run-configuration) to control idle timeout. - **Pair with human gates** — When a workflow pauses at a [human gate](/workflows/human-in-the-loop), connect via VS Code to review the agent's changes before approving. -- **SSH credential lifetime** — Daytona SSH credentials expire after 60 minutes. If your VS Code session disconnects, you'll need to start a new run with `--ssh` to get fresh credentials. +- **SSH credential lifetime** — Daytona SSH credentials expire after 60 minutes by default. If your VS Code session disconnects, run `fabro ssh ` again to get fresh credentials (use `--ttl` to set a custom expiry). diff --git a/docs/integrations/daytona.mdx b/docs/integrations/daytona.mdx index 563bb2ba2..165c91ae2 100644 --- a/docs/integrations/daytona.mdx +++ b/docs/integrations/daytona.mdx @@ -114,17 +114,13 @@ for your organization. Connect to a running Daytona sandbox via SSH for live debugging: ```bash -fabro run workflow.fabro --sandbox daytona --ssh +fabro ssh ``` -This creates temporary SSH credentials (valid for 60 minutes) and prints the connection command: - -``` -SSH access ready: ssh daytona@fabro-20260307-143022-a3f2.ssh.daytona.io -``` +This creates temporary SSH credentials (valid for 60 minutes) and connects directly. Use `--print` to print the SSH command instead of connecting, or `--ttl` to set the credential expiry. -SSH credentials cannot be refreshed during a run. To keep the sandbox alive after the run completes, combine `--ssh` with `--preserve-sandbox`. +To keep the sandbox alive after the run completes, pass `--preserve-sandbox` to `fabro run`. ## Sandbox lifecycle diff --git a/docs/integrations/exe-dev.mdx b/docs/integrations/exe-dev.mdx index 64522dab6..de316af2e 100644 --- a/docs/integrations/exe-dev.mdx +++ b/docs/integrations/exe-dev.mdx @@ -41,7 +41,7 @@ image = "my-custom-image:latest" Connect to a running exe.dev sandbox via SSH for live debugging: ```bash -fabro run workflow.fabro --sandbox exe --ssh +fabro ssh ``` This prints the SSH connection command so you can connect to the VM while the workflow runs. diff --git a/docs/reference/cli.mdx b/docs/reference/cli.mdx index 83a6a0b36..3a013f14f 100644 --- a/docs/reference/cli.mdx +++ b/docs/reference/cli.mdx @@ -59,7 +59,6 @@ fabro run run.toml | `--goal ` | Override the workflow goal (exposed as `$goal` in prompts) | | `--goal-file ` | Read the goal from a file instead of inline text | | `--no-retro` | Skip retro generation after the run | -| `--ssh` | Create SSH access to the sandbox (Daytona or exe.dev) and print the connection command | | `--preserve-sandbox` | Keep the sandbox alive after the run finishes (for debugging) | | `-d, --detach` | Fork the workflow as a background process and print the run ID. Reconnect later with `fabro logs -f`. | @@ -92,7 +91,6 @@ fabro resume --checkpoint path/to/checkpoint.json --workflow workflow.fabro | `--goal ` | Override the workflow goal | | `--goal-file ` | Read the goal from a file | | `--no-retro` | Skip retro generation after the run | -| `--ssh` | Create SSH access to the sandbox | | `--preserve-sandbox` | Keep the sandbox alive after the run finishes | ## `fabro ps` diff --git a/lib/crates/fabro-api/src/server.rs b/lib/crates/fabro-api/src/server.rs index 8da054ac8..809d16c60 100644 --- a/lib/crates/fabro-api/src/server.rs +++ b/lib/crates/fabro-api/src/server.rs @@ -553,7 +553,7 @@ async fn execute_run(state: Arc, run_id: String) { runs.get(&run_id).and_then(|r| r.event_tx.clone()) }; - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); if let Some(tx_clone) = event_tx { emitter.on_event(move |event| { let _ = tx_clone.send(event.clone()); diff --git a/lib/crates/fabro-cli/src/commands/create.rs b/lib/crates/fabro-cli/src/commands/create.rs index 883851ef2..5be374d71 100644 --- a/lib/crates/fabro-cli/src/commands/create.rs +++ b/lib/crates/fabro-cli/src/commands/create.rs @@ -72,7 +72,6 @@ pub async fn create_run( .collect(), verbose: args.verbose, no_retro: args.no_retro, - ssh: args.ssh, preserve_sandbox: args.preserve_sandbox, dry_run: args.dry_run, auto_approve: args.auto_approve, diff --git a/lib/crates/fabro-cli/src/commands/resume.rs b/lib/crates/fabro-cli/src/commands/resume.rs index 4e7c2c8e0..b7bfb3550 100644 --- a/lib/crates/fabro-cli/src/commands/resume.rs +++ b/lib/crates/fabro-cli/src/commands/resume.rs @@ -1,4 +1,5 @@ use std::collections::HashMap; +use std::io::IsTerminal; use std::path::PathBuf; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -83,10 +84,6 @@ pub struct ResumeArgs { #[arg(long)] pub no_retro: bool, - /// Create SSH access to the Daytona sandbox and print the connection command - #[arg(long)] - pub ssh: bool, - /// Keep the sandbox alive after the run finishes (for debugging) #[arg(long)] pub preserve_sandbox: bool, @@ -99,6 +96,8 @@ struct ResumeContext { run_id: String, run_dir: PathBuf, sandbox: Arc, + /// Kept as Arc so the sandbox event callbacks can emit through it. Listeners + /// that need to be added later (e.g. ProgressUI) are registered separately. emitter: Arc, config: RunConfig, setup_commands: Vec, @@ -120,7 +119,7 @@ pub async fn resume_command( git_author: fabro_workflows::git::GitAuthor, ) -> anyhow::Result<()> { let ctx = if args.checkpoint.is_some() { - prepare_from_checkpoint(&args, styles, &github_app, git_author).await? + prepare_from_checkpoint(&args, &run_defaults, styles, &github_app, git_author).await? } else { prepare_from_branch(&args, styles, &run_defaults, &github_app, git_author).await? }; @@ -128,9 +127,10 @@ pub async fn resume_command( run_resumed(ctx, args, run_defaults, styles).await } -/// Checkpoint-file path: load checkpoint and graph from files, use a simple local sandbox. +/// Checkpoint-file path: load checkpoint and graph from files, resolve sandbox from flags/config. async fn prepare_from_checkpoint( args: &ResumeArgs, + run_defaults: &RunDefaults, styles: &Styles, github_app: &Option, git_author: fabro_workflows::git::GitAuthor, @@ -171,7 +171,58 @@ async fn prepare_from_checkpoint( let original_cwd = std::env::current_dir()?; let emitter = Arc::new(EventEmitter::new()); - let sandbox: Arc = local_sandbox_with_callback(original_cwd, Arc::clone(&emitter)); + // Resolve sandbox provider from CLI flag / config / defaults + let sandbox_provider = if args.dry_run { + SandboxProvider::Local + } else { + resolve_sandbox_provider(args.sandbox.map(Into::into), None, run_defaults)? + }; + + let sandbox: Arc = match sandbox_provider { + SandboxProvider::Local | SandboxProvider::Docker => { + local_sandbox_with_callback(original_cwd.clone(), Arc::clone(&emitter)) + } + #[cfg(feature = "exedev")] + SandboxProvider::Exe => { + let exe_config = super::run::resolve_exe_config(None, run_defaults); + let clone_params = super::run::resolve_exe_clone_params(&original_cwd); + let mgmt_ssh = fabro_sandbox::exe::OpensshRunner::connect_raw("exe.dev") + .await + .map_err(|e| anyhow::anyhow!("Failed to connect to exe.dev: {e}"))?; + let config = exe_config.unwrap_or_default(); + let mut env = fabro_sandbox::exe::ExeSandbox::new( + Box::new(mgmt_ssh), + config, + clone_params, + Some(run_id.clone()), + github_app.clone(), + ); + let emitter_cb = Arc::clone(&emitter); + env.set_event_callback(Arc::new(move |event| { + emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event }); + })); + Arc::new(env) + } + SandboxProvider::Ssh => { + let config = resolve_ssh_config(None, run_defaults) + .ok_or_else(|| anyhow::anyhow!("--sandbox ssh requires [sandbox.ssh] config"))?; + let clone_params = resolve_ssh_clone_params(&original_cwd); + let mut env = fabro_sandbox::ssh::SshSandbox::new( + config, + clone_params, + Some(run_id.clone()), + github_app.clone(), + ); + let emitter_cb = Arc::clone(&emitter); + env.set_event_callback(Arc::new(move |event| { + emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event }); + })); + Arc::new(env) + } + SandboxProvider::Daytona => { + bail!("resume from checkpoint is not yet supported with --sandbox daytona"); + } + }; let sandbox: Arc = Arc::new(fabro_agent::ReadBeforeWriteSandbox::new(sandbox)); let config = RunConfig { @@ -408,6 +459,20 @@ async fn run_resumed( original_cwd, } = ctx; + // Create progress UI (verbose mode shows detailed turn/tool counts and token usage) + let is_tty = std::io::stderr().is_terminal(); + let progress_ui = Arc::new(std::sync::Mutex::new(super::run_progress::ProgressUI::new( + is_tty, + args.verbose, + ))); + { + let p = Arc::clone(&progress_ui); + emitter.on_event(move |event| { + let mut ui = p.lock().expect("progress lock poisoned"); + ui.handle_event(event); + }); + } + let interviewer: Arc = if args.auto_approve { Arc::new(AutoApproveInterviewer) } else { @@ -468,6 +533,9 @@ async fn run_resumed( .await; let run_duration_ms = run_start.elapsed().as_millis() as u64; + // Finish progress bars before retro + progress_ui.lock().expect("progress lock poisoned").finish(); + // Restore cwd if we changed it (worktree is kept for `fabro cp` access; pruned separately) if let Some(ref cwd) = original_cwd { let _ = std::env::set_current_dir(cwd); @@ -508,8 +576,10 @@ async fn run_resumed( write_finalize_commit(&config, &run_dir).await; // Cleanup sandbox via engine (fires SandboxCleanup hook) + use super::run::resolve_preserve_sandbox; + let preserve = resolve_preserve_sandbox(args.preserve_sandbox, None, &run_defaults); let _ = engine - .cleanup_sandbox(&config.run_id, &graph.name, false) + .cleanup_sandbox(&config.run_id, &graph.name, preserve) .await; let outcome = engine_result?; diff --git a/lib/crates/fabro-cli/src/commands/run.rs b/lib/crates/fabro-cli/src/commands/run.rs index a40f6b884..22aa712f2 100644 --- a/lib/crates/fabro-cli/src/commands/run.rs +++ b/lib/crates/fabro-cli/src/commands/run.rs @@ -75,6 +75,7 @@ impl From for CliSandboxProvider { #[derive(Args)] pub struct RunArgs { /// Path to a .fabro workflow file or .toml task config + #[arg(required = true)] pub workflow: Option, /// Run output directory @@ -125,10 +126,6 @@ pub struct RunArgs { #[arg(long)] pub no_retro: bool, - /// Create SSH access to the Daytona sandbox and print the connection command - #[arg(long)] - pub ssh: bool, - /// Keep the sandbox alive after the run finishes (for debugging) #[arg(long)] pub preserve_sandbox: bool, @@ -717,7 +714,7 @@ pub async fn run_command( } // 3. Build event emitter - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); // Track the last git commit SHA from CheckpointCompleted events let last_git_sha: Arc>> = Arc::new(Mutex::new(None)); @@ -787,7 +784,7 @@ pub async fn run_command( }); } - run_progress::ProgressUI::register(&progress_ui, &mut emitter); + run_progress::ProgressUI::register(&progress_ui, &emitter); // 4. Build interviewer let interviewer: Arc = if args.auto_approve { @@ -1106,34 +1103,6 @@ pub async fn run_command( }); } - // Register SSH access listener - if args.ssh { - let deferred_sb_ssh = Arc::clone(&deferred_sandbox); - emitter.on_event(move |event| { - if let fabro_workflows::event::WorkflowRunEvent::SandboxInitialized { .. } = event { - if let Ok(rt) = tokio::runtime::Handle::try_current() { - let sb_lock = deferred_sb_ssh.lock().unwrap(); - if let Some(ref sb) = *sb_lock { - let sb = Arc::clone(sb); - rt.spawn(async move { - match sb.ssh_access_command().await { - Ok(Some(ssh_command)) => { - // Note: we can't emit from here since emitter is shared; - // SSH access info is logged via tracing. - tracing::info!(ssh_command, "SSH access ready"); - } - Ok(None) => {} - Err(e) => { - tracing::warn!(error = %e, "Failed to create SSH access"); - } - } - }); - } - } - } - }); - } - // Wrap emitter in Arc so we can share it with exec env callbacks let emitter = Arc::new(emitter); diff --git a/lib/crates/fabro-cli/src/commands/run_progress.rs b/lib/crates/fabro-cli/src/commands/run_progress.rs index 56c5ef9ef..6bc9f4d7c 100644 --- a/lib/crates/fabro-cli/src/commands/run_progress.rs +++ b/lib/crates/fabro-cli/src/commands/run_progress.rs @@ -264,7 +264,7 @@ impl ProgressUI { } /// Register event handlers on the emitter. - pub fn register(progress: &Arc>, emitter: &mut EventEmitter) { + pub fn register(progress: &Arc>, emitter: &EventEmitter) { let p = Arc::clone(progress); emitter.on_event(move |event| { let mut ui = p.lock().expect("progress lock poisoned"); @@ -309,7 +309,7 @@ impl ProgressUI { // ── Event dispatch ────────────────────────────────────────────────── - fn handle_event(&mut self, event: &WorkflowRunEvent) { + pub(crate) fn handle_event(&mut self, event: &WorkflowRunEvent) { match event { WorkflowRunEvent::Sandbox { event: sandbox_event, diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index 48c77f73c..9804a896d 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -757,7 +757,6 @@ async fn main_inner() -> (String, Result<()>) { .map(|(k, v)| format!("{k}={v}")) .collect(), no_retro: spec.no_retro, - ssh: spec.ssh, preserve_sandbox: spec.preserve_sandbox, detach: false, run_id: Some(spec.run_id), diff --git a/lib/crates/fabro-cli/tests/cli.rs b/lib/crates/fabro-cli/tests/cli.rs index a3a85b6f0..afc96268c 100644 --- a/lib/crates/fabro-cli/tests/cli.rs +++ b/lib/crates/fabro-cli/tests/cli.rs @@ -581,7 +581,7 @@ fn setup_run_dir( "labels": {}, "verbose": false, "no_retro": true, - "ssh": false, + "preserve_sandbox": false, "dry_run": true, "auto_approve": true @@ -632,7 +632,7 @@ digraph G { "labels": {}, "verbose": false, "no_retro": true, - "ssh": false, + "preserve_sandbox": false, "dry_run": true, "auto_approve": true diff --git a/lib/crates/fabro-workflows/src/devcontainer_bridge.rs b/lib/crates/fabro-workflows/src/devcontainer_bridge.rs index 46856ad83..2e574e705 100644 --- a/lib/crates/fabro-workflows/src/devcontainer_bridge.rs +++ b/lib/crates/fabro-workflows/src/devcontainer_bridge.rs @@ -378,7 +378,7 @@ mod tests { #[tokio::test] async fn emits_started_and_completed_events() { - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); let events = Arc::new(Mutex::new(Vec::new())); let events_clone = Arc::clone(&events); emitter.on_event(move |event| { @@ -410,7 +410,7 @@ mod tests { #[tokio::test] async fn failed_command_emits_failed_and_returns_error() { - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); let events = Arc::new(Mutex::new(Vec::new())); let events_clone = Arc::clone(&events); emitter.on_event(move |event| { @@ -430,7 +430,7 @@ mod tests { #[tokio::test] async fn empty_commands_is_noop() { - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); let events = Arc::new(Mutex::new(Vec::new())); let events_clone = Arc::clone(&events); emitter.on_event(move |event| { diff --git a/lib/crates/fabro-workflows/src/engine.rs b/lib/crates/fabro-workflows/src/engine.rs index 1de1c8e78..9409a198b 100644 --- a/lib/crates/fabro-workflows/src/engine.rs +++ b/lib/crates/fabro-workflows/src/engine.rs @@ -3228,7 +3228,7 @@ mod tests { let events = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); let events_clone = events.clone(); - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); emitter.on_event(move |event| { events_clone.lock().unwrap().push(format!("{event:?}")); }); @@ -5514,7 +5514,7 @@ mod tests { let events = std::sync::Arc::new(std::sync::Mutex::new(Vec::::new())); let events_clone = events.clone(); - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); emitter.on_event(move |event| { events_clone.lock().unwrap().push(event.clone()); }); @@ -5604,7 +5604,7 @@ mod tests { let events = Arc::new(std::sync::Mutex::new(Vec::::new())); let events_clone = events.clone(); - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); emitter.on_event(move |event| { events_clone.lock().unwrap().push(event.clone()); }); @@ -5635,7 +5635,7 @@ mod tests { let events = Arc::new(std::sync::Mutex::new(Vec::::new())); let events_clone = events.clone(); - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); emitter.on_event(move |event| { events_clone.lock().unwrap().push(event.clone()); }); @@ -5757,7 +5757,7 @@ mod tests { let events = Arc::new(std::sync::Mutex::new(Vec::::new())); let events_clone = events.clone(); - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); emitter.on_event(move |event| { events_clone.lock().unwrap().push(event.clone()); }); @@ -5800,7 +5800,7 @@ mod tests { let event_names = Arc::new(std::sync::Mutex::new(Vec::::new())); let names_clone = event_names.clone(); - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); emitter.on_event(move |event| { let name = match event { WorkflowRunEvent::SandboxInitialized { .. } => "SandboxInitialized", diff --git a/lib/crates/fabro-workflows/src/event.rs b/lib/crates/fabro-workflows/src/event.rs index 03c7d3bd0..6b6fa2aa6 100644 --- a/lib/crates/fabro-workflows/src/event.rs +++ b/lib/crates/fabro-workflows/src/event.rs @@ -1016,19 +1016,20 @@ fn epoch_millis() -> i64 { } /// Listener callback type for workflow run events. -type EventListener = Box; +type EventListener = Arc; /// Callback-based event emitter for workflow run events. pub struct EventEmitter { - listeners: Vec, + listeners: std::sync::Mutex>, /// Epoch milliseconds of the last `emit()` or `touch()` call. 0 until first event. last_event_at: AtomicI64, } impl std::fmt::Debug for EventEmitter { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let count = self.listeners.lock().map(|l| l.len()).unwrap_or(0); f.debug_struct("EventEmitter") - .field("listener_count", &self.listeners.len()) + .field("listener_count", &count) .field("last_event_at", &self.last_event_at.load(Ordering::Relaxed)) .finish() } @@ -1044,19 +1045,29 @@ impl EventEmitter { #[must_use] pub fn new() -> Self { Self { - listeners: Vec::new(), + listeners: std::sync::Mutex::new(Vec::new()), last_event_at: AtomicI64::new(0), } } - pub fn on_event(&mut self, listener: impl Fn(&WorkflowRunEvent) + Send + Sync + 'static) { - self.listeners.push(Box::new(listener)); + pub fn on_event(&self, listener: impl Fn(&WorkflowRunEvent) + Send + Sync + 'static) { + self.listeners + .lock() + .expect("listeners lock poisoned") + .push(Arc::new(listener)); } pub fn emit(&self, event: &WorkflowRunEvent) { self.last_event_at.store(epoch_millis(), Ordering::Relaxed); event.trace(); - for listener in &self.listeners { + // Clone the listener list so we don't hold the lock during dispatch. + // This prevents deadlocks if a listener calls emit() reentrantly. + let snapshot: Vec = self + .listeners + .lock() + .expect("listeners lock poisoned") + .clone(); + for listener in &snapshot { listener(event); } } @@ -1098,12 +1109,12 @@ mod tests { #[test] fn event_emitter_new_has_no_listeners() { let emitter = EventEmitter::new(); - assert_eq!(emitter.listeners.len(), 0); + assert_eq!(emitter.listeners.lock().unwrap().len(), 0); } #[test] fn event_emitter_calls_listener() { - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); let received = Arc::new(Mutex::new(Vec::new())); let received_clone = Arc::clone(&received); emitter.on_event(move |event| { @@ -1161,7 +1172,7 @@ mod tests { #[test] fn event_emitter_default() { let emitter = EventEmitter::default(); - assert_eq!(emitter.listeners.len(), 0); + assert_eq!(emitter.listeners.lock().unwrap().len(), 0); } #[test] @@ -2508,7 +2519,7 @@ mod tests { #[test] fn emitter_captures_retro_events() { - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); let received = Arc::new(Mutex::new(Vec::new())); let r = Arc::clone(&received); emitter.on_event(move |event| { diff --git a/lib/crates/fabro-workflows/src/run_spec.rs b/lib/crates/fabro-workflows/src/run_spec.rs index 5ed9800c9..a71041c11 100644 --- a/lib/crates/fabro-workflows/src/run_spec.rs +++ b/lib/crates/fabro-workflows/src/run_spec.rs @@ -17,7 +17,6 @@ pub struct RunSpec { pub labels: HashMap, pub verbose: bool, pub no_retro: bool, - pub ssh: bool, pub preserve_sandbox: bool, pub dry_run: bool, pub auto_approve: bool, @@ -60,7 +59,6 @@ mod tests { labels, verbose: true, no_retro: false, - ssh: true, preserve_sandbox: false, dry_run: false, auto_approve: true, diff --git a/lib/crates/fabro-workflows/tests/daytona_integration.rs b/lib/crates/fabro-workflows/tests/daytona_integration.rs index ada96fbd9..726e07d56 100644 --- a/lib/crates/fabro-workflows/tests/daytona_integration.rs +++ b/lib/crates/fabro-workflows/tests/daytona_integration.rs @@ -569,7 +569,7 @@ async fn daytona_git_checkpoint_remote_emits_events() { // Set up event collection let dir = tempfile::tempdir().unwrap(); - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); let events = Arc::new(std::sync::Mutex::new(Vec::new())); { let events_clone = Arc::clone(&events); @@ -754,7 +754,7 @@ async fn daytona_parallel_git_branching_e2e() { graph.edges.push(Edge::new("fan_in", "exit")); let run_tmp = tempfile::tempdir().unwrap(); - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); let events = Arc::new(std::sync::Mutex::new(Vec::new())); { let events_clone = Arc::clone(&events); diff --git a/lib/crates/fabro-workflows/tests/integration.rs b/lib/crates/fabro-workflows/tests/integration.rs index 6db83366c..4b43132b4 100644 --- a/lib/crates/fabro-workflows/tests/integration.rs +++ b/lib/crates/fabro-workflows/tests/integration.rs @@ -1300,7 +1300,7 @@ impl Handler for ContextSetterHandler { } } -fn collect_events(emitter: &mut EventEmitter) -> Arc>> { +fn collect_events(emitter: &EventEmitter) -> Arc>> { let events = Arc::new(std::sync::Mutex::new(Vec::new())); let events_clone = Arc::clone(&events); emitter.on_event(move |event| { @@ -1857,8 +1857,8 @@ async fn event_streaming_lifecycle() { }"#; let graph = parse(input).expect("parse"); let dir = tempfile::tempdir().unwrap(); - let mut emitter = EventEmitter::new(); - let events = collect_events(&mut emitter); + let emitter = EventEmitter::new(); + let events = collect_events(&emitter); let engine = WorkflowRunEngine::new(make_linear_registry(), Arc::new(emitter), local_env()); let config = RunConfig { run_dir: dir.path().to_path_buf(), @@ -2376,8 +2376,8 @@ async fn scenario_ship_a_feature() { let interviewer = Arc::new(AutoApproveInterviewer); let dir = tempfile::tempdir().unwrap(); - let mut emitter = EventEmitter::new(); - let events = collect_events(&mut emitter); + let emitter = EventEmitter::new(); + let events = collect_events(&emitter); let engine = WorkflowRunEngine::new( make_full_registry(interviewer), Arc::new(emitter), @@ -3500,8 +3500,8 @@ async fn integration_smoke_plan_implement_review_done() { // Run pipeline let interviewer = Arc::new(AutoApproveInterviewer); let dir = tempfile::tempdir().unwrap(); - let mut emitter = EventEmitter::new(); - let events = collect_events(&mut emitter); + let emitter = EventEmitter::new(); + let events = collect_events(&emitter); let engine = WorkflowRunEngine::new( make_full_registry(interviewer), Arc::new(emitter), @@ -7427,8 +7427,8 @@ fn engine_with_hooks_and_events( Arc>>, ) { let registry = make_linear_registry(); - let mut emitter = EventEmitter::new(); - let events = collect_events(&mut emitter); + let emitter = EventEmitter::new(); + let events = collect_events(&emitter); let sandbox = local_env(); let mut engine = WorkflowRunEngine::new(registry, Arc::new(emitter), sandbox); if !hooks.is_empty() { @@ -8885,8 +8885,8 @@ async fn large_context_values_are_offloaded_to_artifact_store() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let mut emitter = EventEmitter::new(); - let events = collect_events(&mut emitter); + let emitter = EventEmitter::new(); + let events = collect_events(&emitter); let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), local_env()); let config = RunConfig { run_dir: dir.path().to_path_buf(), @@ -10640,8 +10640,8 @@ async fn git_checkpoint_host_emits_events_and_diff_patch() { // 4. Set up event collection and engine let run_dir = tempfile::tempdir().unwrap(); - let mut emitter = EventEmitter::new(); - let events = collect_events(&mut emitter); + let emitter = EventEmitter::new(); + let events = collect_events(&emitter); let env: Arc = Arc::new(fabro_agent::LocalSandbox::new(worktree_path.clone())); @@ -11025,8 +11025,8 @@ async fn parallel_git_branching_host_e2e() { // 4. Set up engine with FileWriterHandler for branches let run_dir = tempfile::tempdir().unwrap(); - let mut emitter = EventEmitter::new(); - let events = collect_events(&mut emitter); + let emitter = EventEmitter::new(); + let events = collect_events(&emitter); let env: Arc = Arc::new(fabro_agent::LocalSandbox::new(worktree_path.clone())); @@ -11297,8 +11297,8 @@ async fn git_checkpoint_host_skips_empty_diff_patch() { graph.edges.push(Edge::new("work", "exit")); let run_dir = tempfile::tempdir().unwrap(); - let mut emitter = EventEmitter::new(); - let _events = collect_events(&mut emitter); + let emitter = EventEmitter::new(); + let _events = collect_events(&emitter); let env: Arc = Arc::new(fabro_agent::LocalSandbox::new(worktree_path.clone())); @@ -12210,8 +12210,8 @@ async fn e2e_circuit_breaker_emits_events_before_abort() { let dir = tempfile::tempdir().unwrap(); let graph = circuit_breaker_self_loop_graph(Some(3)); - let mut emitter = EventEmitter::new(); - let events = collect_events(&mut emitter); + let emitter = EventEmitter::new(); + let events = collect_events(&emitter); let mut registry = HandlerRegistry::new(Box::new(StartHandler)); registry.register("start", Box::new(StartHandler)); @@ -12832,7 +12832,7 @@ async fn e2e_stall_watchdog_triggers_from_dot_parsed_pipeline() { let events = Arc::new(std::sync::Mutex::new(Vec::new())); let events_clone = events.clone(); - let mut emitter = EventEmitter::new(); + let emitter = EventEmitter::new(); emitter.on_event(move |event| { events_clone.lock().unwrap().push(format!("{event:?}")); }); @@ -13123,8 +13123,8 @@ async fn asset_collection_local_sandbox_success() { registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); - let mut emitter = EventEmitter::new(); - let events = collect_events(&mut emitter); + let emitter = EventEmitter::new(); + let events = collect_events(&emitter); let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), sandbox.clone());