From 571ac0e1f3aed03e5173ef5b56db8216246680ea Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 26 Mar 2026 13:53:12 -0400 Subject: [PATCH] Guard resume against completed runs --- lib/crates/fabro-cli/src/commands/run.rs | 28 ++++++ lib/crates/fabro-cli/tests/cli.rs | 85 ++++++++++++++++++ .../fabro-workflows/src/operations/start.rs | 87 ++++++++++++++++++- 3 files changed, 199 insertions(+), 1 deletion(-) diff --git a/lib/crates/fabro-cli/src/commands/run.rs b/lib/crates/fabro-cli/src/commands/run.rs index 4d06502d2..2bf45c4d9 100644 --- a/lib/crates/fabro-cli/src/commands/run.rs +++ b/lib/crates/fabro-cli/src/commands/run.rs @@ -834,6 +834,31 @@ pub async fn resume_from_record( run_command_impl(args, styles, github_app, git_author, Some(record_run), true).await } +fn ensure_resume_target_is_not_already_successful(run_dir: &Path) -> anyhow::Result<()> { + const MESSAGE: &str = "run already finished successfully — nothing to resume"; + + if let Ok(record) = + fabro_workflows::run_status::RunStatusRecord::load(&run_dir.join("status.json")) + { + if record.status == fabro_workflows::run_status::RunStatus::Succeeded { + bail!(MESSAGE); + } + } + + if let Ok(conclusion) = + fabro_workflows::records::Conclusion::load(&run_dir.join("conclusion.json")) + { + if matches!( + conclusion.status, + StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped + ) { + bail!(MESSAGE); + } + } + + Ok(()) +} + /// Execute a full workflow run. /// /// # Errors @@ -906,6 +931,9 @@ async fn run_command_impl( .run_dir .clone() .unwrap_or_else(|| default_run_dir(&run_id, dry_run_flag)); + if resume { + ensure_resume_target_is_not_already_successful(&run_dir)?; + } let cached_run_restart = match &workflow { WorkflowState::Source(_) if !from_record => { let workflow_path = args.workflow.as_ref().unwrap(); diff --git a/lib/crates/fabro-cli/tests/cli.rs b/lib/crates/fabro-cli/tests/cli.rs index dac6dc28e..41fe27835 100644 --- a/lib/crates/fabro-cli/tests/cli.rs +++ b/lib/crates/fabro-cli/tests/cli.rs @@ -944,6 +944,91 @@ digraph G { ); } +#[test] +fn bug4_run_engine_resume_rejects_completed_run_without_mutating_it() { + let home = tempfile::tempdir().unwrap(); + let project = tempfile::tempdir().unwrap(); + let workflow_path = project.path().join("workflow.fabro"); + std::fs::write( + &workflow_path, + "\ +digraph Test { + start [shape=Mdiamond, label=\"Start\"] + exit [shape=Msquare, label=\"Exit\"] + start -> exit +} +", + ) + .unwrap(); + + let run = arc() + .env("HOME", home.path()) + .current_dir(project.path()) + .args([ + "run", + "--dry-run", + "--auto-approve", + "--no-retro", + "--detach", + workflow_path.to_str().unwrap(), + ]) + .assert() + .success(); + let run_id = String::from_utf8(run.get_output().stdout.clone()) + .unwrap() + .trim() + .to_string(); + + arc() + .env("HOME", home.path()) + .args(["wait", &run_id]) + .timeout(std::time::Duration::from_secs(10)) + .assert() + .success(); + + let inspect_before = arc() + .env("HOME", home.path()) + .args(["inspect", &run_id]) + .assert() + .success(); + let before: serde_json::Value = + serde_json::from_slice(&inspect_before.get_output().stdout).unwrap(); + let run_dir = before[0]["run_dir"].as_str().unwrap().to_string(); + let start_time_before = before[0]["start_record"]["start_time"] + .as_str() + .unwrap() + .to_string(); + let conclusion_ts_before = before[0]["conclusion"]["timestamp"] + .as_str() + .unwrap() + .to_string(); + + arc() + .env("HOME", home.path()) + .args(["_run_engine", "--run-dir", &run_dir, "--resume"]) + .timeout(std::time::Duration::from_secs(10)) + .assert() + .failure() + .stderr(predicate::str::contains("nothing to resume")); + + let inspect_after = arc() + .env("HOME", home.path()) + .args(["inspect", &run_id]) + .assert() + .success(); + let after: serde_json::Value = + serde_json::from_slice(&inspect_after.get_output().stdout).unwrap(); + + assert_eq!( + after[0]["start_record"]["start_time"].as_str().unwrap(), + start_time_before + ); + assert_eq!( + after[0]["conclusion"]["timestamp"].as_str().unwrap(), + conclusion_ts_before + ); +} + // Bug 3: attach loop must delete interview_request.json after handling it // to prevent re-prompting the user on the next poll iteration. #[test] diff --git a/lib/crates/fabro-workflows/src/operations/start.rs b/lib/crates/fabro-workflows/src/operations/start.rs index 345a02f9b..eb72166b2 100644 --- a/lib/crates/fabro-workflows/src/operations/start.rs +++ b/lib/crates/fabro-workflows/src/operations/start.rs @@ -10,7 +10,7 @@ use crate::outcome::StageStatus; use crate::pipeline::{ self, FinalizeOptions, Finalized, InitOptions, Persisted, PullRequestOptions, RetroOptions, }; -use crate::records::Checkpoint; +use crate::records::{Checkpoint, Conclusion}; use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions}; pub struct StartRetroOptions { @@ -83,6 +83,23 @@ pub async fn resume( run_dir: &std::path::Path, options: StartOptions, ) -> Result { + if let Ok(record) = crate::run_status::RunStatusRecord::load(&run_dir.join("status.json")) { + if record.status == crate::run_status::RunStatus::Succeeded { + return Err(FabroError::Precondition( + "run already finished successfully — nothing to resume".to_string(), + )); + } + } + if let Ok(conclusion) = Conclusion::load(&run_dir.join("conclusion.json")) { + if matches!( + conclusion.status, + StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped + ) { + return Err(FabroError::Precondition( + "run already finished successfully — nothing to resume".to_string(), + )); + } + } let cp_path = run_dir.join("checkpoint.json"); let checkpoint = Checkpoint::load(&cp_path) .map_err(|e| FabroError::Precondition(format!("no checkpoint to resume from: {e}")))?; @@ -220,6 +237,7 @@ mod tests { use std::sync::atomic::{AtomicUsize, Ordering}; use async_trait::async_trait; + use chrono::Utc; use fabro_agent::{DirEntry, ExecResult, GrepOptions, LocalSandbox, Sandbox}; use fabro_config::config::FabroConfig; use fabro_graphviz::graph::{Graph, Node}; @@ -686,4 +704,71 @@ mod tests { result = result.as_ref().map(|_| "Ok"), ); } + + #[tokio::test] + async fn resume_errors_when_run_already_finished_successfully() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run"); + let emitter = Arc::new(EventEmitter::new()); + let registry = Arc::new(test_registry()); + let sandbox: Arc = + Arc::new(LocalSandbox::new(std::env::current_dir().unwrap())); + + persisted_workflow(MINIMAL_DOT, &run_dir); + + let checkpoint = Checkpoint::from_context( + &Context::new(), + "start", + vec!["start".to_string()], + HashMap::new(), + HashMap::new(), + Some("exit".to_string()), + HashMap::new(), + HashMap::new(), + HashMap::new(), + ); + checkpoint.save(&run_dir.join("checkpoint.json")).unwrap(); + + crate::records::Conclusion { + timestamp: Utc::now(), + status: StageStatus::Success, + duration_ms: 1, + failure_reason: None, + final_git_commit_sha: None, + stages: vec![], + total_cost: None, + total_retries: 0, + total_input_tokens: 0, + total_output_tokens: 0, + total_cache_read_tokens: 0, + total_cache_write_tokens: 0, + total_reasoning_tokens: 0, + has_pricing: false, + } + .save(&run_dir.join("conclusion.json")) + .unwrap(); + + let result = resume( + &run_dir, + test_start_options( + &run_dir, + sandbox, + emitter, + registry, + LifecycleOptions { + setup_commands: vec![], + setup_command_timeout_ms: 1_000, + devcontainer_phases: vec![], + }, + false, + ), + ) + .await; + + assert!( + matches!(&result, Err(crate::error::FabroError::Precondition(_))), + "expected Precondition error, got: {result:?}", + result = result.as_ref().map(|_| "Ok"), + ); + } }