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 1

Add workflow persist stage

Browse source
This commit is contained in:
Bryan Helmkamp 2026-03-25 13:46:34 -04:00
parent 974fa5519f
commit b57bc52416
No known key found for this signature in database
11 changed files with 798 additions and 280 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

55
lib/crates/fabro-api/src/server.rs
View file

@ -25,7 +25,7 @@ use fabro_workflows::context::Context;
use fabro_workflows::event::{EventEmitter, WorkflowRunEvent};
use fabro_workflows::handler::HandlerRegistry;
use fabro_workflows::operations::{self, CreateOptions};
use fabro_workflows::pipeline::{self, InitOptions};
use fabro_workflows::pipeline::{self, InitOptions, PersistOptions, Persisted};
use fabro_workflows::records::Checkpoint;
use fabro_workflows::run_settings::LifecycleConfig;
use fabro_workflows::run_settings::RunSettings;
@ -67,7 +67,6 @@ impl<T: serde::Serialize> ListResponse<T> {
/// Snapshot of a managed run.
struct ManagedRun {
dot_source: String,
graph: fabro_graphviz::graph::Graph,
status: RunStatus,
error: Option<String>,
created_at: chrono::DateTime<chrono::Utc>,
@ -476,14 +475,13 @@ async fn start_run(
State(state): State<Arc<AppState>>,
Json(req): Json<StartRunRequest>,
) -> Response {
// Parse the DOT source
let graph = match operations::create(&req.dot_source, CreateOptions::default()) {
// Parse and persist the DOT source.
let validated = match operations::create(&req.dot_source, CreateOptions::default()) {
Ok(validated) => {
if let Err(e) = validated.raise_on_errors() {
return ApiError::bad_request(e.to_string()).into_response();
}
let (graph, _, _) = validated.into_parts();
graph
validated
}
Err(e) => {
return ApiError::bad_request(e.to_string()).into_response();
@ -495,13 +493,6 @@ async fn start_run(
let created_at = chrono::Utc::now();
let run_dir = std::env::temp_dir().join(format!("fabro-{}", uuid::Uuid::new_v4()));
if let Err(err) = std::fs::create_dir_all(&run_dir) {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("Failed to create run directory: {err}"),
)
.into_response();
}
let run_record = fabro_workflows::records::RunRecord {
run_id: run_id.clone(),
@ -515,7 +506,7 @@ async fn start_run(
}),
..Default::default()
},
graph: graph.clone(),
graph: validated.graph().clone(),
workflow_slug: None,
working_directory: std::env::current_dir()
.unwrap_or_else(|_| std::path::PathBuf::from(".")),
@ -523,10 +514,16 @@ async fn start_run(
base_branch: None,
labels: std::collections::HashMap::new(),
};
if let Err(err) = run_record.save(&run_dir) {
if let Err(err) = fabro_workflows::pipeline::persist(
validated,
PersistOptions {
run_dir: run_dir.clone(),
run_record,
},
) {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("Failed to persist run record: {err}"),
format!("Failed to persist run state: {err}"),
)
.into_response();
}
@ -537,7 +534,6 @@ async fn start_run(
run_id.clone(),
ManagedRun {
dot_source: req.dot_source,
graph,
status: RunStatus::Queued,
error: None,
created_at,
@ -570,7 +566,7 @@ async fn start_run(
/// Execute a single run: transitions queued → starting → running → completed/failed/cancelled.
async fn execute_run(state: Arc<AppState>, run_id: String) {
// Transition to Starting and set up cancel infrastructure
let (cancel_rx, graph, run_dir) = {
let (cancel_rx, run_dir) = {
let mut runs = state.runs.lock().expect("runs lock poisoned");
let managed_run = match runs.get_mut(&run_id) {
Some(r) if r.status == RunStatus::Queued => r,
@ -590,7 +586,7 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
managed_run.cancel_token = Some(Arc::clone(&cancel_token));
managed_run.event_tx = Some(event_tx);
(cancel_rx, managed_run.graph.clone(), run_dir)
(cancel_rx, run_dir)
};
// Create interviewer, sandbox, engine (this is the "provisioning" phase)
@ -640,20 +636,21 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
}
}
let run_record = match fabro_workflows::records::RunRecord::load(&run_dir) {
Ok(r) => r,
let persisted = match Persisted::load(&run_dir) {
Ok(persisted) => persisted,
Err(e) => {
tracing::error!(run_id = %run_id, error = %e, "Failed to load RunRecord");
tracing::error!(run_id = %run_id, error = %e, "Failed to load persisted run");
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
managed_run.status = RunStatus::Failed;
managed_run.error = Some(format!("Failed to load run record: {e}"));
managed_run.error = Some(format!("Failed to load persisted run: {e}"));
managed_run.event_tx = None;
}
state.scheduler_notify.notify_one();
return;
}
};
let run_record = persisted.run_record().clone();
let config = RunSettings {
config: run_record.config,
run_dir: run_dir.clone(),
@ -662,10 +659,10 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
run_id: run_id.clone(),
labels: run_record.labels,
git_author: state.git_author.clone(),
workflow_slug: None,
workflow_slug: run_record.workflow_slug,
github_app: None,
base_branch: None,
host_repo_path: None,
base_branch: run_record.base_branch,
host_repo_path: run_record.host_repo_path.map(Into::into),
git: None,
};
@ -673,19 +670,15 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
let emitter = Arc::clone(&emitter);
let sandbox = Arc::clone(&sandbox);
let registry = Arc::new(registry);
let graph = graph.clone();
let run_dir = run_dir.clone();
let run_id = run_id.clone();
let config = config.clone();
let hooks = state.hooks.clone();
let dry_run = state.dry_run;
async move {
let validated = operations::create_from_graph(graph, String::new());
let initialized = pipeline::initialize(
validated,
persisted,
InitOptions {
run_id,
run_dir,
dry_run,
emitter,
sandbox,

35
lib/crates/fabro-cli/src/commands/create.rs
View file

@ -3,6 +3,7 @@ use std::path::PathBuf;
use chrono::Utc;
use fabro_config::config::FabroConfig;
use fabro_sandbox::SandboxProvider;
use fabro_workflows::pipeline::PersistOptions;
use fabro_workflows::records::RunRecord;
use super::run::{
@ -77,7 +78,6 @@ pub async fn create_run(
let dot_source = prep.source().to_string();
let graph = prep.graph().clone();
// Create run directory
let run_id = args
.run_id
.clone()
@ -86,20 +86,6 @@ pub async fn create_run(
.run_dir
.clone()
.unwrap_or_else(|| default_run_dir(&run_id, args.dry_run));
tokio::fs::create_dir_all(&run_dir).await?;
// Write essential files
tokio::fs::write(cached_graph_path(&run_dir), &dot_source).await?;
tokio::fs::write(run_dir.join("id.txt"), &run_id).await?;
std::fs::File::create(run_dir.join("progress.jsonl"))?;
fabro_workflows::run_status::write_run_status(
&run_dir,
fabro_workflows::run_status::RunStatus::Submitted,
None,
);
// Copy the original workflow TOML into the run dir as a debug artifact.
write_run_config_snapshot(&run_dir, prep.workflow_toml_path.as_deref()).await?;
// Build normalized config and RunRecord
let working_directory = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
@ -134,7 +120,24 @@ pub async fn create_run(
base_branch,
labels,
};
record.save(&run_dir)?;
fabro_workflows::pipeline::persist(
prep.validated,
PersistOptions {
run_dir: run_dir.clone(),
run_record: record,
},
)?;
// Write CLI-owned debug and status artifacts after the run has been persisted.
tokio::fs::write(cached_graph_path(&run_dir), &dot_source).await?;
tokio::fs::write(run_dir.join("id.txt"), &run_id).await?;
std::fs::File::create(run_dir.join("progress.jsonl"))?;
fabro_workflows::run_status::write_run_status(
&run_dir,
fabro_workflows::run_status::RunStatus::Submitted,
None,
);
write_run_config_snapshot(&run_dir, prep.workflow_toml_path.as_deref()).await?;
Ok((run_id, run_dir))
}

224
lib/crates/fabro-cli/src/commands/resume.rs
View file

@ -18,7 +18,7 @@ use fabro_workflows::operations::{
};
use fabro_workflows::outcome::StageStatus;
use fabro_workflows::pipeline::{
build_conclusion, classify_engine_result, persist_terminal_outcome,
build_conclusion, classify_engine_result, persist_terminal_outcome, PersistOptions, Persisted,
};
use fabro_workflows::records::Checkpoint;
use fabro_workflows::records::RunRecord;
@ -97,7 +97,7 @@ pub struct ResumeArgs {
/// Intermediate state produced by the two resolution paths (checkpoint-file vs. git-branch).
struct ResumeContext {
checkpoint: Checkpoint,
validated: fabro_workflows::pipeline::Validated,
persisted: Persisted,
run_id: String,
run_dir: PathBuf,
run_cfg: Option<FabroConfig>,
@ -227,48 +227,56 @@ async fn prepare_from_checkpoint(
.run_dir
.clone()
.unwrap_or_else(|| default_run_dir(&run_id, args.dry_run));
let labels = args
.label
.iter()
.filter_map(|s| s.split_once('='))
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect();
let working_directory = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
let cli_flags = super::create::CliFlags {
dry_run: args.dry_run,
auto_approve: args.auto_approve,
no_retro: args.no_retro,
verbose: args.verbose,
preserve_sandbox: args.preserve_sandbox,
};
let normalized = super::create::normalize_config(
run_cfg.as_ref(),
&prepared_run_defaults,
&prepared_model,
prepared_provider.as_deref(),
sandbox_provider,
&graph,
cli_flags,
);
let persisted = fabro_workflows::pipeline::persist(
validated,
PersistOptions {
run_dir: run_dir.clone(),
run_record: fabro_workflows::records::RunRecord {
run_id: run_id.clone(),
created_at: chrono::Utc::now(),
config: normalized,
graph: graph.clone(),
workflow_slug: workflow_slug.clone(),
working_directory: working_directory.clone(),
host_repo_path: Some(working_directory.to_string_lossy().to_string()),
base_branch: None,
labels,
},
},
)?;
let run_cfg: Option<FabroConfig> = Some(persisted.run_record().config.clone());
let settings_config = persisted.run_record().config.clone();
tokio::fs::create_dir_all(&run_dir).await?;
fabro_util::run_log::activate(&run_dir.join("cli.log"))
.context("Failed to activate per-run log")?;
let status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?;
tokio::fs::write(cached_graph_path(&run_dir), &source).await?;
let run_cfg: Option<FabroConfig> = run_cfg;
write_run_config_snapshot(&run_dir, workflow_toml_path.as_deref()).await?;
// Write RunRecord for the resumed run
let settings_config = {
let working_directory = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
let cli_flags = super::create::CliFlags {
dry_run: args.dry_run,
auto_approve: args.auto_approve,
no_retro: args.no_retro,
verbose: args.verbose,
preserve_sandbox: args.preserve_sandbox,
};
let normalized = super::create::normalize_config(
run_cfg.as_ref(),
&prepared_run_defaults,
&prepared_model,
prepared_provider.as_deref(),
sandbox_provider,
&graph,
cli_flags,
);
let record = fabro_workflows::records::RunRecord {
run_id: run_id.clone(),
created_at: chrono::Utc::now(),
config: normalized.clone(),
graph: graph.clone(),
workflow_slug: workflow_slug.clone(),
working_directory: working_directory.clone(),
host_repo_path: Some(working_directory.to_string_lossy().to_string()),
base_branch: None,
labels: std::collections::HashMap::new(),
};
let _ = record.save(&run_dir);
normalized
};
let original_cwd = std::env::current_dir()?;
let emitter = Arc::new(EventEmitter::new());
@ -482,7 +490,7 @@ async fn prepare_from_checkpoint(
Ok(ResumeContext {
checkpoint,
validated,
persisted,
run_id,
run_dir,
run_cfg,
@ -553,7 +561,7 @@ async fn prepare_from_branch(
.or_else(|| repo_info.as_ref().and_then(|(_, branch)| branch.clone()));
let base_sha = start_record.as_ref().and_then(|s| s.base_sha.clone());
let (validated, graph_source, run_cfg, mut sandbox_provider, workflow_slug) =
let (mut validated, mut graph_source, mut run_cfg, mut sandbox_provider, mut workflow_slug) =
if let Some(ref workflow_path) = args.workflow {
let prepared = prepare_workflow_with_project_config(
&resume_as_run_args(args, workflow_path.clone()),
@ -596,15 +604,6 @@ async fn prepare_from_branch(
} else {
bail!("no run.json found on metadata branch for run {run_id}");
};
let graph = validated.graph().clone();
eprintln!(
"{} {} from branch {} ({})",
styles.bold.apply_to("Resuming workflow:"),
graph.name,
styles.dim.apply_to(&run_branch),
run_id,
);
// Set up logs directory — reuse existing run dir for this run_id to avoid
// "ambiguous prefix" errors when the resume happens on a different day.
@ -618,56 +617,94 @@ async fn prepare_from_branch(
};
tokio::fs::create_dir_all(&run_dir).await?;
let run_dir = tokio::fs::canonicalize(&run_dir).await.unwrap_or(run_dir);
if args.workflow.is_none() {
if let Ok(loaded) = Persisted::load(&run_dir) {
graph_source = loaded.source().to_string();
run_cfg = Some(loaded.run_record().config.clone());
workflow_slug = loaded.run_record().workflow_slug.clone();
validated = create_from_graph(loaded.graph().clone(), loaded.source().to_string());
if !args.dry_run {
if let Some(provider) = loaded
.run_record()
.config
.sandbox
.as_ref()
.and_then(|s| s.provider.as_deref())
.and_then(|s| s.parse::<SandboxProvider>().ok())
{
sandbox_provider = args.sandbox.map(Into::into).unwrap_or(provider);
}
}
}
}
let graph = validated.graph().clone();
eprintln!(
"{} {} from branch {} ({})",
styles.bold.apply_to("Resuming workflow:"),
graph.name,
styles.dim.apply_to(&run_branch),
run_id,
);
let (model_str, provider_str) = resolve_model_provider(
args.model.as_deref(),
args.provider.as_deref(),
run_cfg.as_ref(),
run_defaults,
&graph,
);
let cli_flags = super::create::CliFlags {
dry_run: args.dry_run,
auto_approve: args.auto_approve,
no_retro: args.no_retro,
verbose: args.verbose,
preserve_sandbox: args.preserve_sandbox,
};
let normalized = super::create::normalize_config(
run_cfg.as_ref(),
run_defaults,
&model_str,
provider_str.as_deref(),
sandbox_provider,
&graph,
cli_flags,
);
let persisted = fabro_workflows::pipeline::persist(
validated,
PersistOptions {
run_dir: run_dir.clone(),
run_record: fabro_workflows::records::RunRecord {
run_id: run_id.clone(),
created_at: chrono::Utc::now(),
config: normalized,
graph: graph.clone(),
workflow_slug: workflow_slug.clone(),
working_directory: resume_repo_path.clone(),
host_repo_path: Some(resume_repo_path.to_string_lossy().to_string()),
base_branch: detected_base_branch.clone(),
labels: args
.label
.iter()
.filter_map(|s| s.split_once('='))
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
},
},
)?;
let run_cfg: Option<FabroConfig> = Some(persisted.run_record().config.clone());
let settings_config = persisted.run_record().config.clone();
fabro_util::run_log::activate(&run_dir.join("cli.log"))
.context("Failed to activate per-run log")?;
let status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?;
if !graph_source.is_empty() {
tokio::fs::write(cached_graph_path(&run_dir), &graph_source).await?;
}
let run_cfg: Option<FabroConfig> = run_cfg;
// Git-branch resume: no original TOML available, skip debug snapshot.
write_run_config_snapshot(&run_dir, None).await?;
// Write RunRecord for the resumed run
let settings_config = {
let (model_str, provider_str) = resolve_model_provider(
args.model.as_deref(),
args.provider.as_deref(),
run_cfg.as_ref(),
run_defaults,
&graph,
);
let cli_flags = super::create::CliFlags {
dry_run: args.dry_run,
auto_approve: args.auto_approve,
no_retro: args.no_retro,
verbose: args.verbose,
preserve_sandbox: args.preserve_sandbox,
};
let normalized = super::create::normalize_config(
run_cfg.as_ref(),
run_defaults,
&model_str,
provider_str.as_deref(),
sandbox_provider,
&graph,
cli_flags,
);
let record = fabro_workflows::records::RunRecord {
run_id: run_id.clone(),
created_at: chrono::Utc::now(),
config: normalized.clone(),
graph: graph.clone(),
workflow_slug: workflow_slug.clone(),
working_directory: resume_repo_path.clone(),
host_repo_path: Some(resume_repo_path.to_string_lossy().to_string()),
base_branch: detected_base_branch.clone(),
labels: std::collections::HashMap::new(),
};
let _ = record.save(&run_dir);
normalized
};
let emitter = Arc::new(EventEmitter::new());
// Resolve devcontainer BEFORE sandbox creation (mirrors run_command) so that
@ -914,7 +951,7 @@ async fn prepare_from_branch(
Ok(ResumeContext {
checkpoint,
validated,
persisted,
run_id,
run_dir,
run_cfg,
@ -942,7 +979,7 @@ async fn run_resumed(
) -> anyhow::Result<()> {
let ResumeContext {
checkpoint,
validated,
persisted,
run_id,
run_dir,
mut run_cfg,
@ -959,7 +996,7 @@ async fn run_resumed(
github_app,
mut status_guard,
} = ctx;
let graph = validated.graph().clone();
let graph = persisted.graph().clone();
// Create progress UI (verbose mode shows detailed turn/tool counts and token usage)
let is_tty = std::io::stderr().is_terminal();
@ -1256,11 +1293,10 @@ async fn run_resumed(
settings.pull_request().cloned()
};
let started = start(
validated,
persisted,
StartOptions {
init: fabro_workflows::pipeline::InitOptions {
run_id: run_id.clone(),
run_dir: run_dir.clone(),
dry_run: dry_run_mode,
emitter: Arc::clone(&emitter),
sandbox: Arc::clone(&sandbox),

167
lib/crates/fabro-cli/src/commands/run.rs
View file

@ -21,13 +21,12 @@ use fabro_workflows::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
use fabro_workflows::git::GitSyncStatus;
use fabro_workflows::handler::default_registry;
use fabro_workflows::handler::llm::{AgentApiBackend, AgentCliBackend, BackendRouter};
use fabro_workflows::operations::{
create_from_graph, start, StartFinalizeConfig, StartOptions, StartRetroConfig,
};
use fabro_workflows::operations::{start, StartFinalizeConfig, StartOptions, StartRetroConfig};
use fabro_workflows::outcome::StageStatus;
use fabro_workflows::outcome::{compute_stage_cost, format_cost};
use fabro_workflows::pipeline::{
build_conclusion, classify_engine_result, persist_terminal_outcome,
build_conclusion, classify_engine_result, persist_terminal_outcome, PersistOptions, Persisted,
Validated,
};
use fabro_workflows::records::Checkpoint;
use fabro_workflows::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings};
@ -571,6 +570,11 @@ impl PreparedWorkflow {
}
}
enum WorkflowState {
Validated(Validated),
Persisted(Box<Persisted>),
}
/// Resolve config, parse/validate the workflow graph, and resolve sandbox + model.
///
/// Shared between `create_run` (which only persists the spec) and
@ -747,7 +751,7 @@ pub(crate) fn prepare_workflow_with_project_config(
/// Pre-prepared run state, used to skip workflow preparation in `run_command_impl`.
struct RecordBasedRun {
validated: fabro_workflows::pipeline::Validated,
workflow: WorkflowState,
raw_source: String,
run_cfg: Option<FabroConfig>,
sandbox_provider: SandboxProvider,
@ -763,13 +767,14 @@ struct RecordBasedRun {
///
/// Used by `run_engine_entrypoint` for detached runs that already have a RunRecord on disk.
pub async fn run_from_record(
record: fabro_workflows::records::RunRecord,
persisted: Persisted,
run_dir: PathBuf,
run_defaults: FabroConfig,
styles: &'static Styles,
github_app: Option<fabro_github::GitHubAppCredentials>,
git_author: fabro_workflows::git::GitAuthor,
) -> anyhow::Result<()> {
let record = persisted.run_record().clone();
let sandbox_provider = record
.config
.sandbox
@ -792,8 +797,8 @@ pub async fn run_from_record(
.filter(|s| !s.is_empty());
let record_run = RecordBasedRun {
workflow: WorkflowState::Persisted(Box::new(persisted)),
raw_source: String::new(), // Raw DOT provenance is best-effort for record-based runs
validated: create_from_graph(record.graph.clone(), String::new()),
run_cfg: Some(record.config.clone()),
sandbox_provider,
model: model.clone(),
@ -817,7 +822,7 @@ pub async fn run_from_record(
sandbox: Some(CliSandboxProvider::from(sandbox_provider)),
label: record
.labels
.into_iter()
.iter()
.map(|(k, v)| format!("{k}={v}"))
.collect(),
no_retro: record.config.no_retro_enabled(),
@ -828,7 +833,7 @@ pub async fn run_from_record(
.and_then(|s| s.preserve)
.unwrap_or(false),
detach: false,
run_id: Some(record.run_id),
run_id: Some(record.run_id.clone()),
};
run_command_impl(args, styles, github_app, git_author, Some(record_run)).await
@ -859,7 +864,7 @@ pub async fn run_command(
} = prepare_workflow(&args, run_defaults, styles, false)?;
let record_run = RecordBasedRun {
validated,
workflow: WorkflowState::Validated(validated),
raw_source,
run_cfg,
sandbox_provider,
@ -881,7 +886,7 @@ async fn run_command_impl(
record_run: Option<RecordBasedRun>,
) -> anyhow::Result<()> {
let (
validated,
workflow,
raw_source,
mut run_cfg,
sandbox_provider,
@ -892,7 +897,7 @@ async fn run_command_impl(
workflow_toml_path,
) = match record_run {
Some(rr) => (
rr.validated,
rr.workflow,
rr.raw_source,
rr.run_cfg,
rr.sandbox_provider,
@ -904,10 +909,13 @@ async fn run_command_impl(
),
None => unreachable!("run_command_impl always receives a RecordBasedRun"),
};
let graph = validated.graph().clone();
let graph = match &workflow {
WorkflowState::Validated(validated) => validated.graph().clone(),
WorkflowState::Persisted(persisted) => persisted.graph().clone(),
};
// For record-based runs from run_from_record, workflow is None (preparation was skipped).
let from_record = args.workflow.is_none();
// For record-based runs from run_from_record, the workflow has already been persisted.
let from_record = matches!(&workflow, WorkflowState::Persisted(_)) && args.workflow.is_none();
// Collect setup commands — they'll be run inside the sandbox
let setup_commands: Vec<String> = run_cfg
@ -955,26 +963,61 @@ async fn run_command_impl(
let run_dir = args
.run_dir
.unwrap_or_else(|| default_run_dir(&run_id, dry_run_flag));
tokio::fs::create_dir_all(&run_dir).await?;
let cached_run_restart = if from_record {
// Record-based runs already have RunRecord on disk — skip re-writing.
true
false
} else {
let workflow_path = args.workflow.as_ref().unwrap();
is_cached_run_restart(workflow_path, &run_dir)
};
let existing_record = if cached_run_restart {
fabro_workflows::records::RunRecord::load(&run_dir).ok()
} else {
None
};
let workflow_slug = if cached_run_restart {
existing_record
.as_ref()
.and_then(|r| r.workflow_slug.clone())
} else {
prepared_workflow_slug
let persisted = match (cached_run_restart, workflow) {
(true, _) => Persisted::load(&run_dir)?,
(false, WorkflowState::Persisted(persisted)) => *persisted,
(false, WorkflowState::Validated(validated)) => {
let cli_flags = super::create::CliFlags {
dry_run: dry_run_flag,
auto_approve: auto_approve_flag,
no_retro: no_retro_flag,
verbose: verbose_flag,
preserve_sandbox: preserve_sandbox_flag,
};
let normalized_config = super::create::normalize_config(
run_cfg.as_ref(),
&run_defaults,
&model,
provider.as_deref(),
sandbox_provider,
validated.graph(),
cli_flags,
);
let run_record = fabro_workflows::records::RunRecord {
run_id: run_id.clone(),
created_at: chrono::Utc::now(),
config: normalized_config,
graph: validated.graph().clone(),
workflow_slug: prepared_workflow_slug.clone(),
working_directory: original_cwd.clone(),
host_repo_path: Some(original_cwd.to_string_lossy().to_string()),
base_branch: detected_base_branch.clone(),
labels: label_vec
.iter()
.filter_map(|s| s.split_once('='))
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
};
fabro_workflows::pipeline::persist(
validated,
PersistOptions {
run_dir: run_dir.clone(),
run_record,
},
)?
}
};
run_cfg = Some(persisted.run_record().config.clone());
let workflow_slug = persisted.run_record().workflow_slug.clone();
tokio::fs::create_dir_all(&run_dir).await?;
fabro_util::run_log::activate(&run_dir.join("cli.log"))
.context("Failed to activate per-run log")?;
if !from_record && !raw_source.is_empty() {
@ -988,47 +1031,7 @@ async fn run_command_impl(
write_run_config_snapshot(&run_dir, workflow_toml_path.as_deref()).await?;
}
// Write RunRecord and compute normalized config for RunSettings
let settings_config = if !cached_run_restart {
let cli_flags = super::create::CliFlags {
dry_run: dry_run_flag,
auto_approve: auto_approve_flag,
no_retro: no_retro_flag,
verbose: verbose_flag,
preserve_sandbox: preserve_sandbox_flag,
};
let normalized_config = super::create::normalize_config(
run_cfg.as_ref(),
&run_defaults,
&model,
provider.as_deref(),
sandbox_provider,
&graph,
cli_flags,
);
let record = fabro_workflows::records::RunRecord {
run_id: run_id.clone(),
created_at: chrono::Utc::now(),
config: normalized_config.clone(),
graph: graph.clone(),
workflow_slug: workflow_slug.clone(),
working_directory: original_cwd.clone(),
host_repo_path: Some(original_cwd.to_string_lossy().to_string()),
base_branch: detected_base_branch.clone(),
labels: label_vec
.iter()
.filter_map(|s| s.split_once('='))
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
};
record.save(&run_dir)?;
normalized_config
} else {
existing_record
.as_ref()
.map(|r| r.config.clone())
.unwrap_or_default()
};
let settings_config = persisted.run_record().config.clone();
// Now resolve ${env.VARNAME} references for runtime use.
if let Some(ref mut cfg) = run_cfg {
@ -1701,21 +1704,20 @@ async fn run_command_impl(
cancel_token: None,
dry_run: dry_run_mode,
run_id: run_id.clone(),
labels: label_vec
.iter()
.filter_map(|s| s.split_once('='))
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
labels: persisted.run_record().labels.clone(),
git_author: git_author.clone(),
workflow_slug: workflow_slug.clone(),
github_app: github_app.clone(),
base_branch: existing_record
.as_ref()
.and_then(|r| r.base_branch.clone())
base_branch: persisted
.run_record()
.base_branch
.clone()
.or(detected_base_branch),
host_repo_path: existing_record
.as_ref()
.and_then(|r| r.host_repo_path.as_deref().map(PathBuf::from))
host_repo_path: persisted
.run_record()
.host_repo_path
.as_deref()
.map(PathBuf::from)
.or_else(|| Some(original_cwd.clone())),
git,
};
@ -1745,11 +1747,10 @@ async fn run_command_impl(
config.pull_request().cloned()
};
let started = start(
validated,
persisted,
StartOptions {
init: fabro_workflows::pipeline::InitOptions {
run_id: run_id.clone(),
run_dir: run_dir.clone(),
dry_run: dry_run_mode,
emitter: Arc::clone(&emitter),
sandbox: Arc::clone(&sandbox),

24
lib/crates/fabro-cli/src/main.rs
View file

@ -324,10 +324,10 @@ async fn run_engine_entrypoint(
cli_config.git_author().and_then(|a| a.email.clone()),
);
let record = match fabro_workflows::records::RunRecord::load(&run_dir) {
Ok(record) => record,
let persisted = match fabro_workflows::pipeline::Persisted::load(&run_dir) {
Ok(persisted) => persisted,
Err(err) => {
let anyhow_err: anyhow::Error = anyhow::anyhow!("Failed to load run record: {err}");
let anyhow_err: anyhow::Error = anyhow::anyhow!("Failed to load persisted run: {err}");
let _ = commands::detached_support::persist_detached_failure(
&run_dir,
"bootstrap",
@ -338,12 +338,14 @@ async fn run_engine_entrypoint(
}
};
if let Err(err) = std::env::set_current_dir(&record.working_directory).map_err(|e| {
anyhow::anyhow!(
"Failed to set working directory to {}: {e}",
record.working_directory.display()
)
}) {
if let Err(err) =
std::env::set_current_dir(&persisted.run_record().working_directory).map_err(|e| {
anyhow::anyhow!(
"Failed to set working directory to {}: {e}",
persisted.run_record().working_directory.display()
)
})
{
let _ = commands::detached_support::persist_detached_failure(
&run_dir,
"bootstrap",
@ -353,10 +355,10 @@ async fn run_engine_entrypoint(
return Err(err);
}
// Use run_from_record: loads config + graph directly from RunRecord,
// Use run_from_record: loads config + graph directly from persisted state,
// skipping prepare_workflow() entirely. No TOML/DOT re-parsing needed.
match commands::run::run_from_record(
record,
persisted,
run_dir.clone(),
cli_config,
styles,

38
lib/crates/fabro-workflows/src/operations/start.rs
View file

@ -4,7 +4,7 @@ use std::time::{Duration, Instant};
use crate::error::FabroError;
use crate::event::WorkflowRunEvent;
use crate::outcome::StageStatus;
use crate::pipeline::{self, FinalizeOptions, Finalized, InitOptions, RetroOptions, Validated};
use crate::pipeline::{self, FinalizeOptions, Finalized, InitOptions, Persisted, RetroOptions};
pub struct StartRetroConfig {
pub enabled: bool,
@ -34,8 +34,8 @@ pub struct Started {
pub retro_duration: Duration,
}
/// Run a validated workflow through initialize, execute, retro, and finalize.
pub async fn start(validated: Validated, options: StartOptions) -> Result<Started, FabroError> {
/// Run a persisted workflow through initialize, execute, retro, and finalize.
pub async fn start(persisted: Persisted, options: StartOptions) -> Result<Started, FabroError> {
let preserve_sandbox = options.finalize.preserve_sandbox;
let sandbox_for_cleanup = Arc::clone(&options.init.sandbox);
let cleanup_guard = scopeguard::guard((), move |()| {
@ -49,7 +49,7 @@ pub async fn start(validated: Validated, options: StartOptions) -> Result<Starte
}
});
let initialized = pipeline::initialize(validated, options.init).await?;
let initialized = pipeline::initialize(persisted, options.init).await?;
let last_git_sha: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
{
@ -123,6 +123,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};
@ -135,6 +136,8 @@ mod tests {
use crate::handler::start::StartHandler;
use crate::handler::{Handler, HandlerRegistry};
use crate::outcome::Outcome;
use crate::pipeline::PersistOptions;
use crate::records::RunRecord;
use crate::run_settings::{LifecycleConfig, RunSettings};
const MINIMAL_DOT: &str = r#"digraph Test {
@ -340,11 +343,29 @@ mod tests {
}
}
fn validated_workflow(dot: &str) -> Validated {
fn persisted_workflow(dot: &str, run_dir: &std::path::Path) -> Persisted {
let validated =
crate::operations::create(dot, crate::operations::CreateOptions::default()).unwrap();
validated.raise_on_errors().unwrap();
validated
let graph = validated.graph().clone();
crate::pipeline::persist(
validated,
PersistOptions {
run_dir: run_dir.to_path_buf(),
run_record: RunRecord {
run_id: "run-test".to_string(),
created_at: Utc::now(),
config: FabroConfig::default(),
graph,
workflow_slug: Some("test".to_string()),
working_directory: std::env::current_dir().unwrap(),
host_repo_path: Some(std::env::current_dir().unwrap().display().to_string()),
base_branch: Some("main".to_string()),
labels: HashMap::new(),
},
},
)
.unwrap()
}
fn test_settings(run_dir: &std::path::Path) -> RunSettings {
@ -383,7 +404,6 @@ mod tests {
StartOptions {
init: InitOptions {
run_id: "run-test".to_string(),
run_dir: run_dir.to_path_buf(),
dry_run: false,
emitter,
sandbox,
@ -433,7 +453,7 @@ mod tests {
let (sandbox, cleanup_count) = counting_sandbox();
let result = start(
validated_workflow(MINIMAL_DOT),
persisted_workflow(MINIMAL_DOT, &run_dir),
test_start_options(
&run_dir,
sandbox,
@ -465,7 +485,7 @@ mod tests {
Arc::new(LocalSandbox::new(std::env::current_dir().unwrap()));
let started = start(
validated_workflow(EMIT_DOT),
persisted_workflow(EMIT_DOT, &run_dir),
test_start_options(
&run_dir,
sandbox,

40
lib/crates/fabro-workflows/src/pipeline/execute/tests.rs
View file

@ -5,6 +5,7 @@ use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use chrono::Utc;
use fabro_agent::Sandbox;
use fabro_config::config::FabroConfig;
use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
@ -18,11 +19,10 @@ use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::handler::default_registry;
use crate::handler::start::StartHandler;
use crate::handler::{Handler as HandlerTrait, HandlerRegistry};
use crate::operations::create_from_graph;
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
use crate::pipeline::initialize;
use crate::pipeline::types::{InitOptions, Validated};
use crate::records::Checkpoint;
use crate::pipeline::types::{InitOptions, Persisted};
use crate::records::{Checkpoint, RunRecord};
use crate::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings};
use crate::test_support::run_graph;
@ -106,6 +106,31 @@ fn simple_validated_graph() -> (Graph, String) {
(graph, source)
}
fn persisted_workflow(graph: Graph, source: String, run_dir: &Path, run_id: &str) -> Persisted {
Persisted::new(
graph.clone(),
source,
vec![],
run_dir.to_path_buf(),
RunRecord {
run_id: run_id.to_string(),
created_at: Utc::now(),
config: FabroConfig::default(),
graph,
workflow_slug: Some("test".to_string()),
working_directory: std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")),
host_repo_path: Some(
std::env::current_dir()
.unwrap_or_else(|_| PathBuf::from("."))
.display()
.to_string(),
),
base_branch: Some("main".to_string()),
labels: HashMap::new(),
},
)
}
fn test_lifecycle(setup_commands: Vec<String>) -> LifecycleConfig {
LifecycleConfig {
setup_commands,
@ -118,12 +143,12 @@ fn test_lifecycle(setup_commands: Vec<String>) -> LifecycleConfig {
async fn execute_runs_start_to_exit_and_returns_final_context() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let (graph, source) = simple_validated_graph();
let initialized = initialize(
Validated::new(graph, source, vec![]),
persisted_workflow(graph, source, &run_dir, "run-test"),
InitOptions {
run_id: "run-test".to_string(),
run_dir: run_dir.clone(),
dry_run: false,
emitter: Arc::new(crate::event::EventEmitter::new()),
sandbox: Arc::new(fabro_agent::LocalSandbox::new(
@ -169,12 +194,11 @@ async fn run_with_lifecycle(
) -> Result<Outcome, FabroError> {
let run_dir = settings.run_dir.clone();
let run_id = settings.run_id.clone();
let validated = create_from_graph(graph.clone(), String::new());
std::fs::create_dir_all(&run_dir).unwrap();
let initialized = initialize(
validated,
persisted_workflow(graph.clone(), String::new(), &run_dir, &run_id),
InitOptions {
run_id,
run_dir,
dry_run: settings.dry_run,
emitter,
sandbox,

59
lib/crates/fabro-workflows/src/pipeline/initialize.rs
View file

@ -8,7 +8,7 @@ use crate::error::FabroError;
use crate::event::WorkflowRunEvent;
use crate::run_settings::GitCheckpointSettings;
use super::types::{InitOptions, Initialized, Validated};
use super::types::{InitOptions, Initialized, Persisted};
async fn run_hooks(
hook_runner: Option<&HookRunner>,
@ -28,17 +28,11 @@ async fn run_hooks(
///
/// Returns `FabroError` if sandbox preparation fails.
pub async fn initialize(
validated: Validated,
persisted: Persisted,
mut options: InitOptions,
) -> Result<Initialized, FabroError> {
let (graph, source, _diagnostics) = validated.into_parts();
// Create run directory and write graph
std::fs::create_dir_all(&options.run_dir)?;
if !source.is_empty() {
let graph_path = options.run_dir.join("graph.fabro");
std::fs::write(&graph_path, &source)?;
}
let (graph, source, _diagnostics, run_dir, _run_record) = persisted.into_parts();
options.run_settings.run_dir = run_dir;
let hook_runner = if options.hooks.hooks.is_empty() {
None
@ -196,13 +190,14 @@ mod tests {
use std::collections::HashMap;
use std::sync::Arc;
use chrono::Utc;
use fabro_config::config::FabroConfig;
use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
use fabro_interview::AutoApproveInterviewer;
use super::*;
use crate::handler::default_registry;
use crate::pipeline::types::Validated;
use crate::records::RunRecord;
use crate::run_settings::RunSettings;
fn simple_graph() -> (Graph, String) {
@ -246,12 +241,33 @@ mod tests {
}
}
fn test_persisted(graph: Graph, source: String, run_dir: &std::path::Path) -> Persisted {
Persisted::new(
graph.clone(),
source,
vec![],
run_dir.to_path_buf(),
RunRecord {
run_id: "run-test".to_string(),
created_at: Utc::now(),
config: FabroConfig::default(),
graph,
workflow_slug: Some("test".to_string()),
working_directory: std::env::current_dir().unwrap(),
host_repo_path: Some(std::env::current_dir().unwrap().display().to_string()),
base_branch: Some("main".to_string()),
labels: HashMap::new(),
},
)
}
#[tokio::test]
async fn initialize_prepares_sandbox_and_writes_graph() {
async fn initialize_prepares_sandbox_and_uses_persisted_run_dir() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let (graph, source) = simple_graph();
let validated = Validated::new(graph, source.clone(), vec![]);
let persisted = test_persisted(graph, source.clone(), &run_dir);
let emitter = Arc::new(crate::event::EventEmitter::new());
let sandbox = Arc::new(fabro_agent::LocalSandbox::new(
std::env::current_dir().unwrap(),
@ -259,10 +275,9 @@ mod tests {
let registry = Arc::new(default_registry(Arc::new(AutoApproveInterviewer), || None));
let initialized = initialize(
validated,
persisted,
InitOptions {
run_id: "run-test".to_string(),
run_dir: run_dir.clone(),
dry_run: false,
emitter,
sandbox,
@ -282,11 +297,8 @@ mod tests {
.await
.unwrap();
assert!(run_dir.join("graph.fabro").exists());
assert_eq!(
std::fs::read_to_string(run_dir.join("graph.fabro")).unwrap(),
source
);
assert_eq!(initialized.settings.run_dir, run_dir);
assert_eq!(initialized.source, source);
assert!(initialized.hook_runner.is_none());
assert_eq!(
initialized.env.get("TEST_KEY").map(String::as_str),
@ -298,8 +310,9 @@ mod tests {
async fn initialize_skips_empty_graph_source() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let (graph, _source) = simple_graph();
let validated = Validated::new(graph, String::new(), vec![]);
let persisted = test_persisted(graph, String::new(), &run_dir);
let emitter = Arc::new(crate::event::EventEmitter::new());
let sandbox = Arc::new(fabro_agent::LocalSandbox::new(
std::env::current_dir().unwrap(),
@ -307,10 +320,9 @@ mod tests {
let registry = Arc::new(default_registry(Arc::new(AutoApproveInterviewer), || None));
let initialized = initialize(
validated,
persisted,
InitOptions {
run_id: "run-test".to_string(),
run_dir: run_dir.clone(),
dry_run: false,
emitter,
sandbox,
@ -330,7 +342,6 @@ mod tests {
.await
.unwrap();
assert!(!run_dir.join("graph.fabro").exists());
assert!(initialized.source.is_empty());
}
}

2
lib/crates/fabro-workflows/src/pipeline/mod.rs
View file

@ -2,6 +2,7 @@ mod execute;
mod finalize;
mod initialize;
mod parse;
mod persist;
mod retro;
mod transform;
pub mod types;
@ -14,6 +15,7 @@ pub use finalize::{
};
pub use initialize::initialize;
pub use parse::parse;
pub use persist::persist;
pub use retro::{retro, run_retro};
pub use transform::transform;
pub use types::*;

333
lib/crates/fabro-workflows/src/pipeline/persist.rs Normal file
View file

@ -0,0 +1,333 @@
use std::path::Path;
use crate::error::FabroError;
use super::types::{PersistOptions, Persisted, Validated};
const GRAPH_FILE_NAME: &str = "graph.fabro";
/// PERSIST phase: create run directory, write graph.fabro and run.json to disk.
///
/// Overwrites `run_record.graph` with the validated graph before saving.
pub fn persist(validated: Validated, mut options: PersistOptions) -> Result<Persisted, FabroError> {
let (graph, source, diagnostics) = validated.into_parts();
options.run_record.graph = graph.clone();
std::fs::create_dir_all(&options.run_dir)?;
if !source.is_empty() {
std::fs::write(options.run_dir.join(GRAPH_FILE_NAME), &source)?;
}
options.run_record.save(&options.run_dir)?;
Ok(Persisted::new(
graph,
source,
diagnostics,
options.run_dir,
options.run_record,
))
}
/// Load a previously persisted run from disk.
///
/// `run.json` is authoritative for graph + config; `graph.fabro` provides the
/// original DOT source string when present.
pub(crate) fn load(run_dir: &Path) -> Result<Persisted, FabroError> {
let run_record = crate::records::RunRecord::load(run_dir)?;
let graph = run_record.graph.clone();
let source = match std::fs::read_to_string(run_dir.join(GRAPH_FILE_NAME)) {
Ok(source) => source,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
Err(err) => return Err(err.into()),
};
Ok(Persisted::new(
graph,
source,
Vec::new(),
run_dir.to_path_buf(),
run_record,
))
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::path::PathBuf;
use chrono::Utc;
use fabro_config::config::FabroConfig;
use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
use super::*;
use crate::records::RunRecord;
fn graph_and_source() -> (Graph, String) {
let source = r#"digraph test {
graph [goal="Ship feature"];
start [shape=Mdiamond];
exit [shape=Msquare];
start -> exit;
}"#
.to_string();
let mut graph = Graph::new("test");
graph.attrs.insert(
"goal".to_string(),
AttrValue::String("Ship feature".to_string()),
);
let mut start = Node::new("start");
start.attrs.insert(
"shape".to_string(),
AttrValue::String("Mdiamond".to_string()),
);
graph.nodes.insert("start".to_string(), start);
let mut exit = Node::new("exit");
exit.attrs.insert(
"shape".to_string(),
AttrValue::String("Msquare".to_string()),
);
graph.nodes.insert("exit".to_string(), exit);
graph.edges.push(Edge::new("start", "exit"));
(graph, source)
}
fn different_graph() -> Graph {
let mut graph = Graph::new("different");
let mut start = Node::new("start");
start.attrs.insert(
"shape".to_string(),
AttrValue::String("Mdiamond".to_string()),
);
graph.nodes.insert("start".to_string(), start);
graph
}
fn sample_record(graph: Graph) -> RunRecord {
RunRecord {
run_id: "run-123".to_string(),
created_at: Utc::now(),
config: FabroConfig {
dry_run: Some(true),
verbose: Some(true),
..Default::default()
},
graph,
workflow_slug: Some("ship".to_string()),
working_directory: PathBuf::from("/tmp/project"),
host_repo_path: Some("/tmp/project".to_string()),
base_branch: Some("main".to_string()),
labels: HashMap::from([
("env".to_string(), "test".to_string()),
("team".to_string(), "workflow".to_string()),
]),
}
}
#[test]
fn persist_creates_run_dir_and_writes_graph_and_record() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, source) = graph_and_source();
let persisted = persist(
Validated::new(graph.clone(), source.clone(), vec![]),
PersistOptions {
run_dir: run_dir.clone(),
run_record: sample_record(different_graph()),
},
)
.unwrap();
assert!(run_dir.is_dir());
assert_eq!(
std::fs::read_to_string(run_dir.join(GRAPH_FILE_NAME)).unwrap(),
source
);
assert!(run_dir.join(RunRecord::file_name()).exists());
assert_eq!(persisted.run_dir(), run_dir.as_path());
assert_eq!(
serde_json::to_value(persisted.run_record().graph.clone()).unwrap(),
serde_json::to_value(graph).unwrap()
);
}
#[test]
fn persist_skips_graph_file_when_source_is_empty() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, _source) = graph_and_source();
persist(
Validated::new(graph, String::new(), vec![]),
PersistOptions {
run_dir: run_dir.clone(),
run_record: sample_record(different_graph()),
},
)
.unwrap();
assert!(!run_dir.join(GRAPH_FILE_NAME).exists());
assert!(run_dir.join(RunRecord::file_name()).exists());
}
#[test]
fn persist_overwrites_run_record_graph_with_validated_graph() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, source) = graph_and_source();
persist(
Validated::new(graph.clone(), source, vec![]),
PersistOptions {
run_dir: run_dir.clone(),
run_record: sample_record(different_graph()),
},
)
.unwrap();
let saved = RunRecord::load(&run_dir).unwrap();
assert_eq!(saved.graph.name, graph.name);
assert!(saved.graph.nodes.contains_key("exit"));
assert_eq!(
serde_json::to_value(saved.graph).unwrap(),
serde_json::to_value(graph).unwrap()
);
}
#[test]
fn persist_roundtrips_full_run_record_fields_through_load() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, source) = graph_and_source();
let mut expected = sample_record(different_graph());
expected.graph = graph.clone();
persist(
Validated::new(graph, source, vec![]),
PersistOptions {
run_dir: run_dir.clone(),
run_record: expected.clone(),
},
)
.unwrap();
let loaded = Persisted::load(&run_dir).unwrap();
assert_eq!(
serde_json::to_value(loaded.run_record()).unwrap(),
serde_json::to_value(expected).unwrap()
);
assert!(loaded.diagnostics().is_empty());
}
#[test]
fn persist_returns_error_on_io_failure() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::write(&run_dir, "not a directory").unwrap();
let (graph, source) = graph_and_source();
let err = persist(
Validated::new(graph, source, vec![]),
PersistOptions {
run_dir,
run_record: sample_record(different_graph()),
},
)
.unwrap_err();
assert!(matches!(err, FabroError::Io(_)));
}
#[test]
fn load_roundtrips_persisted_workflow() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, source) = graph_and_source();
let mut expected = sample_record(different_graph());
expected.graph = graph.clone();
let persisted = persist(
Validated::new(graph, source.clone(), vec![]),
PersistOptions {
run_dir: run_dir.clone(),
run_record: expected.clone(),
},
)
.unwrap();
let loaded = Persisted::load(&run_dir).unwrap();
assert_eq!(loaded.source(), source);
assert_eq!(loaded.run_dir(), run_dir.as_path());
assert_eq!(
serde_json::to_value(loaded.run_record()).unwrap(),
serde_json::to_value(expected).unwrap()
);
assert_eq!(
serde_json::to_value(loaded.graph()).unwrap(),
serde_json::to_value(persisted.graph()).unwrap()
);
}
#[test]
fn load_uses_empty_source_when_graph_file_is_missing() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let (graph, _source) = graph_and_source();
let mut record = sample_record(different_graph());
record.graph = graph;
record.save(&run_dir).unwrap();
let loaded = Persisted::load(&run_dir).unwrap();
assert!(loaded.source().is_empty());
}
#[test]
fn load_reads_graph_from_run_json_and_source_from_graph_file() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let (graph, _) = graph_and_source();
let mut record = sample_record(different_graph());
record.graph = graph.clone();
record.save(&run_dir).unwrap();
std::fs::write(
run_dir.join(GRAPH_FILE_NAME),
"digraph mismatch { a -> b; }",
)
.unwrap();
let loaded = Persisted::load(&run_dir).unwrap();
assert_eq!(
serde_json::to_value(loaded.graph()).unwrap(),
serde_json::to_value(graph).unwrap()
);
assert_eq!(loaded.source(), "digraph mismatch { a -> b; }");
}
#[test]
fn load_reads_graph_source_from_graph_file() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, source) = graph_and_source();
persist(
Validated::new(graph, source.clone(), vec![]),
PersistOptions {
run_dir: run_dir.clone(),
run_record: sample_record(different_graph()),
},
)
.unwrap();
let loaded = Persisted::load(&run_dir).unwrap();
assert_eq!(loaded.source(), source);
}
}

101
lib/crates/fabro-workflows/src/pipeline/types.rs
View file

@ -1,5 +1,5 @@
use std::collections::HashMap;
use std::path::PathBuf;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use fabro_agent::Sandbox;
@ -12,8 +12,7 @@ use crate::error::FabroError;
use crate::event::EventEmitter;
use crate::handler::HandlerRegistry;
use crate::outcome::Outcome;
use crate::records::Checkpoint;
use crate::records::Conclusion;
use crate::records::{Checkpoint, Conclusion, RunRecord};
use crate::run_settings::{LifecycleConfig, RunSettings};
use fabro_validate::Severity;
@ -94,10 +93,104 @@ impl Validated {
}
}
/// Options for the PERSIST phase.
pub struct PersistOptions {
pub run_dir: PathBuf,
pub run_record: RunRecord,
}
/// Output of the PERSIST phase. Run directory created, run.json and graph.fabro written.
#[derive(Debug)]
#[non_exhaustive]
pub struct Persisted {
graph: Graph,
source: String,
diagnostics: Vec<Diagnostic>,
run_dir: PathBuf,
run_record: RunRecord,
}
impl Persisted {
/// Create a new `Persisted` from its parts.
pub(crate) fn new(
graph: Graph,
source: String,
diagnostics: Vec<Diagnostic>,
run_dir: PathBuf,
run_record: RunRecord,
) -> Self {
Self {
graph,
source,
diagnostics,
run_dir,
run_record,
}
}
pub fn graph(&self) -> &Graph {
&self.graph
}
pub fn source(&self) -> &str {
&self.source
}
pub fn diagnostics(&self) -> &[Diagnostic] {
&self.diagnostics
}
pub fn run_dir(&self) -> &Path {
&self.run_dir
}
pub fn run_record(&self) -> &RunRecord {
&self.run_record
}
/// True if any diagnostic has Error severity.
#[must_use]
pub fn has_errors(&self) -> bool {
self.diagnostics
.iter()
.any(|d| d.severity == Severity::Error)
}
/// Returns `Err(FabroError::Validation)` if any Error-severity diagnostics exist.
pub fn raise_on_errors(&self) -> Result<(), FabroError> {
if self.has_errors() {
let message = self
.diagnostics
.iter()
.filter(|d| d.severity == Severity::Error)
.map(|d| d.message.as_str())
.collect::<Vec<_>>()
.join("; ");
return Err(FabroError::Validation(message));
}
Ok(())
}
/// Consume into owned graph, source, diagnostics, run dir, and run record.
pub fn into_parts(self) -> (Graph, String, Vec<Diagnostic>, PathBuf, RunRecord) {
(
self.graph,
self.source,
self.diagnostics,
self.run_dir,
self.run_record,
)
}
/// Load a previously persisted run from disk.
pub fn load(run_dir: &Path) -> Result<Self, FabroError> {
super::persist::load(run_dir)
}
}
/// Options for the INITIALIZE phase.
pub struct InitOptions {
pub run_id: String,
pub run_dir: PathBuf,
pub dry_run: bool,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn Sandbox>,