From 835132cdae0e98d8bc78da15686aecf3d6a97d9a Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 15 Mar 2026 14:35:17 -0400 Subject: [PATCH] =?UTF-8?q?Add=20granular=20git=20events,=20rename=20GitCh?= =?UTF-8?q?eckpoint=20=E2=86=92=20CheckpointCompleted?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rename GitCheckpoint/GitCheckpointFailed to CheckpointCompleted/CheckpointFailed to separate checkpoint lifecycle from git operations. Add 7 new granular git events: GitCommit, GitPush, GitBranch, GitWorktreeAdd, GitWorktreeRemove, GitFetch, GitReset. Emit at all relevant call sites in engine.rs and parallel.rs. Update push helpers to return bool for GitPush success tracking. Co-Authored-By: Claude Opus 4.6 (1M context) --- docs-internal/events-strategy.md | 13 +- docs/execution/observability.mdx | 9 +- lib/crates/fabro-workflows/src/cli/logs.rs | 10 +- lib/crates/fabro-workflows/src/cli/run.rs | 6 +- lib/crates/fabro-workflows/src/engine.rs | 53 ++-- lib/crates/fabro-workflows/src/event.rs | 229 +++++++++++++++++- .../fabro-workflows/src/handler/parallel.rs | 23 +- .../tests/daytona_integration.rs | 6 +- .../fabro-workflows/tests/integration.rs | 10 +- 9 files changed, 317 insertions(+), 42 deletions(-) diff --git a/docs-internal/events-strategy.md b/docs-internal/events-strategy.md index d21a7a071..01b2ebfec 100644 --- a/docs-internal/events-strategy.md +++ b/docs-internal/events-strategy.md @@ -185,8 +185,15 @@ WorkflowRunEvent::MyNewEvent { node_id, duration_ms, .. } => { | Event | JSONL fields | |---|---| | `CheckpointSaved` | `node_id`, `node_label` | -| `GitCheckpoint` | `run_id`, `node_id`, `node_label`, `status`, `git_commit_sha` | -| `GitCheckpointFailed` | `node_id`, `node_label`, `error` | +| `CheckpointCompleted` | `run_id`, `node_id`, `node_label`, `status`, `git_commit_sha` | +| `CheckpointFailed` | `node_id`, `node_label`, `error` | +| `GitCommit` | `node_id` (optional), `node_label` (optional), `sha` | +| `GitPush` | `branch`, `success` | +| `GitBranch` | `branch`, `sha` | +| `GitWorktreeAdd` | `path`, `branch` | +| `GitWorktreeRemove` | `path` | +| `GitFetch` | `branch`, `success` | +| `GitReset` | `sha` | ### Human interaction @@ -296,5 +303,5 @@ Error information is stored as plain strings. The `error` field contains the hum | `cli/run.rs` non-verbose listener | `name`, `duration_ms`, `status`, `usage` from `StageCompleted/Failed` | CLI progress output | | `cli/mod.rs` `format_event_summary()` | All events | `-v` verbose output | | `cli/run.rs` cost accumulator | `usage` from `StageCompleted` | Total cost tracking | -| `cli/run.rs` git SHA tracker | `git_commit_sha` from `GitCheckpoint` | Final SHA for `conclusion.json` | +| `cli/run.rs` git SHA tracker | `git_commit_sha` from `CheckpointCompleted` | Final SHA for `conclusion.json` | | External tooling | `progress.jsonl` | Live monitoring, dashboards | diff --git a/docs/execution/observability.mdx b/docs/execution/observability.mdx index 8cf9713d4..3ac3931b3 100644 --- a/docs/execution/observability.mdx +++ b/docs/execution/observability.mdx @@ -59,7 +59,14 @@ Events fall into several categories: | `EdgeSelected` | `from_node`, `to_node`, `label`, `condition` | Transition between nodes | | `LoopRestart` | `from_node`, `to_node` | Loop restart edge taken | | `CheckpointSaved` | `node_id` | Checkpoint written to disk | -| `GitCheckpoint` | `node_id`, `git_commit_sha` | Checkpoint committed to Git | +| `CheckpointCompleted` | `node_id`, `git_commit_sha` | Checkpoint committed to Git | +| `GitCommit` | `node_id`, `sha` | Git commit created | +| `GitPush` | `branch`, `success` | Git push attempted | +| `GitBranch` | `branch`, `sha` | Git branch created | +| `GitWorktreeAdd` | `path`, `branch` | Git worktree added | +| `GitWorktreeRemove` | `path` | Git worktree removed | +| `GitFetch` | `branch`, `success` | Git fetch attempted | +| `GitReset` | `sha` | Git reset executed | | `Failover` | `stage`, `from_provider`, `to_provider`, `error` | LLM provider failover | **Parallel execution:** diff --git a/lib/crates/fabro-workflows/src/cli/logs.rs b/lib/crates/fabro-workflows/src/cli/logs.rs index fe13ef2f3..d3502b31c 100644 --- a/lib/crates/fabro-workflows/src/cli/logs.rs +++ b/lib/crates/fabro-workflows/src/cli/logs.rs @@ -568,7 +568,15 @@ pub fn format_event_pretty(line: &str, styles: &fabro_util::terminal::Styles) -> | "SetupCommandStarted" | "SetupCommandCompleted" | "CheckpointSaved" - | "GitCheckpoint" + | "CheckpointCompleted" + | "CheckpointFailed" + | "GitCommit" + | "GitPush" + | "GitBranch" + | "GitWorktreeAdd" + | "GitWorktreeRemove" + | "GitFetch" + | "GitReset" | "AssetsCaptured" => None, _ => None, diff --git a/lib/crates/fabro-workflows/src/cli/run.rs b/lib/crates/fabro-workflows/src/cli/run.rs index ca5420c9e..653f2a32e 100644 --- a/lib/crates/fabro-workflows/src/cli/run.rs +++ b/lib/crates/fabro-workflows/src/cli/run.rs @@ -486,12 +486,14 @@ pub async fn run_command( // 3. Build event emitter let mut emitter = EventEmitter::new(); - // Track the last git commit SHA from GitCheckpoint events + // Track the last git commit SHA from CheckpointCompleted events let last_git_sha: Arc>> = Arc::new(Mutex::new(None)); { let sha_clone = Arc::clone(&last_git_sha); emitter.on_event(move |event| { - if let crate::event::WorkflowRunEvent::GitCheckpoint { git_commit_sha, .. } = event { + if let crate::event::WorkflowRunEvent::CheckpointCompleted { git_commit_sha, .. } = + event + { *sha_clone.lock().unwrap() = Some(git_commit_sha.clone()); } }); diff --git a/lib/crates/fabro-workflows/src/engine.rs b/lib/crates/fabro-workflows/src/engine.rs index d4d6415ed..f88d78f68 100644 --- a/lib/crates/fabro-workflows/src/engine.rs +++ b/lib/crates/fabro-workflows/src/engine.rs @@ -665,12 +665,12 @@ pub(crate) async fn git_push_host( refspec: &str, github_app: &Option, label: &str, -) { +) -> bool { let (origin_url, _) = match crate::daytona_sandbox::detect_repo_info(repo_path) { Ok(info) => info, Err(e) => { tracing::warn!(error = %e, label, "Cannot detect origin for push"); - return; + return false; } }; @@ -680,12 +680,12 @@ pub(crate) async fn git_push_host( Ok(url) => url, Err(e) => { tracing::warn!(error = %e, label, "Failed to get token for push"); - return; + return false; } }, None => { tracing::warn!(label, "No GitHub App credentials for push"); - return; + return false; } }; @@ -696,13 +696,19 @@ pub(crate) async fn git_push_host( }) .await; match result { - Ok(()) => tracing::info!(label, "Pushed to origin"), - Err(e) => tracing::warn!(error = %e, label, "Failed to push"), + Ok(()) => { + tracing::info!(label, "Pushed to origin"); + true + } + Err(e) => { + tracing::warn!(error = %e, label, "Failed to push"); + false + } } } /// Push the run branch to origin inside a remote sandbox (best-effort). -async fn git_push_remote(sandbox: &dyn Sandbox, branch: &str) { +async fn git_push_remote(sandbox: &dyn Sandbox, branch: &str) -> bool { if let Err(e) = sandbox.refresh_push_credentials().await { tracing::warn!(error = %e, "Failed to refresh push credentials"); } @@ -710,12 +716,15 @@ async fn git_push_remote(sandbox: &dyn Sandbox, branch: &str) { match sandbox.exec_command(&cmd, 60_000, None, None, None).await { Ok(r) if r.exit_code == 0 => { tracing::info!(branch, "Pushed run branch to origin"); + true } Ok(r) => { tracing::warn!(branch, exit_code = r.exit_code, "Failed to push run branch"); + false } Err(e) => { tracing::warn!(branch, error = %e, "Failed to push run branch"); + false } } } @@ -1949,18 +1958,22 @@ impl WorkflowRunEngine { } self.services .emitter - .emit(&WorkflowRunEvent::GitCheckpoint { + .emit(&WorkflowRunEvent::CheckpointCompleted { run_id: run_id.clone(), node_id: node.id.clone(), status: outcome.status.to_string(), git_commit_sha: sha.clone(), }); + self.services.emitter.emit(&WorkflowRunEvent::GitCommit { + node_id: Some(node.id.clone()), + sha: sha.clone(), + }); // Push run branch (skip in dry-run mode) if !config.dry_run { if let Some(ref branch) = config.run_branch { - if self.services.sandbox.is_remote() { - git_push_remote(&*self.services.sandbox, branch).await; + let push_ok = if self.services.sandbox.is_remote() { + git_push_remote(&*self.services.sandbox, branch).await } else if let Some(ref repo_path) = config.host_repo_path { let refspec = format!("refs/heads/{branch}"); git_push_host( @@ -1969,8 +1982,14 @@ impl WorkflowRunEngine { &config.github_app, "run branch", ) - .await; - } + .await + } else { + false + }; + self.services.emitter.emit(&WorkflowRunEvent::GitPush { + branch: branch.clone(), + success: push_ok, + }); } // Push metadata branch (always from host) if let (Some(ref meta_branch), Some(ref repo_path)) = @@ -1985,13 +2004,17 @@ impl WorkflowRunEngine { .unwrap_or(meta_branch); let refspec = format!("{meta_branch}:refs/heads/fabro/meta/{run_id_part}"); - git_push_host( + let meta_push_ok = git_push_host( repo_path, &refspec, &config.github_app, "metadata branch", ) .await; + self.services.emitter.emit(&WorkflowRunEvent::GitPush { + branch: format!("fabro/meta/{run_id_part}"), + success: meta_push_ok, + }); } } @@ -2018,7 +2041,7 @@ impl WorkflowRunEngine { Err(e) => { self.services .emitter - .emit(&WorkflowRunEvent::GitCheckpointFailed { + .emit(&WorkflowRunEvent::CheckpointFailed { node_id: node.id.clone(), error: e.clone(), }); @@ -5171,7 +5194,7 @@ mod tests { let git_checkpoint_node_ids: Vec<&str> = collected .iter() .filter_map(|e| match e { - WorkflowRunEvent::GitCheckpoint { node_id, .. } => Some(node_id.as_str()), + WorkflowRunEvent::CheckpointCompleted { node_id, .. } => Some(node_id.as_str()), _ => None, }) .collect(); diff --git a/lib/crates/fabro-workflows/src/event.rs b/lib/crates/fabro-workflows/src/event.rs index 2c608b744..e700f3cd9 100644 --- a/lib/crates/fabro-workflows/src/event.rs +++ b/lib/crates/fabro-workflows/src/event.rs @@ -115,16 +115,43 @@ pub enum WorkflowRunEvent { CheckpointSaved { node_id: String, }, - GitCheckpoint { + CheckpointCompleted { run_id: String, node_id: String, status: String, git_commit_sha: String, }, - GitCheckpointFailed { + CheckpointFailed { node_id: String, error: String, }, + GitCommit { + #[serde(default, skip_serializing_if = "Option::is_none")] + node_id: Option, + sha: String, + }, + GitPush { + branch: String, + success: bool, + }, + GitBranch { + branch: String, + sha: String, + }, + GitWorktreeAdd { + path: String, + branch: String, + }, + GitWorktreeRemove { + path: String, + }, + GitFetch { + branch: String, + success: bool, + }, + GitReset { + sha: String, + }, EdgeSelected { from_node: String, to_node: String, @@ -447,16 +474,48 @@ impl WorkflowRunEvent { Self::CheckpointSaved { node_id } => { debug!(node_id, "Checkpoint saved"); } - Self::GitCheckpoint { + Self::CheckpointCompleted { run_id, node_id, status, .. } => { - debug!(run_id, node_id, status, "Git checkpoint"); + debug!(run_id, node_id, status, "Checkpoint completed"); } - Self::GitCheckpointFailed { node_id, error } => { - error!(node_id, error, "Git checkpoint commit failed"); + Self::CheckpointFailed { node_id, error } => { + error!(node_id, error, "Checkpoint failed"); + } + Self::GitCommit { node_id, sha } => { + debug!( + node_id = node_id.as_deref().unwrap_or(""), + sha, "Git commit" + ); + } + Self::GitPush { branch, success } => { + if *success { + debug!(branch, "Git push succeeded"); + } else { + warn!(branch, "Git push failed"); + } + } + Self::GitBranch { branch, sha } => { + debug!(branch, sha, "Git branch created"); + } + Self::GitWorktreeAdd { path, branch } => { + debug!(path, branch, "Git worktree added"); + } + Self::GitWorktreeRemove { path } => { + debug!(path, "Git worktree removed"); + } + Self::GitFetch { branch, success } => { + if *success { + debug!(branch, "Git fetch succeeded"); + } else { + warn!(branch, "Git fetch failed"); + } + } + Self::GitReset { sha } => { + debug!(sha, "Git reset"); } Self::EdgeSelected { from_node, @@ -931,10 +990,14 @@ fn rename_fields(event_name: &str, fields: &mut serde_json::Map Some(r.stdout.trim().to_string()), + Ok(r) if r.exit_code == 0 => { + let sha = r.stdout.trim().to_string(); + emitter.emit(&WorkflowRunEvent::GitCommit { + node_id: Some(setup.target_id.clone()), + sha: sha.clone(), + }); + Some(sha) + } _ => None, } } else { @@ -571,6 +589,9 @@ impl Handler for ParallelHandler { if let Some(ref wt_path) = result.worktree_path { let wt_str = wt_path.to_string_lossy().to_string(); crate::engine::git_remove_worktree(&*services.sandbox, &wt_str).await; + services + .emitter + .emit(&WorkflowRunEvent::GitWorktreeRemove { path: wt_str }); } } diff --git a/lib/crates/fabro-workflows/tests/daytona_integration.rs b/lib/crates/fabro-workflows/tests/daytona_integration.rs index 15d67cd7d..6324f2384 100644 --- a/lib/crates/fabro-workflows/tests/daytona_integration.rs +++ b/lib/crates/fabro-workflows/tests/daytona_integration.rs @@ -613,13 +613,13 @@ async fn daytona_git_checkpoint_remote_emits_events() { .expect("pipeline should succeed"); assert_eq!(outcome.status, StageStatus::Success); - // Assert GitCheckpoint events were emitted + // Assert CheckpointCompleted events were emitted { let events = events.lock().unwrap(); let git_events: Vec<_> = events .iter() .filter_map(|e| { - if let fabro_workflows::event::WorkflowRunEvent::GitCheckpoint { + if let fabro_workflows::event::WorkflowRunEvent::CheckpointCompleted { node_id, git_commit_sha, .. @@ -636,7 +636,7 @@ async fn daytona_git_checkpoint_remote_emits_events() { assert_eq!( git_events.len(), 1, - "expected 1 GitCheckpoint event (work node only), got {}", + "expected 1 CheckpointCompleted event (work node only), got {}", git_events.len() ); assert!( diff --git a/lib/crates/fabro-workflows/tests/integration.rs b/lib/crates/fabro-workflows/tests/integration.rs index 8f3a2e88d..73a1eb2ff 100644 --- a/lib/crates/fabro-workflows/tests/integration.rs +++ b/lib/crates/fabro-workflows/tests/integration.rs @@ -10679,7 +10679,7 @@ impl Handler for FileWriterHandler { } } -/// End-to-end test: pipeline with git checkpointing enabled emits `GitCheckpoint` +/// End-to-end test: pipeline with git checkpointing enabled emits `CheckpointCompleted` /// events with valid commit SHAs and writes `diff.patch` per stage. #[tokio::test] async fn git_checkpoint_host_emits_events_and_diff_patch() { @@ -10796,12 +10796,12 @@ async fn git_checkpoint_host_emits_events_and_diff_patch() { .expect("pipeline should succeed"); assert_eq!(outcome.status, StageStatus::Success); - // 6. Assert GitCheckpoint events were emitted + // 6. Assert CheckpointCompleted events were emitted let events = events.lock().unwrap(); let git_events: Vec<_> = events .iter() .filter_map(|e| { - if let WorkflowRunEvent::GitCheckpoint { + if let WorkflowRunEvent::CheckpointCompleted { node_id, git_commit_sha, .. @@ -10816,12 +10816,12 @@ async fn git_checkpoint_host_emits_events_and_diff_patch() { // work node gets a checkpoint commit (start is skipped, exit is terminal) assert!( !git_events.is_empty(), - "expected at least 1 GitCheckpoint event, got {}", + "expected at least 1 CheckpointCompleted event, got {}", git_events.len() ); assert!( !git_events.iter().any(|(id, _)| id == "start"), - "start node should not have a git checkpoint" + "start node should not have a checkpoint" ); // Each SHA should be a valid 40-char hex string assert!(