From 71e151c699f44a24cf5e2bb64d4ac3a8224a5239 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 27 Mar 2026 21:35:30 -0400 Subject: [PATCH] Split validate.rs and resume.rs out of operations Extract validate() into its own file from create.rs and resume() into its own file from start.rs, maintaining one public operation per file. Shared helpers (preprocess_and_validate, execute_persisted_run) become pub(super) so the new modules can call them. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../fabro-workflows/src/operations/create.rs | 31 +---------- .../fabro-workflows/src/operations/mod.rs | 8 ++- .../fabro-workflows/src/operations/resume.rs | 52 +++++++++++++++++++ .../fabro-workflows/src/operations/start.rs | 50 ++---------------- .../src/operations/validate.rs | 38 ++++++++++++++ 5 files changed, 102 insertions(+), 77 deletions(-) create mode 100644 lib/crates/fabro-workflows/src/operations/resume.rs create mode 100644 lib/crates/fabro-workflows/src/operations/validate.rs diff --git a/lib/crates/fabro-workflows/src/operations/create.rs b/lib/crates/fabro-workflows/src/operations/create.rs index b4366c9e8..25fa8d752 100644 --- a/lib/crates/fabro-workflows/src/operations/create.rs +++ b/lib/crates/fabro-workflows/src/operations/create.rs @@ -17,13 +17,6 @@ use super::source::{resolve_workflow, ResolveWorkflowInput, WorkflowInput}; const RUN_CONFIG_FILE: &str = "workflow.toml"; -pub struct ValidateInput { - pub workflow: WorkflowInput, - pub settings: FabroSettings, - pub cwd: PathBuf, - pub custom_transforms: Vec>, -} - #[derive(Clone, Debug)] pub struct CreateRunInput { pub workflow: WorkflowInput, @@ -57,27 +50,6 @@ struct PersistCreateOptions { base_dir: Option, } -/// Parse, transform, and validate a DOT source string. -/// -/// Returns `Validated` even when validation produced errors. Call -/// `validated.raise_on_errors()` if the caller wants to fail fast. -pub fn validate(input: ValidateInput) -> Result { - let resolved = resolve_workflow(ResolveWorkflowInput { - workflow: input.workflow, - settings: input.settings, - cwd: input.cwd, - }) - .map_err(|err| FabroError::Parse(err.to_string()))?; - - preprocess_and_validate( - &resolved.raw_source, - resolved.base_dir, - input.custom_transforms, - Some(&resolved.settings), - resolved.goal_override.as_deref(), - ) -} - /// Resolve workflow inputs, normalize settings, and persist a run directory. pub fn create(request: CreateRunInput) -> Result { let resolved = resolve_workflow(ResolveWorkflowInput { @@ -196,7 +168,7 @@ fn create_from_source( persist_validated(validated, options) } -fn preprocess_and_validate( +pub(super) fn preprocess_and_validate( dot_source: &str, base_dir: Option, custom_transforms: Vec>, @@ -344,6 +316,7 @@ mod tests { use super::*; use fabro_graphviz::graph::AttrValue; + use crate::operations::{validate, ValidateInput}; use crate::run_status::RunStatusRecordExt; fn validate_dot(dot_source: &str, settings: FabroSettings) -> Validated { diff --git a/lib/crates/fabro-workflows/src/operations/mod.rs b/lib/crates/fabro-workflows/src/operations/mod.rs index ee63a6dfd..4ab837971 100644 --- a/lib/crates/fabro-workflows/src/operations/mod.rs +++ b/lib/crates/fabro-workflows/src/operations/mod.rs @@ -1,15 +1,19 @@ mod create; mod fork; +mod resume; mod rewind; mod source; mod start; +mod validate; pub use crate::pipeline::{DevcontainerSpec, LlmSpec, SandboxEnvSpec, SandboxSpec}; -pub use create::{create, validate, CreateRunInput, CreatedRun, ValidateInput}; +pub use create::{create, CreateRunInput, CreatedRun}; pub use fork::{fork, ForkRunInput}; +pub use resume::resume; pub use rewind::{ build_timeline, find_run_id_by_prefix, rewind, RewindInput, RewindTarget, RunTimeline, TimelineEntry, }; pub use source::WorkflowInput; -pub use start::{resume, start, StartServices, Started}; +pub use start::{start, StartServices, Started}; +pub use validate::{validate, ValidateInput}; diff --git a/lib/crates/fabro-workflows/src/operations/resume.rs b/lib/crates/fabro-workflows/src/operations/resume.rs new file mode 100644 index 000000000..54ec5a4b1 --- /dev/null +++ b/lib/crates/fabro-workflows/src/operations/resume.rs @@ -0,0 +1,52 @@ +use std::path::Path; + +use crate::error::FabroError; +use crate::outcome::StageStatus; +use crate::records::{Checkpoint, CheckpointExt, Conclusion, ConclusionExt}; +use crate::run_status::{self, RunStatus, RunStatusRecordExt}; + +use super::start::{execute_persisted_run, StartServices, Started}; + +/// Resume a workflow run from its checkpoint. Errors if no checkpoint is found. +pub async fn resume(run_dir: &Path, services: StartServices) -> Result { + if let Ok(record) = run_status::RunStatusRecord::load(&run_dir.join("status.json")) { + if record.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}")))?; + + cleanup_resume_artifacts(run_dir); + run_status::write_run_status(run_dir, RunStatus::Submitted, None); + + execute_persisted_run(run_dir, Some(checkpoint), services).await +} + +fn cleanup_resume_artifacts(run_dir: &Path) { + for name in [ + "conclusion.json", + "pull_request.json", + "detached_failure.json", + "interview_request.json", + "interview_response.json", + "interview_request.claim", + "progress.jsonl", + ] { + let _ = std::fs::remove_file(run_dir.join(name)); + } +} diff --git a/lib/crates/fabro-workflows/src/operations/start.rs b/lib/crates/fabro-workflows/src/operations/start.rs index c08e0af08..97fd920bb 100644 --- a/lib/crates/fabro-workflows/src/operations/start.rs +++ b/lib/crates/fabro-workflows/src/operations/start.rs @@ -22,7 +22,7 @@ use crate::pipeline::{ FinalizeOptions, Finalized, InitOptions, LlmSpec, Persisted, PullRequestOptions, RetroOptions, SandboxEnvSpec, SandboxSpec, }; -use crate::records::{Checkpoint, CheckpointExt, Conclusion, ConclusionExt, RunRecordExt}; +use crate::records::{Checkpoint, Conclusion, ConclusionExt, RunRecordExt}; use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions}; use crate::run_status::{self, RunStatus, RunStatusRecordExt, StatusReason}; @@ -97,37 +97,7 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result Result { - if let Ok(record) = run_status::RunStatusRecord::load(&run_dir.join("status.json")) { - if record.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}")))?; - - cleanup_resume_artifacts(run_dir); - run_status::write_run_status(run_dir, RunStatus::Submitted, None); - - execute_persisted_run(run_dir, Some(checkpoint), services).await -} - -async fn execute_persisted_run( +pub(super) async fn execute_persisted_run( run_dir: &Path, checkpoint: Option, services: StartServices, @@ -186,20 +156,6 @@ fn persist_terminal_engine_failure(run_dir: &Path, error: &FabroError, duration: persist_terminal_outcome(run_dir, &conclusion, run_status, status_reason); } -fn cleanup_resume_artifacts(run_dir: &Path) { - for name in [ - "conclusion.json", - "pull_request.json", - "detached_failure.json", - "interview_request.json", - "interview_response.json", - "interview_request.claim", - "progress.jsonl", - ] { - let _ = std::fs::remove_file(run_dir.join(name)); - } -} - fn derive_start_options( persisted: &Persisted, services: StartServices, @@ -766,6 +722,8 @@ mod tests { use crate::handler::exit::ExitHandler; use crate::handler::start::StartHandler; use crate::handler::HandlerRegistry; + use crate::operations::resume; + use crate::records::CheckpointExt; const MINIMAL_DOT: &str = r#"digraph Test { graph [goal="Build feature"] diff --git a/lib/crates/fabro-workflows/src/operations/validate.rs b/lib/crates/fabro-workflows/src/operations/validate.rs new file mode 100644 index 000000000..18a0d2dcf --- /dev/null +++ b/lib/crates/fabro-workflows/src/operations/validate.rs @@ -0,0 +1,38 @@ +use std::path::PathBuf; + +use fabro_config::FabroSettings; + +use crate::error::FabroError; +use crate::pipeline::Validated; +use crate::transforms::Transform; + +use super::create::preprocess_and_validate; +use super::source::{resolve_workflow, ResolveWorkflowInput, WorkflowInput}; + +pub struct ValidateInput { + pub workflow: WorkflowInput, + pub settings: FabroSettings, + pub cwd: PathBuf, + pub custom_transforms: Vec>, +} + +/// Parse, transform, and validate a DOT source string. +/// +/// Returns `Validated` even when validation produced errors. Call +/// `validated.raise_on_errors()` if the caller wants to fail fast. +pub fn validate(input: ValidateInput) -> Result { + let resolved = resolve_workflow(ResolveWorkflowInput { + workflow: input.workflow, + settings: input.settings, + cwd: input.cwd, + }) + .map_err(|err| FabroError::Parse(err.to_string()))?; + + preprocess_and_validate( + &resolved.raw_source, + resolved.base_dir, + input.custom_transforms, + Some(&resolved.settings), + resolved.goal_override.as_deref(), + ) +}