scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-05 08:10:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Fix resume command bugs, remove --ssh flag, simplify EventEmitter

Browse source 
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<Vec<Arc<...>>>) 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) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-21 13:07:58 -04:00
parent 756915b1ea
commit e2f884df23
21 changed files with 160 additions and 142 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
docs/changelog/2026-03-04.mdx
View file

@ -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 <run-id>` 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.

2
docs/changelog/2026-03-09.mdx
View file

@ -35,7 +35,7 @@ image = "my-custom-image:latest"
```
```bash
fabro run --ssh my-workflow.fabro
fabro ssh <run-id>
```
## `fabro cp` — copy files to and from sandboxes

6
docs/execution/environments.mdx
View file

@ -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 <run-id>
```
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 <run-id>
```
This prints the SSH connection command so you can connect to the VM while the workflow runs.

25
docs/human-tools/ssh-access.mdx
View file

@ -24,29 +24,12 @@ fabro ssh <run-id> --print
fabro ssh <run-id> --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 <MINUTES>`. 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 <MINUTES>`. 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.

13
docs/human-tools/vs-code.mdx
View file

@ -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 <run-id> --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 <run-id>` again to get fresh credentials (use `--ttl` to set a custom expiry).

10
docs/integrations/daytona.mdx
View file

@ -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 <run-id>
```
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.
<Note>
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`.
</Note>
## Sandbox lifecycle

2
docs/integrations/exe-dev.mdx
View file

@ -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 <run-id>
```
This prints the SSH connection command so you can connect to the VM while the workflow runs.

2
docs/reference/cli.mdx
View file

@ -59,7 +59,6 @@ fabro run run.toml
| `--goal <GOAL>` | Override the workflow goal (exposed as `$goal` in prompts) |
| `--goal-file <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 <GOAL>` | Override the workflow goal |
| `--goal-file <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`

2
lib/crates/fabro-api/src/server.rs
View file

@ -553,7 +553,7 @@ async fn execute_run(state: Arc<AppState>, 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());

1
lib/crates/fabro-cli/src/commands/create.rs
View file

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

86
lib/crates/fabro-cli/src/commands/resume.rs
View file

@ -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<dyn Sandbox>,
/// 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<EventEmitter>,
config: RunConfig,
setup_commands: Vec<String>,
@ -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<fabro_github::GitHubAppCredentials>,
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<dyn Sandbox> = 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<dyn Sandbox> = 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<dyn Sandbox> = 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<dyn Interviewer> = 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?;

37
lib/crates/fabro-cli/src/commands/run.rs
View file

@ -75,6 +75,7 @@ impl From<SandboxProvider> for CliSandboxProvider {
#[derive(Args)]
pub struct RunArgs {
/// Path to a .fabro workflow file or .toml task config
#[arg(required = true)]
pub workflow: Option<PathBuf>,
/// 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<Mutex<Option<String>>> = 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<dyn Interviewer> = 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);

4
lib/crates/fabro-cli/src/commands/run_progress.rs
View file

@ -264,7 +264,7 @@ impl ProgressUI {
}
/// Register event handlers on the emitter.
pub fn register(progress: &Arc<Mutex<Self>>, emitter: &mut EventEmitter) {
pub fn register(progress: &Arc<Mutex<Self>>, 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,

1
lib/crates/fabro-cli/src/main.rs
View file

@ -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),

4
lib/crates/fabro-cli/tests/cli.rs
View file

@ -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

6
lib/crates/fabro-workflows/src/devcontainer_bridge.rs
View file

@ -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| {

12
lib/crates/fabro-workflows/src/engine.rs
View file

@ -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::<WorkflowRunEvent>::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::<WorkflowRunEvent>::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::<WorkflowRunEvent>::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::<WorkflowRunEvent>::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::<String>::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",

33
lib/crates/fabro-workflows/src/event.rs
View file

@ -1016,19 +1016,20 @@ fn epoch_millis() -> i64 {
}
/// Listener callback type for workflow run events.
type EventListener = Box<dyn Fn(&WorkflowRunEvent) + Send + Sync>;
type EventListener = Arc<dyn Fn(&WorkflowRunEvent) + Send + Sync>;
/// Callback-based event emitter for workflow run events.
pub struct EventEmitter {
listeners: Vec<EventListener>,
listeners: std::sync::Mutex<Vec<EventListener>>,
/// 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<EventListener> = 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| {

2
lib/crates/fabro-workflows/src/run_spec.rs
View file

@ -17,7 +17,6 @@ pub struct RunSpec {
pub labels: HashMap<String, String>,
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,

4
lib/crates/fabro-workflows/tests/daytona_integration.rs
View file

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

44
lib/crates/fabro-workflows/tests/integration.rs
View file

@ -1300,7 +1300,7 @@ impl Handler for ContextSetterHandler {
}
}
fn collect_events(emitter: &mut EventEmitter) -> Arc<std::sync::Mutex<Vec<WorkflowRunEvent>>> {
fn collect_events(emitter: &EventEmitter) -> Arc<std::sync::Mutex<Vec<WorkflowRunEvent>>> {
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<std::sync::Mutex<Vec<WorkflowRunEvent>>>,
) {
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<dyn fabro_agent::Sandbox> =
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<dyn fabro_agent::Sandbox> =
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<dyn fabro_agent::Sandbox> =
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());