From 296eca568b10ff516f8d36ed340e52e29bca462c Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 19 Apr 2026 15:39:17 -0400 Subject: [PATCH] feat(workflow): capture final_patch on RunFailed Previously only Success/PartialSuccess outcomes captured the final unified-patch string into the run projection. Failed runs left RunProjection.final_patch empty, which meant the upcoming Files Changed tab could not degrade to a patch-only view once the sandbox was gone. Extend on_run_end to run git diff on Failed too, with a tighter 10 s timeout (vs 30 s on success) so a pathological workspace doesn't stall downstream terminal notifications (Slack, SSE, CI). Plumb the optional field through Event::WorkflowRunFailed, RunFailedProps, and the projection. Back-compat: final_patch is serde default-None, so pre-change events in SlateDB replay cleanly as None. No backfill required; old Failed runs show R4(c) empty state on the Files tab. Refs plan docs/plans/2026-04-19-002-feat-run-files-changed-tab-plan.md Co-Authored-By: Claude Opus 4.7 (1M context) --- lib/crates/fabro-server/src/server.rs | 9 +++ lib/crates/fabro-store/src/run_state.rs | 43 ++++++++++++++ lib/crates/fabro-types/src/run_event/run.rs | 4 ++ lib/crates/fabro-workflow/src/event.rs | 5 ++ .../fabro-workflow/src/lifecycle/event.rs | 2 + .../fabro-workflow/src/lifecycle/git.rs | 58 +++++++++++-------- .../fabro-workflow/src/operations/start.rs | 4 ++ lib/crates/fabro-workflow/src/sandbox_git.rs | 21 ++++++- 8 files changed, 119 insertions(+), 27 deletions(-) diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index bb62a3645..f017496eb 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -3157,6 +3157,7 @@ pub(crate) async fn reconcile_incomplete_runs_on_startup( duration_ms: 0, reason, git_commit_sha: None, + final_patch: None, }, ) .await?; @@ -3213,6 +3214,7 @@ async fn persist_shutdown_run_failures( duration_ms: 0, reason, git_commit_sha: None, + final_patch: None, }, ) .await?; @@ -3286,6 +3288,7 @@ async fn persist_cancelled_run_status(state: &AppState, run_id: RunId) -> anyhow duration_ms: 0, reason: Some(WorkflowStatusReason::Cancelled), git_commit_sha: None, + final_patch: None, }, ) .await @@ -3522,6 +3525,7 @@ async fn append_worker_exit_failure( duration_ms: 0, reason, git_commit_sha: None, + final_patch: None, }, ) .await @@ -4476,6 +4480,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { duration_ms: 0, reason: Some(WorkflowStatusReason::LaunchFailed), git_commit_sha: None, + final_patch: None, }, ) .await; @@ -4497,6 +4502,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { duration_ms: 0, reason: Some(WorkflowStatusReason::LaunchFailed), git_commit_sha: None, + final_patch: None, }, ) .await; @@ -4526,6 +4532,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { duration_ms: 0, reason: Some(WorkflowStatusReason::LaunchFailed), git_commit_sha: None, + final_patch: None, }, ) .await; @@ -4546,6 +4553,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { duration_ms: 0, reason: Some(WorkflowStatusReason::LaunchFailed), git_commit_sha: None, + final_patch: None, }, ) .await; @@ -4578,6 +4586,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { duration_ms: 0, reason: Some(WorkflowStatusReason::Terminated), git_commit_sha: None, + final_patch: None, }, ) .await; diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index c31454b45..4acdabf95 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -207,6 +207,7 @@ impl RunProjection { self.status = Some(run_status_record(RunStatus::Failed, props.reason, ts)); self.pending_control = None; self.conclusion = Some(conclusion_from_failed(props, ts)); + self.final_patch.clone_from(&props.final_patch); self.pending_interviews.clear(); } EventBody::RunRewound(_) => { @@ -655,6 +656,7 @@ mod tests { use std::collections::HashMap; use chrono::Utc; + use fabro_types::run_event::run::RunFailedProps; use fabro_types::run_event::{ InterviewCompletedProps, InterviewOption, InterviewStartedProps, RunControlEffectProps, }; @@ -1097,4 +1099,45 @@ mod tests { events[1].payload.as_value()["properties"]["definition_blob"] ); } + + #[test] + fn run_failed_with_final_patch_populates_projection() { + let mut state = RunProjection::default(); + let patch = "diff --git a/foo.rs b/foo.rs\n@@ -1 +1 @@\n-a\n+b\n"; + state + .apply_event(&test_event( + 1, + EventBody::RunFailed(RunFailedProps { + error: "boom".to_string(), + duration_ms: 42, + reason: None, + git_commit_sha: Some("abc123".to_string()), + final_patch: Some(patch.to_string()), + }), + None, + )) + .unwrap(); + + assert_eq!(state.final_patch.as_deref(), Some(patch)); + } + + #[test] + fn legacy_run_failed_event_without_final_patch_replays_as_none() { + let mut state = RunProjection::default(); + // Existing SlateDB events predating the final_patch field should deserialize + // cleanly as None — no backfill required. + state + .apply_event(&test_raw_event( + 1, + "run.failed", + &json!({ + "error": "boom", + "duration_ms": 1, + }), + None, + )) + .unwrap(); + + assert!(state.final_patch.is_none()); + } } diff --git a/lib/crates/fabro-types/src/run_event/run.rs b/lib/crates/fabro-types/src/run_event/run.rs index e7948284f..0fa6972cd 100644 --- a/lib/crates/fabro-types/src/run_event/run.rs +++ b/lib/crates/fabro-types/src/run_event/run.rs @@ -118,6 +118,10 @@ pub struct RunFailedProps { pub reason: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub git_commit_sha: Option, + // Optional unified-patch text captured at run end. Additive for back-compat: + // pre-change events replay with `final_patch: None` via serde default. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub final_patch: Option, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 7db5a75e5..0f688d188 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -141,6 +141,8 @@ pub enum Event { reason: Option, #[serde(default, skip_serializing_if = "Option::is_none")] git_commit_sha: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + final_patch: Option, }, RunNotice { level: RunNoticeLevel, @@ -1608,11 +1610,13 @@ fn event_body_from_event(event: &Event) -> EventBody { duration_ms, reason, git_commit_sha, + final_patch, } => EventBody::RunFailed(fabro_types::RunFailedProps { error: error.to_string(), duration_ms: *duration_ms, reason: *reason, git_commit_sha: git_commit_sha.clone(), + final_patch: final_patch.clone(), }), Event::RunNotice { level, @@ -3093,6 +3097,7 @@ mod tests { duration_ms: 900, reason: Some(StatusReason::WorkflowError), git_commit_sha: Some("abc123".to_string()), + final_patch: None, }); assert_eq!(stored.event_name(), "run.failed"); diff --git a/lib/crates/fabro-workflow/src/lifecycle/event.rs b/lib/crates/fabro-workflow/src/lifecycle/event.rs index 273e6d733..d4714944c 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/event.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/event.rs @@ -456,6 +456,7 @@ impl RunLifecycle for EventLifecycle { duration_ms, reason: Some(StatusReason::Cancelled), git_commit_sha: last_sha, + final_patch: final_patch.clone(), }); return; } @@ -484,6 +485,7 @@ impl RunLifecycle for EventLifecycle { duration_ms, reason: Some(StatusReason::WorkflowError), git_commit_sha: last_sha, + final_patch, }); } } diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index faa6f10ba..3900a545b 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -18,7 +18,7 @@ use crate::outcome::{BilledModelUsage, Outcome, StageStatus}; use crate::run_dump::RunDump; use crate::run_options::RunOptions; use crate::runtime_store::RunStoreHandle; -use crate::sandbox_git::{git_checkpoint, git_diff, git_push_host}; +use crate::sandbox_git::{git_checkpoint, git_diff, git_diff_with_timeout, git_push_host}; type WfRunState = ExecutionState>; type WfNodeResult = NodeResult>; @@ -301,31 +301,39 @@ impl RunLifecycle for GitLifecycle { } async fn on_run_end(&self, outcome: &Outcome, _state: &WfRunState) { - // Capture the final diff on success for event/store projection. - if (outcome.status == StageStatus::Success || outcome.status == StageStatus::PartialSuccess) - && self.run_options.git.is_some() + // Capture the final diff for event/store projection. + // + // Success/PartialSuccess uses the standard 30 s timeout. Failed runs + // use a shorter 10 s timeout: a pathological workspace (FS locks, + // corrupted index) must not stall terminal event emission downstream + // (Slack notifier, SSE RunFailed, CI hooks). + if self.run_options.git.is_none() { + return; + } + let timeout_ms = match outcome.status { + StageStatus::Success | StageStatus::PartialSuccess => 30_000, + _ => 10_000, + }; + if let Some(base_sha) = self + .run_options + .git + .as_ref() + .and_then(|g| g.base_sha.clone()) { - if let Some(base_sha) = self - .run_options - .git - .as_ref() - .and_then(|g| g.base_sha.clone()) - { - match git_diff(&*self.sandbox, &base_sha).await { - Ok(patch) if !patch.is_empty() => { - *self.final_patch.lock().unwrap() = Some(patch.clone()); - } - Ok(_) => { - *self.final_patch.lock().unwrap() = None; - } - Err(err) => { - *self.final_patch.lock().unwrap() = None; - self.emitter.emit(&Event::RunNotice { - level: RunNoticeLevel::Warn, - code: "git_diff_failed".to_string(), - message: format!("final diff failed: {err}"), - }); - } + match git_diff_with_timeout(&*self.sandbox, &base_sha, timeout_ms).await { + Ok(patch) if !patch.is_empty() => { + *self.final_patch.lock().unwrap() = Some(patch.clone()); + } + Ok(_) => { + *self.final_patch.lock().unwrap() = None; + } + Err(err) => { + *self.final_patch.lock().unwrap() = None; + self.emitter.emit(&Event::RunNotice { + level: RunNoticeLevel::Warn, + code: "git_diff_failed".to_string(), + message: format!("final diff failed: {err}"), + }); } } } diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index 58471ccd1..1e736d309 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -267,6 +267,7 @@ async fn persist_terminal_engine_failure( duration_ms: u64::try_from(duration.as_millis()).unwrap(), reason: status_reason, git_commit_sha: None, + final_patch: None, }) .await { @@ -862,6 +863,7 @@ impl Drop for DetachedRunBootstrapGuard { duration_ms: 0, reason: Some(reason), git_commit_sha: None, + final_patch: None, }) .await; }); @@ -929,6 +931,7 @@ impl Drop for DetachedRunCompletionGuard { duration_ms: 0, reason: Some(reason), git_commit_sha: None, + final_patch: None, }) .await; let _ = append_event_to_sink(&event_sink, &run_id, &Event::RunNotice { @@ -957,6 +960,7 @@ async fn persist_detached_failure( duration_ms: 0, reason: Some(reason), git_commit_sha: None, + final_patch: None, }) .await { diff --git a/lib/crates/fabro-workflow/src/sandbox_git.rs b/lib/crates/fabro-workflow/src/sandbox_git.rs index 75fff9d4e..76abc579e 100644 --- a/lib/crates/fabro-workflow/src/sandbox_git.rs +++ b/lib/crates/fabro-workflow/src/sandbox_git.rs @@ -181,13 +181,30 @@ pub async fn git_push_host( } } -/// Run a git diff via the sandbox. +/// Run a git diff via the sandbox (30 s default timeout). pub(crate) async fn git_diff( sandbox: &dyn Sandbox, base: &str, +) -> std::result::Result { + git_diff_with_timeout(sandbox, base, 30_000).await +} + +/// Run a git diff via the sandbox with a caller-supplied timeout in +/// milliseconds. +/// +/// Failure-path capture uses a shorter timeout than the checkpoint path so a +/// pathological workspace (FS locks, corrupted index) doesn't stall terminal +/// event emission downstream (Slack notifier, SSE, CI hooks). +pub(crate) async fn git_diff_with_timeout( + sandbox: &dyn Sandbox, + base: &str, + timeout_ms: u64, ) -> std::result::Result { let cmd = format!("{GIT_REMOTE} diff {base} HEAD"); - match sandbox.exec_command(&cmd, 30_000, None, None, None).await { + match sandbox + .exec_command(&cmd, timeout_ms, None, None, None) + .await + { Ok(r) if r.exit_code == 0 => Ok(r.stdout), Ok(r) => Err(format!("exit {}: {}", r.exit_code, r.stderr.trim())), Err(e) => Err(e.clone()),