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

Add granular git events, rename GitCheckpoint → CheckpointCompleted

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-15 14:35:17 -04:00
parent 92a0582b4a
commit 835132cdae
9 changed files with 317 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
docs-internal/events-strategy.md
View file

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

9
docs/execution/observability.mdx
View file

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

10
lib/crates/fabro-workflows/src/cli/logs.rs
View file

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

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

@ -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<Mutex<Option<String>>> = 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());
}
});

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

@ -665,12 +665,12 @@ pub(crate) async fn git_push_host(
refspec: &str,
github_app: &Option<fabro_github::GitHubAppCredentials>,
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();

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

@ -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<String>,
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<String, serde_js
rename(fields, "start_node", "start_node_id");
} else if event_name == "SubgraphCompleted"
|| event_name == "CheckpointSaved"
|| event_name == "GitCheckpoint"
|| event_name == "GitCheckpointFailed"
|| event_name == "CheckpointCompleted"
|| event_name == "CheckpointFailed"
{
default_node_label(fields);
} else if event_name == "GitCommit" {
if fields.contains_key("node_id") {
default_node_label(fields);
}
} else if event_name.starts_with("DevcontainerLifecycleCommand")
|| event_name == "DevcontainerLifecycleFailed"
{
@ -1868,13 +1931,27 @@ mod tests {
}
#[test]
fn rename_fields_git_checkpoint_failed() {
let event = WorkflowRunEvent::GitCheckpointFailed {
fn rename_fields_checkpoint_completed() {
let event = WorkflowRunEvent::CheckpointCompleted {
run_id: "r1".to_string(),
node_id: "work".to_string(),
status: "success".to_string(),
git_commit_sha: "abc123".to_string(),
};
let (name, fields) = flatten_event(&event);
assert_eq!(name, "CheckpointCompleted");
assert_eq!(fields["node_id"], "work");
assert_eq!(fields["node_label"], "work");
}
#[test]
fn rename_fields_checkpoint_failed() {
let event = WorkflowRunEvent::CheckpointFailed {
node_id: "fix_lints".to_string(),
error: "git add failed (exit 1): fatal: not a git repository".to_string(),
};
let (name, fields) = flatten_event(&event);
assert_eq!(name, "GitCheckpointFailed");
assert_eq!(name, "CheckpointFailed");
assert_eq!(fields["node_id"], "fix_lints");
assert_eq!(fields["node_label"], "fix_lints");
assert_eq!(
@ -1883,6 +1960,136 @@ mod tests {
);
}
#[test]
fn rename_fields_git_commit_with_node_id() {
let event = WorkflowRunEvent::GitCommit {
node_id: Some("work".to_string()),
sha: "abc123".to_string(),
};
let (name, fields) = flatten_event(&event);
assert_eq!(name, "GitCommit");
assert_eq!(fields["node_id"], "work");
assert_eq!(fields["node_label"], "work");
assert_eq!(fields["sha"], "abc123");
}
#[test]
fn rename_fields_git_commit_without_node_id() {
let event = WorkflowRunEvent::GitCommit {
node_id: None,
sha: "abc123".to_string(),
};
let (name, fields) = flatten_event(&event);
assert_eq!(name, "GitCommit");
assert!(!fields.contains_key("node_label"));
assert_eq!(fields["sha"], "abc123");
}
#[test]
fn git_commit_serialization() {
let event = WorkflowRunEvent::GitCommit {
node_id: Some("work".to_string()),
sha: "abc123".to_string(),
};
let json = serde_json::to_string(&event).unwrap();
assert!(json.contains("GitCommit"));
assert!(json.contains("\"sha\":\"abc123\""));
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(deserialized, WorkflowRunEvent::GitCommit { sha, .. } if sha == "abc123"));
// node_id None is omitted
let event_none = WorkflowRunEvent::GitCommit {
node_id: None,
sha: "def456".to_string(),
};
let json_none = serde_json::to_string(&event_none).unwrap();
assert!(!json_none.contains("node_id"));
}
#[test]
fn git_push_serialization() {
let event = WorkflowRunEvent::GitPush {
branch: "fabro/run/123".to_string(),
success: true,
};
let json = serde_json::to_string(&event).unwrap();
assert!(json.contains("GitPush"));
assert!(json.contains("\"success\":true"));
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(
deserialized,
WorkflowRunEvent::GitPush { success: true, .. }
));
}
#[test]
fn git_branch_serialization() {
let event = WorkflowRunEvent::GitBranch {
branch: "fabro/run/123/work".to_string(),
sha: "abc123".to_string(),
};
let json = serde_json::to_string(&event).unwrap();
assert!(json.contains("GitBranch"));
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(
matches!(deserialized, WorkflowRunEvent::GitBranch { branch, .. } if branch == "fabro/run/123/work")
);
}
#[test]
fn git_worktree_add_serialization() {
let event = WorkflowRunEvent::GitWorktreeAdd {
path: "/tmp/wt".to_string(),
branch: "work".to_string(),
};
let json = serde_json::to_string(&event).unwrap();
assert!(json.contains("GitWorktreeAdd"));
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(
matches!(deserialized, WorkflowRunEvent::GitWorktreeAdd { path, .. } if path == "/tmp/wt")
);
}
#[test]
fn git_worktree_remove_serialization() {
let event = WorkflowRunEvent::GitWorktreeRemove {
path: "/tmp/wt".to_string(),
};
let json = serde_json::to_string(&event).unwrap();
assert!(json.contains("GitWorktreeRemove"));
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(
matches!(deserialized, WorkflowRunEvent::GitWorktreeRemove { path } if path == "/tmp/wt")
);
}
#[test]
fn git_fetch_serialization() {
let event = WorkflowRunEvent::GitFetch {
branch: "main".to_string(),
success: false,
};
let json = serde_json::to_string(&event).unwrap();
assert!(json.contains("GitFetch"));
assert!(json.contains("\"success\":false"));
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(
deserialized,
WorkflowRunEvent::GitFetch { success: false, .. }
));
}
#[test]
fn git_reset_serialization() {
let event = WorkflowRunEvent::GitReset {
sha: "abc123".to_string(),
};
let json = serde_json::to_string(&event).unwrap();
assert!(json.contains("GitReset"));
let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap();
assert!(matches!(deserialized, WorkflowRunEvent::GitReset { sha } if sha == "abc123"));
}
#[test]
fn rename_fields_sandbox_snapshot_pulling() {
let event = WorkflowRunEvent::Sandbox {

23
lib/crates/fabro-workflows/src/handler/parallel.rs
View file

@ -329,6 +329,10 @@ impl Handler for ParallelHandler {
"failed to create branch {branch_name}"
)));
}
services.emitter.emit(&WorkflowRunEvent::GitBranch {
branch: branch_name.clone(),
sha: bsha.to_string(),
});
if !crate::engine::git_replace_worktree(
&*services.sandbox,
&wt_path_str,
@ -340,6 +344,10 @@ impl Handler for ParallelHandler {
"failed to add worktree {wt_path_str}"
)));
}
services.emitter.emit(&WorkflowRunEvent::GitWorktreeAdd {
path: wt_path_str.clone(),
branch: branch_name.clone(),
});
let reset_cmd = format!("{} reset --hard {bsha}", crate::engine::GIT_REMOTE);
let reset_result = services
.sandbox
@ -350,6 +358,9 @@ impl Handler for ParallelHandler {
"failed to reset worktree {wt_path_str}"
)));
}
services.emitter.emit(&WorkflowRunEvent::GitReset {
sha: bsha.to_string(),
});
branch_context.set(keys::INTERNAL_WORK_DIR, serde_json::json!(&wt_path_str));
@ -476,7 +487,14 @@ impl Handler for ParallelHandler {
.exec_command(&sha_cmd, 10_000, None, None, None)
.await;
match sha_result {
Ok(r) if r.exit_code == 0 => 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 });
}
}

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

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

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

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