scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-08 22:21:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

refactor: clean CREATE/START/RESUME separation

Browse source 
Resume now follows the same subprocess pattern as run: look up run
directory by ID prefix, validate checkpoint exists, clean stale
artifacts, reset status to Submitted, spawn _run_engine --resume, and
attach. This eliminates ~1600 lines of duplicated env/sandbox setup
from resume.rs.

Key changes:
- operations::start() and operations::resume() take run_dir instead
  of Persisted, loading state from disk internally
- run_engine() builds RunOptions from RunRecord on disk, so callers
  no longer extract record fields manually
- StartOptions flattened (no more nested InitOptions)
- FabroError::Precondition variant for start/resume guard checks
- _run_engine accepts --resume flag to dispatch to resume path
- operations::restore removed (no longer needed)
- Resume CLI stripped to just <RUN_ID> + --detach

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-26 12:58:47 -04:00
parent f4f4762ab4
commit d45b4516ab
No known key found for this signature in database
12 changed files with 476 additions and 2058 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
docs/core-concepts/how-fabro-works.mdx
View file

@ -98,11 +98,5 @@ Because Fabro checkpoints after every stage, interrupted runs can be resumed fro
fabro resume <RUN_ID>
```
Or resume from a checkpoint file:
```bash
fabro resume --checkpoint path/to/checkpoint.json --workflow workflow.fabro
```
The engine restores the full context, node visit counts, and retry state, then continues execution from the next node.
The engine restores the full context, node visit counts, and retry state from the run directory, then continues execution from the next node.

31
docs/execution/checkpoints.mdx
View file

@ -90,35 +90,22 @@ For Daytona sandboxes, the worktree is created inside the remote sandbox instead
## Resuming a run
There are two ways to resume an interrupted run:
### From a checkpoint file
Resume from a `checkpoint.json` saved in the run directory:
```bash
fabro resume --checkpoint path/to/logs/checkpoint.json --workflow workflow.fabro
```
Fabro loads the checkpoint, restores the context and execution state, and continues from the next node after the checkpoint.
### From a run branch
Resume from the Git branches created during a previous run:
Resume an interrupted run from its checkpoint on disk:
```bash
fabro resume 01JKXYZ
```
This reads the checkpoint, run record, and Graphviz graph from the metadata branch (`fabro/meta/01JKXYZ...`), re-attaches a worktree to the existing run branch, and resumes execution. No workflow file argument is needed — everything is recovered from Git.
Fabro looks up the run directory by ID prefix, loads `checkpoint.json` and `run.json` from the run directory, and spawns a new engine process to continue execution. No workflow file or override flags are needed — all configuration is read from the persisted run state.
<Accordion title="What happens during resume">
1. Fabro reads `checkpoint.json` from the metadata branch
2. Reads `run.json` to reconstruct the workflow graph and config
3. Creates a fresh worktree attached to the existing run branch
4. Restores the full context, completed node list, retry counts, and failure signatures
5. If the checkpointed node used `full` fidelity, downgrades the first resumed node to `summary:high` (since the original conversation thread no longer exists in memory)
6. Continues execution from `next_node_id`
1. Fabro looks up the run directory by ID prefix
2. Validates that `checkpoint.json` exists and no engine process is already running
3. Cleans stale artifacts from the previous execution (conclusion, PID file, etc.)
4. Resets status to `Submitted` and spawns a new engine subprocess with `--resume`
5. The engine loads `run.json` and `checkpoint.json`, restores the full context, completed node list, retry counts, and failure signatures
6. If the checkpointed node used `full` fidelity, downgrades the first resumed node to `summary:high` (since the original conversation thread no longer exists in memory)
7. Continues execution from `next_node_id`
</Accordion>
## The checkpoint cycle

22
docs/reference/cli.mdx
View file

@ -68,31 +68,17 @@ fabro run run.toml
## `fabro resume`
Resume an interrupted workflow run from its last checkpoint.
Resume an interrupted workflow run from its last checkpoint. The run is looked up by ID prefix and uses the configuration persisted at create time — no runtime overrides are accepted.
```bash
fabro resume <RUN_ID>
fabro resume <RUN_ID> --workflow updated.fabro
fabro resume --checkpoint path/to/checkpoint.json --workflow workflow.fabro
fabro resume <RUN_ID> --detach
```
| Argument / Flag | Description |
|---|---|
| `<RUN_ID>` | Run ID, prefix, or branch (`fabro/run/...`). Not required when using `--checkpoint`. |
| `--checkpoint <FILE>` | Resume from a checkpoint file (requires `--workflow`) |
| `--workflow <FILE>` | Override workflow graph (required with `--checkpoint`) |
| `--run-dir <DIR>` | Run output directory |
| `--dry-run` | Execute with a simulated LLM backend |
| `--auto-approve` | Auto-approve all human gates |
| `--model <MODEL>` | Override default LLM model |
| `--provider <PROVIDER>` | Override default LLM provider |
| `-v, --verbose` | Enable verbose output |
| `--sandbox <SANDBOX>` | Sandbox for agent tools: `local`, `docker`, `daytona`, `ssh`, or `exe` |
| `--goal <GOAL>` | Override the workflow goal |
| `--goal-file <FILE>` | Read the goal from a file |
| `--no-retro` | Skip retro generation after the run |
| `--preserve-sandbox` | Keep the sandbox alive after the run finishes |
| `--label <KEY=VALUE>` | Attach a label to this run (repeatable) |
| `<RUN_ID>` | Run ID or unambiguous prefix |
| `-d, --detach` | Run in the background and print the run ID |
## `fabro ps`

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

File diff suppressed because it is too large Load diff

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

@ -22,7 +22,8 @@ use fabro_workflows::git::GitSyncStatus;
use fabro_workflows::handler::default_registry;
use fabro_workflows::handler::llm::{AgentApiBackend, AgentCliBackend, BackendRouter};
use fabro_workflows::operations::{
start, StartFinalizeOptions, StartOptions, StartPullRequestConfig, StartRetroOptions,
resume as operations_resume, start, StartFinalizeOptions, StartOptions, StartPullRequestConfig,
StartRetroOptions,
};
use fabro_workflows::outcome::StageStatus;
use fabro_workflows::outcome::{compute_stage_cost, format_cost};
@ -30,7 +31,7 @@ use fabro_workflows::pipeline::{
build_conclusion, classify_engine_result, persist_terminal_outcome, Persisted, Validated,
};
use fabro_workflows::records::Checkpoint;
use fabro_workflows::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
use fabro_workflows::run_options::{GitCheckpointOptions, LifecycleOptions};
use indicatif::HumanDuration;
use std::time::Duration;
use tracing::debug;
@ -755,7 +756,82 @@ pub async fn run_from_record(
run_id: Some(record.run_id.clone()),
};
run_command_impl(args, styles, github_app, git_author, Some(record_run)).await
run_command_impl(
args,
styles,
github_app,
git_author,
Some(record_run),
false,
)
.await
}
/// Resume an existing workflow run from its persisted checkpoint.
pub async fn resume_from_record(
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 record_run = RecordBasedRun {
workflow: WorkflowState::Persisted(Box::new(persisted)),
run_defaults,
};
let sandbox_provider = record
.config
.sandbox
.as_ref()
.and_then(|s| s.provider.as_deref())
.unwrap_or("local")
.parse()
.unwrap_or(SandboxProvider::Local);
let model = record
.config
.llm
.as_ref()
.and_then(|l| l.model.clone())
.unwrap_or_default();
let provider = record
.config
.llm
.as_ref()
.and_then(|l| l.provider.clone())
.filter(|s| !s.is_empty());
let args = RunArgs {
workflow: None,
run_dir: Some(run_dir),
dry_run: record.config.dry_run_enabled(),
preflight: false,
auto_approve: record.config.auto_approve_enabled(),
goal: record.config.goal.clone(),
goal_file: None,
model: Some(model),
provider,
verbose: record.config.verbose_enabled(),
sandbox: Some(CliSandboxProvider::from(sandbox_provider)),
label: record
.labels
.iter()
.map(|(k, v)| format!("{k}={v}"))
.collect(),
no_retro: record.config.no_retro_enabled(),
preserve_sandbox: record
.config
.sandbox
.as_ref()
.and_then(|s| s.preserve)
.unwrap_or(false),
detach: false,
run_id: Some(record.run_id.clone()),
};
run_command_impl(args, styles, github_app, git_author, Some(record_run), true).await
}
/// Execute a full workflow run.
@ -778,7 +854,15 @@ pub async fn run_command(
run_defaults: resolved_run_defaults,
};
run_command_impl(args, styles, github_app, git_author, Some(record_run)).await
run_command_impl(
args,
styles,
github_app,
git_author,
Some(record_run),
false,
)
.await
}
async fn run_command_impl(
@ -787,6 +871,7 @@ async fn run_command_impl(
github_app: Option<fabro_github::GitHubAppCredentials>,
git_author: fabro_workflows::git::GitAuthor,
record_run: Option<RecordBasedRun>,
resume: bool,
) -> anyhow::Result<()> {
let (workflow, run_defaults) = match record_run {
Some(rr) => (rr.workflow, rr.run_defaults),
@ -937,7 +1022,6 @@ async fn run_command_impl(
}
};
let mut run_cfg = Some(persisted.run_record().config.clone());
let workflow_slug = persisted.run_record().workflow_slug.clone();
let sandbox_provider = run_cfg
.as_ref()
.and_then(|cfg| cfg.sandbox.as_ref())
@ -978,8 +1062,6 @@ async fn run_command_impl(
write_run_config_snapshot(&run_dir, workflow_toml_path.as_deref()).await?;
}
let settings_config = persisted.run_record().config.clone();
// Now resolve ${env.VARNAME} references for runtime use.
if let Some(ref mut cfg) = run_cfg {
run_config::resolve_sandbox_env(cfg)?;
@ -1645,30 +1727,6 @@ async fn run_command_impl(
None
};
let run_options = RunOptions {
config: settings_config,
run_dir: run_dir.clone(),
cancel_token: None,
dry_run: dry_run_mode,
run_id: run_id.clone(),
labels: persisted.run_record().labels.clone(),
git_author: git_author.clone(),
workflow_slug: workflow_slug.clone(),
github_app: github_app.clone(),
base_branch: persisted
.run_record()
.base_branch
.clone()
.or(detected_base_branch),
host_repo_path: persisted
.run_record()
.host_repo_path
.as_deref()
.map(PathBuf::from)
.or_else(|| Some(original_cwd.clone())),
git,
};
// Build lifecycle config for sandbox init, setup commands, and devcontainer phases
let lifecycle = LifecycleOptions {
setup_commands,
@ -1691,46 +1749,46 @@ async fn run_command_impl(
let pr_config = if dry_run_mode {
None
} else {
run_options.pull_request().cloned()
persisted.run_record().config.pull_request.clone()
};
let started = start(
persisted,
StartOptions {
init: fabro_workflows::pipeline::InitOptions {
run_id: run_id.clone(),
dry_run: dry_run_mode,
emitter: Arc::clone(&emitter),
sandbox: Arc::clone(&sandbox),
registry: Arc::new(registry),
lifecycle,
run_options,
hooks: fabro_hooks::HookConfig {
hooks: run_cfg
.as_ref()
.map(|c| c.hooks.clone())
.unwrap_or_else(|| run_defaults.hooks.clone()),
},
sandbox_env,
checkpoint: None,
seed_context: None,
},
retro: StartRetroOptions {
enabled: !no_retro_flag && project_config::is_retro_enabled(),
dry_run: dry_run_mode,
llm_client: llm_client.clone(),
provider: provider_enum,
model: model.clone(),
},
finalize: StartFinalizeOptions { preserve_sandbox },
pull_request: StartPullRequestConfig {
pr_config,
github_app: github_app.clone(),
origin_url: origin_url.clone(),
model: model.clone(),
},
let start_options = StartOptions {
cancel_token: None,
emitter: Arc::clone(&emitter),
sandbox: Arc::clone(&sandbox),
registry: Arc::new(registry),
lifecycle,
hooks: fabro_hooks::HookConfig {
hooks: run_cfg
.as_ref()
.map(|c| c.hooks.clone())
.unwrap_or_else(|| run_defaults.hooks.clone()),
},
)
.await;
sandbox_env,
seed_context: None,
git_author,
git,
github_app: github_app.clone(),
dry_run: dry_run_mode,
retro: StartRetroOptions {
enabled: !no_retro_flag && project_config::is_retro_enabled(),
dry_run: dry_run_mode,
llm_client: llm_client.clone(),
provider: provider_enum,
model: model.clone(),
},
finalize: StartFinalizeOptions { preserve_sandbox },
pull_request: StartPullRequestConfig {
pr_config,
github_app: github_app.clone(),
origin_url: origin_url.clone(),
model: model.clone(),
},
};
let started = if resume {
operations_resume(&run_dir, start_options).await
} else {
start(&run_dir, start_options).await
};
let run_duration_ms = run_start.elapsed().as_millis() as u64;
let mut completion_guard = DetachedRunCompletionGuard::arm(&run_dir);

12
lib/crates/fabro-cli/src/commands/start.rs
View file

@ -9,7 +9,7 @@ use super::detached_support::persist_detached_failure;
///
/// The engine process reads `run.json` from the run directory and executes the
/// workflow. Returns the child process handle (use `.id()` for the PID).
pub fn start_run(run_dir: &Path) -> Result<std::process::Child> {
pub fn start_run(run_dir: &Path, resume: bool) -> Result<std::process::Child> {
// Validate status is Submitted
let status_path = run_dir.join("status.json");
match fabro_workflows::run_status::RunStatusRecord::load(&status_path) {
@ -56,9 +56,11 @@ pub fn start_run(run_dir: &Path) -> Result<std::process::Child> {
return Err(err);
}
};
cmd.args(["_run_engine", "--run-dir"])
.arg(run_dir)
.stdout(stdout_log)
cmd.args(["_run_engine", "--run-dir"]).arg(run_dir);
if resume {
cmd.arg("--resume");
}
cmd.stdout(stdout_log)
.stderr(log_file)
.stdin(std::process::Stdio::null());
@ -134,7 +136,7 @@ mod tests {
sample_record().save(dir.path()).unwrap();
std::fs::create_dir(dir.path().join("detach.log")).unwrap();
let _ = start_run(dir.path());
let _ = start_run(dir.path(), false);
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(

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

@ -90,6 +90,9 @@ enum Command {
/// Path to the run directory
#[arg(long)]
run_dir: PathBuf,
/// Resume from checkpoint instead of fresh start
#[arg(long)]
resume: bool,
},
/// Validate a workflow
Validate(commands::validate::ValidateArgs),
@ -315,6 +318,7 @@ pub(crate) fn build_github_app_credentials(
async fn run_engine_entrypoint(
run_dir: PathBuf,
resume: bool,
styles: &'static fabro_util::terminal::Styles,
) -> Result<()> {
let cli_config = cli_config::load_cli_config(None)?;
@ -355,18 +359,29 @@ async fn run_engine_entrypoint(
return Err(err);
}
// Use run_from_record: loads config + graph directly from persisted state,
// skipping workflow source loading and preprocessing entirely.
match commands::run::run_from_record(
persisted,
run_dir.clone(),
cli_config,
styles,
github_app,
git_author,
)
.await
{
let result = if resume {
commands::run::resume_from_record(
persisted,
run_dir.clone(),
cli_config,
styles,
github_app,
git_author,
)
.await
} else {
commands::run::run_from_record(
persisted,
run_dir.clone(),
cli_config,
styles,
github_app,
git_author,
)
.await
};
match result {
Ok(()) => Ok(()),
Err(err) => {
let _ = commands::detached_support::persist_detached_failure(
@ -752,7 +767,7 @@ async fn main_inner() -> (String, Result<()>) {
#[cfg(feature = "sleep_inhibitor")]
let _sleep_guard = fabro_beastie::guard(_prevent_idle_sleep);
let child = commands::start::start_run(&run_dir)?;
let child = commands::start::start_run(&run_dir, false)?;
if args.detach {
println!("{run_id}");
@ -778,7 +793,7 @@ async fn main_inner() -> (String, Result<()>) {
Command::Start { run } => {
let base = fabro_workflows::run_lookup::default_runs_base();
let run_info = fabro_workflows::run_lookup::resolve_run(&base, &run)?;
let child = commands::start::start_run(&run_info.path)?;
let child = commands::start::start_run(&run_info.path, false)?;
eprintln!("Started engine process (PID {})", child.id());
}
Command::Attach { run } => {
@ -792,10 +807,10 @@ async fn main_inner() -> (String, Result<()>) {
std::process::exit(1);
}
}
Command::RunEngine { run_dir } => {
Command::RunEngine { run_dir, resume } => {
let styles: &'static fabro_util::terminal::Styles =
Box::leak(Box::new(fabro_util::terminal::Styles::detect_stderr()));
run_engine_entrypoint(run_dir, styles).await?;
run_engine_entrypoint(run_dir, resume, styles).await?;
}
Command::Validate(args) => {
let styles = fabro_util::terminal::Styles::detect_stderr();
@ -946,20 +961,16 @@ async fn main_inner() -> (String, Result<()>) {
commands::secret::set_command(&args)?;
}
},
Command::Resume(mut args) => {
Command::Resume(args) => {
let styles: &'static fabro_util::terminal::Styles =
Box::leak(Box::new(fabro_util::terminal::Styles::detect_stderr()));
let cli_config = cli_config::load_cli_config(None)?;
args.verbose = args.verbose || cli_config.verbose_enabled();
#[cfg(feature = "sleep_inhibitor")]
let _sleep_guard = fabro_beastie::guard(cli_config.prevent_idle_sleep_enabled());
let github_app = build_github_app_credentials(cli_config.app_id());
let git_author = fabro_workflows::git::GitAuthor::from_options(
cli_config.git_author().and_then(|a| a.name.clone()),
cli_config.git_author().and_then(|a| a.email.clone()),
);
commands::resume::resume_command(args, cli_config, styles, github_app, git_author)
.await?;
{
let cli_config = cli_config::load_cli_config(None)?;
let _sleep_guard =
fabro_beastie::guard(cli_config.prevent_idle_sleep_enabled());
}
commands::resume::resume_command(args, styles).await?;
}
Command::Rewind(args) => {
let styles = fabro_util::terminal::Styles::detect_stderr();
@ -1114,8 +1125,28 @@ mod tests {
let cli = Cli::try_parse_from(["fabro", "_run_engine", "--run-dir", "/tmp/runs/test"])
.expect("should parse");
match cli.command {
Command::RunEngine { run_dir } => {
Command::RunEngine { run_dir, resume } => {
assert_eq!(run_dir, std::path::PathBuf::from("/tmp/runs/test"));
assert!(!resume);
}
_ => panic!("unexpected command variant"),
}
}
#[test]
fn parse_run_engine_with_resume() {
let cli = Cli::try_parse_from([
"fabro",
"_run_engine",
"--run-dir",
"/tmp/runs/test",
"--resume",
])
.expect("should parse");
match cli.command {
Command::RunEngine { run_dir, resume } => {
assert_eq!(run_dir, std::path::PathBuf::from("/tmp/runs/test"));
assert!(resume);
}
_ => panic!("unexpected command variant"),
}

85
lib/crates/fabro-cli/tests/cli.rs
View file

@ -523,12 +523,13 @@ fn resume_help_shows_expected_args() {
.args(["resume", "--help"])
.assert()
.success()
.stdout(predicate::str::contains("--checkpoint"))
.stdout(predicate::str::contains("--workflow"));
.stdout(predicate::str::contains("--detach"))
.stdout(predicate::str::contains("--checkpoint").not())
.stdout(predicate::str::contains("--workflow").not());
}
#[test]
fn resume_requires_run_or_checkpoint() {
fn resume_requires_run_arg() {
arc().args(["resume"]).assert().failure();
}
@ -745,84 +746,6 @@ digraph FooWorkflow {
assert_eq!(run_record["workflow_slug"].as_str(), Some("alpha"));
}
#[test]
fn resumed_run_preserves_workflow_slug_for_lookup() {
let home = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let workflow_dir = project.path().join("workflows").join("sluggy");
std::fs::create_dir_all(&workflow_dir).unwrap();
let workflow_path = workflow_dir.join("workflow.fabro");
std::fs::write(
&workflow_path,
"\
digraph BarBaz {
start [shape=Mdiamond, label=\"Start\"]
exit [shape=Msquare, label=\"Exit\"]
start -> exit
}
",
)
.unwrap();
let original_run_dir = project.path().join("original-run");
arc()
.env("HOME", home.path())
.current_dir(project.path())
.args([
"run",
"--dry-run",
"--auto-approve",
"--no-retro",
"--run-dir",
original_run_dir.to_str().unwrap(),
workflow_path.to_str().unwrap(),
])
.assert()
.success();
arc()
.env("HOME", home.path())
.current_dir(project.path())
.args([
"resume",
"--checkpoint",
original_run_dir.join("checkpoint.json").to_str().unwrap(),
"--workflow",
workflow_path.to_str().unwrap(),
"--dry-run",
"--auto-approve",
"--no-retro",
])
.assert()
.success();
arc()
.env("HOME", home.path())
.current_dir(project.path())
.args(["attach", "sluggy"])
.timeout(std::time::Duration::from_secs(10))
.assert()
.success();
let resumed_runs_dir = home.path().join(".fabro").join("runs");
let resumed_run_dir = std::fs::read_dir(&resumed_runs_dir)
.unwrap()
.flatten()
.map(|entry| entry.path())
.find(|path| path.is_dir())
.unwrap_or_else(|| {
panic!(
"expected a resumed run under {}",
resumed_runs_dir.display()
)
});
let run_record: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(resumed_run_dir.join("run.json")).unwrap())
.unwrap();
assert_eq!(run_record["graph"]["name"].as_str(), Some("BarBaz"));
assert_eq!(run_record["workflow_slug"].as_str(), Some("sluggy"));
}
#[test]
fn dry_run_create_start_attach_works_with_default_run_lookup() {
let home = tempfile::tempdir().unwrap();

5
lib/crates/fabro-workflows/src/error.rs
View file

@ -233,6 +233,9 @@ pub enum FabroError {
#[error("I/O error: {0}")]
Io(String),
#[error("Precondition failed: {0}")]
Precondition(String),
#[error("Pipeline cancelled")]
Cancelled,
}
@ -274,6 +277,7 @@ impl FabroError {
| Self::ValidationFailed { .. }
| Self::Stylesheet(_)
| Self::Checkpoint(_)
| Self::Precondition(_)
| Self::Cancelled => false,
}
}
@ -290,6 +294,7 @@ impl FabroError {
| Self::ValidationFailed { .. }
| Self::Stylesheet(_)
| Self::Checkpoint(_) => FailureCategory::Deterministic,
Self::Precondition(_) => FailureCategory::Structural,
Self::Handler { failure_class, .. } | Self::Engine { failure_class, .. } => {
*failure_class
}

5
lib/crates/fabro-workflows/src/operations/mod.rs
View file

@ -1,6 +1,5 @@
mod create;
mod fork;
mod restore;
mod rewind;
mod start;
@ -9,11 +8,11 @@ pub use create::{
ValidateOptions,
};
pub use fork::fork;
pub use restore::{restore, RestoreOptions};
pub use rewind::{
build_timeline, find_run_id_by_prefix, load_parallel_map, parse_target, resolve_target, rewind,
TimelineEntry,
};
pub use start::{
start, StartFinalizeOptions, StartOptions, StartPullRequestConfig, StartRetroOptions, Started,
resume, start, StartFinalizeOptions, StartOptions, StartPullRequestConfig, StartRetroOptions,
Started,
};

163
lib/crates/fabro-workflows/src/operations/restore.rs
View file

@ -1,163 +0,0 @@
use std::path::PathBuf;
use crate::error::FabroError;
use crate::pipeline::types::PersistOptions;
use crate::pipeline::{self, Persisted, Validated};
use crate::records::RunRecord;
use super::create::finalize_config;
pub struct RestoreOptions {
pub run_dir: PathBuf,
pub run_record: RunRecord,
}
/// Materialize an existing run record to local disk.
///
/// Unlike `create()`, this skips parsing, transforms, and validation because
/// the caller already has the resolved graph from the original run.
pub fn restore(options: RestoreOptions) -> Result<Persisted, FabroError> {
let mut run_record = options.run_record;
finalize_config(&mut run_record.config, &run_record.graph);
let graph = run_record.graph.clone();
let validated = Validated::new(graph, String::new(), vec![]);
pipeline::persist(
validated,
PersistOptions {
run_dir: options.run_dir,
run_record,
},
)
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
use chrono::{TimeZone, Utc};
use fabro_config::config::FabroConfig;
use fabro_graphviz::graph::{AttrValue, Graph};
fn sample_graph() -> Graph {
let mut graph = Graph::new("restore-test");
graph.attrs.insert(
"goal".to_string(),
AttrValue::String("Ship feature".to_string()),
);
graph
}
fn sample_record() -> RunRecord {
RunRecord {
run_id: "restore-run-123".to_string(),
created_at: Utc.with_ymd_and_hms(2025, 1, 2, 3, 4, 5).single().unwrap(),
config: FabroConfig {
llm: Some(fabro_config::run::LlmConfig {
model: Some("sonnet".to_string()),
provider: None,
fallbacks: None,
}),
pull_request: Some(fabro_config::run::PullRequestConfig {
enabled: false,
..Default::default()
}),
dry_run: Some(true),
..Default::default()
},
graph: sample_graph(),
workflow_slug: Some("restore-slug".to_string()),
working_directory: PathBuf::from("/tmp/original-project"),
host_repo_path: Some("/tmp/original-project".to_string()),
base_branch: Some("main".to_string()),
labels: HashMap::from([("env".to_string(), "test".to_string())]),
}
}
#[test]
fn restore_roundtrips_and_normalizes_config() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let persisted = restore(RestoreOptions {
run_dir: run_dir.clone(),
run_record: sample_record(),
})
.unwrap();
let loaded = Persisted::load(&run_dir).unwrap();
assert_eq!(persisted.run_record().run_id, "restore-run-123");
assert_eq!(
persisted
.run_record()
.config
.llm
.as_ref()
.and_then(|llm| llm.model.as_deref()),
Some("claude-sonnet-4-6")
);
assert_eq!(
persisted
.run_record()
.config
.llm
.as_ref()
.and_then(|llm| llm.provider.as_deref()),
Some("anthropic")
);
assert_eq!(
persisted.run_record().config.goal.as_deref(),
Some("Ship feature")
);
assert!(persisted.run_record().config.pull_request.is_none());
assert_eq!(
serde_json::to_value(loaded.run_record()).unwrap(),
serde_json::to_value(persisted.run_record()).unwrap()
);
}
#[test]
fn restore_preserves_run_record_fields() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let record = sample_record();
restore(RestoreOptions {
run_dir: run_dir.clone(),
run_record: record.clone(),
})
.unwrap();
let loaded = Persisted::load(&run_dir).unwrap();
assert_eq!(loaded.run_record().run_id, record.run_id);
assert_eq!(loaded.run_record().workflow_slug, record.workflow_slug);
assert_eq!(loaded.run_record().labels, record.labels);
assert_eq!(
loaded.run_record().working_directory,
record.working_directory
);
assert_eq!(loaded.run_record().host_repo_path, record.host_repo_path);
assert_eq!(loaded.run_record().base_branch, record.base_branch);
}
#[test]
fn restore_preserves_created_at_and_run_lookup_uses_it_without_start_record() {
let temp = tempfile::tempdir().unwrap();
let runs_base = temp.path().join("runs");
let run_dir = runs_base.join("restore-run-123");
let record = sample_record();
restore(RestoreOptions {
run_dir,
run_record: record.clone(),
})
.unwrap();
let runs = crate::run_lookup::scan_runs(&runs_base).unwrap();
assert_eq!(runs.len(), 1);
assert_eq!(runs[0].run_id, record.run_id);
assert_eq!(runs[0].start_time, record.created_at.to_rfc3339());
assert_eq!(runs[0].start_time_dt, Some(record.created_at));
}
}

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

@ -1,12 +1,17 @@
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use crate::context::Context;
use crate::error::FabroError;
use crate::event::WorkflowRunEvent;
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::handler::HandlerRegistry;
use crate::outcome::StageStatus;
use crate::pipeline::{
self, FinalizeOptions, Finalized, InitOptions, Persisted, PullRequestOptions, RetroOptions,
};
use crate::records::Checkpoint;
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
pub struct StartRetroOptions {
pub enabled: bool,
@ -27,8 +32,27 @@ pub struct StartPullRequestConfig {
pub model: String,
}
/// Options for `start()` and `resume()`.
///
/// Fields that are derivable from `RunRecord` (run_id, labels, base_branch,
/// host_repo_path, config, workflow_slug) are read from disk by `run_engine()`.
/// Callers only provide truly external values.
pub struct StartOptions {
pub init: InitOptions,
// Truly external (not derivable from RunRecord)
pub cancel_token: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn fabro_agent::Sandbox>,
pub registry: Arc<HandlerRegistry>,
pub lifecycle: LifecycleOptions,
pub hooks: fabro_hooks::HookConfig,
pub sandbox_env: HashMap<String, String>,
pub seed_context: Option<Context>,
pub git_author: crate::git::GitAuthor,
pub git: Option<GitCheckpointOptions>,
pub github_app: Option<fabro_github::GitHubAppCredentials>,
// Still external for now — could be derived from RunRecord.config in follow-up
pub dry_run: bool,
pub retro: StartRetroOptions,
pub finalize: StartFinalizeOptions,
pub pull_request: StartPullRequestConfig,
@ -40,10 +64,40 @@ pub struct Started {
pub retro_duration: Duration,
}
/// Run a persisted workflow through initialize, execute, retro, finalize, and pull_request.
pub async fn start(persisted: Persisted, options: StartOptions) -> Result<Started, FabroError> {
/// Start a fresh workflow run. Errors if a checkpoint already exists (use `resume()` instead).
pub async fn start(
run_dir: &std::path::Path,
options: StartOptions,
) -> Result<Started, FabroError> {
if run_dir.join("checkpoint.json").exists() {
return Err(FabroError::Precondition(
"checkpoint.json exists in run directory — did you mean to resume?".to_string(),
));
}
let persisted = Persisted::load(run_dir)?;
run_engine(persisted, None, options).await
}
/// Resume a workflow run from its checkpoint. Errors if no checkpoint is found.
pub async fn resume(
run_dir: &std::path::Path,
options: StartOptions,
) -> Result<Started, FabroError> {
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}")))?;
let persisted = Persisted::load(run_dir)?;
run_engine(persisted, Some(checkpoint), options).await
}
/// Shared engine: initialize, execute, retro, finalize, pull_request.
async fn run_engine(
persisted: Persisted,
checkpoint: Option<Checkpoint>,
options: StartOptions,
) -> Result<Started, FabroError> {
let preserve_sandbox = options.finalize.preserve_sandbox;
let sandbox_for_cleanup = Arc::clone(&options.init.sandbox);
let sandbox_for_cleanup = Arc::clone(&options.sandbox);
let cleanup_guard = scopeguard::guard((), move |()| {
if preserve_sandbox {
return;
@ -55,7 +109,40 @@ pub async fn start(persisted: Persisted, options: StartOptions) -> Result<Starte
}
});
let initialized = pipeline::initialize(persisted, options.init).await?;
// Build RunOptions from the persisted RunRecord + external caller options
let record = persisted.run_record();
let run_options = RunOptions {
config: record.config.clone(),
run_dir: persisted.run_dir().to_path_buf(),
cancel_token: options.cancel_token,
dry_run: options.dry_run,
run_id: record.run_id.clone(),
labels: record.labels.clone(),
git_author: options.git_author,
workflow_slug: record.workflow_slug.clone(),
github_app: options.github_app.clone(),
host_repo_path: record
.host_repo_path
.as_deref()
.map(std::path::PathBuf::from),
base_branch: record.base_branch.clone(),
git: options.git,
};
let init_options = InitOptions {
run_id: record.run_id.clone(),
dry_run: options.dry_run,
emitter: options.emitter,
sandbox: options.sandbox,
registry: options.registry,
lifecycle: options.lifecycle,
run_options,
hooks: options.hooks,
sandbox_env: options.sandbox_env,
checkpoint,
seed_context: options.seed_context,
};
let initialized = pipeline::initialize(persisted, init_options).await?;
let last_git_sha: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
{
@ -145,7 +232,7 @@ mod tests {
use crate::handler::start::StartHandler;
use crate::handler::{Handler, HandlerRegistry};
use crate::outcome::Outcome;
use crate::run_options::{LifecycleOptions, RunOptions};
use crate::run_options::LifecycleOptions;
const MINIMAL_DOT: &str = r#"digraph Test {
graph [goal="Build feature"]
@ -369,23 +456,6 @@ mod tests {
.unwrap()
}
fn test_run_options(run_dir: &std::path::Path) -> RunOptions {
RunOptions {
config: FabroConfig::default(),
run_dir: run_dir.to_path_buf(),
cancel_token: None,
dry_run: false,
run_id: "run-test".to_string(),
labels: HashMap::new(),
git_author: crate::git::GitAuthor::default(),
workflow_slug: None,
github_app: None,
host_repo_path: None,
base_branch: None,
git: None,
}
}
fn test_registry() -> HandlerRegistry {
let mut registry = HandlerRegistry::new(Box::new(StartHandler));
registry.register("start", Box::new(StartHandler));
@ -395,7 +465,7 @@ mod tests {
}
fn test_start_options(
run_dir: &std::path::Path,
_run_dir: &std::path::Path,
sandbox: Arc<dyn Sandbox>,
emitter: Arc<EventEmitter>,
registry: Arc<HandlerRegistry>,
@ -403,19 +473,18 @@ mod tests {
preserve_sandbox: bool,
) -> StartOptions {
StartOptions {
init: InitOptions {
run_id: "run-test".to_string(),
dry_run: false,
emitter,
sandbox,
registry,
lifecycle,
run_options: test_run_options(run_dir),
hooks: fabro_hooks::HookConfig { hooks: vec![] },
sandbox_env: HashMap::new(),
checkpoint: None,
seed_context: None,
},
cancel_token: None,
emitter,
sandbox,
registry,
lifecycle,
hooks: fabro_hooks::HookConfig { hooks: vec![] },
sandbox_env: HashMap::new(),
seed_context: None,
git_author: crate::git::GitAuthor::default(),
git: None,
github_app: None,
dry_run: false,
retro: StartRetroOptions {
enabled: false,
dry_run: false,
@ -453,8 +522,9 @@ mod tests {
let registry = Arc::new(test_registry());
let (sandbox, cleanup_count) = counting_sandbox();
persisted_workflow(MINIMAL_DOT, &run_dir);
let result = start(
persisted_workflow(MINIMAL_DOT, &run_dir),
&run_dir,
test_start_options(
&run_dir,
sandbox,
@ -485,8 +555,9 @@ mod tests {
let sandbox: Arc<dyn Sandbox> =
Arc::new(LocalSandbox::new(std::env::current_dir().unwrap()));
persisted_workflow(EMIT_DOT, &run_dir);
let started = start(
persisted_workflow(EMIT_DOT, &run_dir),
&run_dir,
test_start_options(
&run_dir,
sandbox,
@ -512,22 +583,20 @@ mod tests {
}
#[tokio::test]
async fn start_runs_loaded_persisted_workflow() {
async fn start_loads_persisted_from_run_dir() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let wrong_run_dir = temp.path().join("wrong-run-dir");
let emitter = Arc::new(EventEmitter::new());
let registry = Arc::new(test_registry());
let sandbox: Arc<dyn Sandbox> =
Arc::new(LocalSandbox::new(std::env::current_dir().unwrap()));
persisted_workflow(MINIMAL_DOT, &run_dir);
let loaded = Persisted::load(&run_dir).unwrap();
let started = start(
loaded,
&run_dir,
test_start_options(
&wrong_run_dir,
&run_dir,
sandbox,
emitter,
registry,
@ -544,6 +613,77 @@ mod tests {
assert_eq!(started.finalized.conclusion.status, StageStatus::Success);
assert!(run_dir.join("conclusion.json").exists());
assert!(!wrong_run_dir.join("conclusion.json").exists());
}
#[tokio::test]
async fn start_errors_when_checkpoint_exists() {
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<dyn Sandbox> =
Arc::new(LocalSandbox::new(std::env::current_dir().unwrap()));
persisted_workflow(MINIMAL_DOT, &run_dir);
// Create a fake checkpoint file
std::fs::write(run_dir.join("checkpoint.json"), "{}").unwrap();
let result = start(
&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"),
);
}
#[tokio::test]
async fn resume_errors_when_checkpoint_missing() {
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<dyn Sandbox> =
Arc::new(LocalSandbox::new(std::env::current_dir().unwrap()));
persisted_workflow(MINIMAL_DOT, &run_dir);
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"),
);
}
}