refactor(workflows): remove legacy engine module

lib/crates/fabro-api/src/server.rs

@@ -23,9 +23,11 @@ use crate::jwt_auth::{AuthMode, AuthenticatedService, AuthenticatedUser};
 use fabro_interview::{Answer, Interviewer, QuestionType, WebInterviewer};
 use fabro_workflows::checkpoint::Checkpoint;
 use fabro_workflows::context::Context;
-use fabro_workflows::engine::WorkflowRunEngine;
 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::run_settings::LifecycleConfig;
 use fabro_workflows::run_settings::RunSettings;
 
 pub use fabro_types::{
@@ -475,8 +477,14 @@ async fn start_run(
     Json(req): Json<StartRunRequest>,
 ) -> Response {
     // Parse the DOT source
-    let graph = match fabro_workflows::workflow::prepare_from_source(&req.dot_source) {
-        Ok(g) => g,
+    let graph = 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
+        }
         Err(e) => {
             return ApiError::bad_request(e.to_string()).into_response();
         }
@@ -615,24 +623,7 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
     let sandbox: Arc<dyn fabro_agent::Sandbox> = Arc::new(
         fabro_agent::ReadBeforeWriteSandbox::new(Arc::new(LocalSandbox::new(cwd))),
     );
-    let mut engine = WorkflowRunEngine::with_interviewer(
-        registry,
-        Arc::new(emitter),
-        Arc::clone(&interviewer) as Arc<dyn Interviewer>,
-        sandbox,
-    );
-    if state.dry_run {
-        engine.set_dry_run(true);
-    }
-
-    // Wire up hook runner from server config
-    if !state.hooks.is_empty() {
-        let hook_config = fabro_hooks::HookConfig {
-            hooks: state.hooks.clone(),
-        };
-        let runner = fabro_hooks::HookRunner::new(hook_config);
-        engine.set_hook_runner(std::sync::Arc::new(runner));
-    }
+    let emitter = Arc::new(emitter);
 
     // Transition to Running, populate interviewer + context
     {
@@ -665,7 +656,7 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
     };
     let config = RunSettings {
         config: run_record.config,
-        run_dir,
+        run_dir: run_dir.clone(),
         cancel_token: Some(cancel_token),
         dry_run: state.dry_run,
         run_id: run_id.clone(),
@@ -678,8 +669,49 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
         git: None,
     };
 
-    let result = tokio::select! {
-        result = engine.run(&graph, &config) => result,
+    let execution = {
+        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,
+                InitOptions {
+                    run_id,
+                    run_dir,
+                    dry_run,
+                    emitter,
+                    sandbox,
+                    registry,
+                    lifecycle: LifecycleConfig {
+                        setup_commands: Vec::new(),
+                        setup_command_timeout_ms: 300_000,
+                        devcontainer_phases: Vec::new(),
+                    },
+                    run_settings: config,
+                    hooks: fabro_hooks::HookConfig { hooks },
+                    sandbox_env: HashMap::new(),
+                    checkpoint: None,
+                    seed_context: None,
+                },
+            )
+            .await?;
+            Ok::<_, fabro_workflows::error::FabroError>(pipeline::execute(initialized).await)
+        }
+    };
+
+    let (result, final_context) = tokio::select! {
+        result = execution => match result {
+            Ok(executed) => (executed.outcome, Some(executed.final_context)),
+            Err(err) => (Err(err), None),
+        },
         _ = cancel_rx => {
             let mut runs = state.runs.lock().expect("runs lock poisoned");
             if let Some(managed_run) = runs.get_mut(&run_id) {
@@ -748,6 +780,9 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
             }
         }
         managed_run.checkpoint = checkpoint;
+        if let Some(ctx) = final_context {
+            managed_run.context = Some(ctx);
+        }
         managed_run.run_dir = Some(config.run_dir.clone());
         managed_run.event_tx = None;
     }

lib/crates/fabro-workflows/README.md

@@ -41,7 +41,7 @@ digraph MyPipeline {
 ### Parsing and Validating a Pipeline
 
 ```rust
-use arc_workflows::pipeline::prepare_pipeline;
+use fabro_workflows::operations::{create, CreateOptions};
 
 let dot_source = r#"digraph Simple {
     graph [goal="Run tests"]
@@ -51,49 +51,24 @@ let dot_source = r#"digraph Simple {
     start -> work -> exit
 }"#;
 
-let graph = prepare_pipeline(dot_source)
-    .expect("pipeline should parse and validate");
+let validated = create(dot_source, CreateOptions::default())
+    .expect("pipeline should parse");
+validated.raise_on_errors().expect("pipeline should validate");
+let (graph, _, _) = validated.into_parts();
 assert_eq!(graph.name, "Simple");
 assert_eq!(graph.goal(), "Run tests");
 ```
 
-`prepare_pipeline` parses the DOT source, applies built-in transforms (variable expansion, stylesheet application, preamble injection), and validates the graph against 14 built-in lint rules.
+`operations::create` parses the DOT source, applies built-in transforms (variable expansion, stylesheet application, preamble injection), and returns diagnostics through `Validated`.
 
 ### Running a Pipeline
 
 ```rust
-use arc_workflows::engine::{PipelineEngine, RunSettings};
-use arc_workflows::event::EventEmitter;
-use arc_workflows::handler::HandlerRegistry;
-use arc_workflows::handler::start::StartHandler;
-use arc_workflows::handler::exit::ExitHandler;
-use arc_workflows::handler::agent::AgentHandler;
-use arc_workflows::pipeline::prepare_pipeline;
+use fabro_workflows::operations::start;
+use fabro_workflows::pipeline;
 
-let graph = prepare_pipeline(dot_source).unwrap();
-
-let mut registry = HandlerRegistry::new(Box::new(AgentHandler::new(None)));
-registry.register("start", Box::new(StartHandler));
-registry.register("exit", Box::new(ExitHandler));
-registry.register("agent", Box::new(AgentHandler::new(None)));
-
-let engine = PipelineEngine::new(registry, EventEmitter::new());
-let config = RunSettings {
-    config: fabro_config::FabroConfig::default(),
-    run_dir: "/tmp/pipeline-run".into(),
-    cancel_token: None,
-    dry_run: false,
-    run_id: "example-run".into(),
-    labels: std::collections::HashMap::new(),
-    git_author: fabro_workflows::git::GitAuthor::default(),
-    workflow_slug: None,
-    github_app: None,
-    base_branch: None,
-    host_repo_path: None,
-    git: None,
-};
-
-// engine.run(&graph, &config).await
+// Use `operations::start(...)` for the full initialize -> execute -> retro -> finalize flow.
+// Use `pipeline::initialize(...)` + `pipeline::execute(...)` when you need partial lifecycle control.
 ```
 
 ### Custom Handlers

lib/crates/fabro-workflows/src/handler/manager_loop.rs

@@ -9,10 +9,11 @@ use async_trait::async_trait;
 use crate::condition::evaluate_condition;
 use crate::context::keys;
 use crate::context::{Context, WorkflowContext};
-use crate::engine::WorkflowRunEngine;
 use crate::error::FabroError;
 use crate::operations::{create, create_from_file, CreateOptions};
 use crate::outcome::{Outcome, OutcomeExt, StageStatus};
+use crate::pipeline;
+use crate::pipeline::types::Initialized;
 use crate::run_settings::RunSettings;
 use fabro_graphviz::graph::{Graph, Node};
 
@@ -171,12 +172,30 @@ impl Handler for SubWorkflowHandler {
         }
         let before_snapshot = context.snapshot();
 
+        let emitter = Arc::clone(&services.emitter);
+        let sandbox = Arc::clone(&services.sandbox);
+        let registry = Arc::clone(&services.registry);
+        let hook_runner = services.hook_runner.clone();
+        let env = services.env.clone();
+        let dry_run = services.dry_run;
+
         // Spawn child engine
-        let engine = WorkflowRunEngine::from_services(services);
         let mut child_handle = tokio::spawn(async move {
-            engine
-                .run_with_context(&child_graph, &child_config, child_context)
-                .await
+            let initialized = Initialized {
+                graph: child_graph,
+                source: String::new(),
+                settings: child_config,
+                checkpoint: None,
+                seed_context: Some(child_context),
+                emitter,
+                sandbox,
+                registry,
+                hook_runner,
+                env,
+                dry_run,
+            };
+            let executed = pipeline::execute(initialized).await;
+            Ok::<_, FabroError>((executed.outcome?, executed.final_context))
         });
 
         // Poll loop

lib/crates/fabro-workflows/src/lib.rs

@@ -99,7 +99,6 @@ pub mod context;
 pub mod core_adapter;
 pub mod cost;
 pub mod devcontainer_bridge;
-pub mod engine;
 pub mod error;
 pub mod event;
 pub mod git;
@@ -112,10 +111,8 @@ pub mod pipeline;
 pub mod preamble;
 pub mod pull_request;
 pub mod run_dir;
-pub mod run_fork;
 pub mod run_lookup;
 pub mod run_record;
-pub mod run_rewind;
 pub mod run_settings;
 pub mod run_status;
 pub mod sandbox_git;
@@ -128,4 +125,3 @@ pub mod stylesheet;
 pub mod test_support;
 pub mod transform;
 pub mod vars;
-pub mod workflow;

lib/crates/fabro-workflows/src/operations/fork.rs

@@ -1 +1,322 @@
-pub use crate::run_fork::execute_fork as fork;
+use anyhow::{Context, Result};
+use fabro_git_storage::branchstore::BranchStore;
+use fabro_git_storage::gitobj::Store;
+use git2::{Oid, Signature};
+
+use crate::git::MetadataStore;
+use crate::run_record::RunRecord;
+use crate::start_record::StartRecord;
+
+use super::rewind::TimelineEntry;
+
+/// Create a new run that branches from an existing run at a specific checkpoint.
+///
+/// Returns the new run ID.
+pub fn fork(
+    store: &Store,
+    source_run_id: &str,
+    entry: &TimelineEntry,
+    push: bool,
+) -> Result<String> {
+    let new_run_id = ulid::Ulid::new().to_string();
+    let sig = Signature::now("Fabro", "noreply@fabro.sh")?;
+
+    let new_run_branch = format!("{}{new_run_id}", crate::git::RUN_BRANCH_PREFIX);
+    match &entry.run_commit_sha {
+        Some(sha) => {
+            let oid =
+                Oid::from_str(sha).with_context(|| format!("invalid run commit SHA: {sha}"))?;
+            store
+                .update_ref(&new_run_branch, oid)
+                .map_err(|e| anyhow::anyhow!("failed to create run branch ref: {e}"))?;
+        }
+        None => {
+            anyhow::bail!(
+                "checkpoint @{} has no git_commit_sha; cannot fork",
+                entry.ordinal
+            );
+        }
+    }
+
+    let source_meta_branch = MetadataStore::branch_name(source_run_id);
+    let new_meta_branch = MetadataStore::branch_name(&new_run_id);
+    let source_bs = BranchStore::new(store, &source_meta_branch, &sig);
+    let new_bs = BranchStore::new(store, &new_meta_branch, &sig);
+
+    new_bs
+        .ensure_branch()
+        .map_err(|e| anyhow::anyhow!("failed to create metadata branch: {e}"))?;
+
+    let source_entries = source_bs
+        .read_entries(&["run.json", "start.json", "sandbox.json"])
+        .map_err(|e| anyhow::anyhow!("failed to read source metadata: {e}"))?;
+
+    let mut run_record_bytes = None;
+    let mut start_record_bytes = None;
+    let mut sandbox_bytes = None;
+    for (path, data) in source_entries {
+        match path {
+            "run.json" => run_record_bytes = Some(data),
+            "start.json" => start_record_bytes = Some(data),
+            "sandbox.json" => sandbox_bytes = Some(data),
+            _ => {}
+        }
+    }
+    let run_record_bytes =
+        run_record_bytes.ok_or_else(|| anyhow::anyhow!("source run has no run.json"))?;
+
+    let now = chrono::Utc::now();
+
+    let mut run_record: RunRecord =
+        serde_json::from_slice(&run_record_bytes).context("failed to parse source run.json")?;
+    run_record.run_id = new_run_id.clone();
+    run_record.created_at = now;
+    let new_run_record_bytes =
+        serde_json::to_vec_pretty(&run_record).context("failed to serialize new run.json")?;
+
+    let new_start_record_bytes = if start_record_bytes.is_some() {
+        let start_record = StartRecord {
+            run_id: new_run_id.clone(),
+            start_time: now,
+            run_branch: Some(new_run_branch.clone()),
+            base_sha: None,
+        };
+        Some(
+            serde_json::to_vec_pretty(&start_record)
+                .context("failed to serialize new start.json")?,
+        )
+    } else {
+        None
+    };
+
+    let checkpoint_bytes = store
+        .read_blob_at(entry.metadata_commit_oid, "checkpoint.json")
+        .map_err(|e| anyhow::anyhow!("failed to read checkpoint blob: {e}"))?
+        .ok_or_else(|| {
+            anyhow::anyhow!(
+                "no checkpoint.json at metadata commit {}",
+                entry.metadata_commit_oid
+            )
+        })?;
+
+    let mut file_entries: Vec<(&str, &[u8])> = vec![
+        ("run.json", &new_run_record_bytes),
+        ("checkpoint.json", &checkpoint_bytes),
+    ];
+    if let Some(ref start_record) = new_start_record_bytes {
+        file_entries.push(("start.json", start_record));
+    }
+    if let Some(ref sandbox) = sandbox_bytes {
+        file_entries.push(("sandbox.json", sandbox));
+    }
+
+    let commit_msg = format!("fork from {} @{}", source_run_id, entry.ordinal);
+    new_bs
+        .write_entries(&file_entries, &commit_msg)
+        .map_err(|e| anyhow::anyhow!("failed to write metadata entries: {e}"))?;
+
+    if push {
+        let repo_path = store
+            .repo()
+            .workdir()
+            .or_else(|| store.repo().path().parent())
+            .unwrap_or(store.repo().path());
+
+        let source_run_branch = format!("{}{source_run_id}", crate::git::RUN_BRANCH_PREFIX);
+        let remote_ref = format!("refs/remotes/origin/{source_run_branch}");
+        let has_remote_tracking = store.repo().find_reference(&remote_ref).is_ok();
+
+        if has_remote_tracking {
+            eprintln!("Pushing new branches to origin...");
+
+            let run_refspec = format!("refs/heads/{new_run_branch}:refs/heads/{new_run_branch}");
+            crate::git::push_branch(repo_path, "origin", &run_refspec)
+                .map_err(|e| anyhow::anyhow!("failed to push run branch: {e}"))?;
+
+            let meta_refspec = format!("refs/heads/{new_meta_branch}:refs/heads/{new_meta_branch}");
+            crate::git::push_branch(repo_path, "origin", &meta_refspec)
+                .map_err(|e| anyhow::anyhow!("failed to push metadata branch: {e}"))?;
+
+            eprintln!("Remote refs updated.");
+        }
+    }
+
+    Ok(new_run_id)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use super::*;
+    use git2::Repository;
+
+    use crate::operations::{build_timeline, find_run_id_by_prefix, parse_target, resolve_target};
+
+    fn temp_repo() -> (tempfile::TempDir, Store) {
+        let dir = tempfile::TempDir::new().unwrap();
+        let repo = Repository::init(dir.path()).unwrap();
+        (dir, Store::new(repo))
+    }
+
+    fn test_sig() -> Signature<'static> {
+        Signature::now("Test", "test@example.com").unwrap()
+    }
+
+    fn make_checkpoint_json(current_node: &str, visit: usize, git_sha: Option<&str>) -> Vec<u8> {
+        let mut node_visits = HashMap::new();
+        node_visits.insert(current_node.to_string(), visit);
+        let cp = serde_json::json!({
+            "timestamp": "2025-01-01T00:00:00Z",
+            "current_node": current_node,
+            "completed_nodes": [current_node],
+            "node_retries": {},
+            "context_values": {},
+            "logs": [],
+            "node_visits": node_visits,
+            "git_commit_sha": git_sha,
+        });
+        serde_json::to_vec(&cp).unwrap()
+    }
+
+    fn make_run_record_json(run_id: &str) -> Vec<u8> {
+        let record = serde_json::json!({
+            "run_id": run_id,
+            "created_at": "2025-01-01T00:00:00Z",
+            "config": {},
+            "graph": {
+                "name": "test_workflow",
+                "nodes": {
+                    "start": {"id": "start", "attrs": {}},
+                    "build": {"id": "build", "attrs": {}},
+                    "test": {"id": "test", "attrs": {}}
+                },
+                "edges": [
+                    {"from": "start", "to": "build", "attrs": {}},
+                    {"from": "build", "to": "test", "attrs": {}}
+                ],
+                "attrs": {}
+            },
+            "working_directory": "/tmp/test",
+        });
+        serde_json::to_vec_pretty(&record).unwrap()
+    }
+
+    fn make_start_record_json(run_id: &str) -> Vec<u8> {
+        let record = serde_json::json!({
+            "run_id": run_id,
+            "start_time": "2025-01-01T00:00:00Z",
+            "run_branch": format!("{}{}", crate::git::RUN_BRANCH_PREFIX, run_id),
+        });
+        serde_json::to_vec_pretty(&record).unwrap()
+    }
+
+    fn setup_source_run(store: &Store, run_id: &str, nodes: &[&str]) -> Vec<Oid> {
+        let sig = test_sig();
+
+        let run_branch = format!("{}{run_id}", crate::git::RUN_BRANCH_PREFIX);
+        let empty_tree = store.write_empty_tree().unwrap();
+        let mut run_oids = Vec::new();
+        let mut parent: Option<Oid> = None;
+
+        for node in nodes {
+            let parents = match parent {
+                Some(p) => vec![p],
+                None => vec![],
+            };
+            let oid = store
+                .write_commit(
+                    empty_tree,
+                    &parents,
+                    &format!("fabro({run_id}): {node} (completed)"),
+                    &sig,
+                )
+                .unwrap();
+            store.update_ref(&run_branch, oid).unwrap();
+            run_oids.push(oid);
+            parent = Some(oid);
+        }
+
+        let meta_branch = MetadataStore::branch_name(run_id);
+        let bs = BranchStore::new(store, &meta_branch, &sig);
+        bs.ensure_branch().unwrap();
+
+        let run_record = make_run_record_json(run_id);
+        let start_record = make_start_record_json(run_id);
+        bs.write_entries(
+            &[("run.json", &run_record), ("start.json", &start_record)],
+            "init run",
+        )
+        .unwrap();
+
+        for (i, node) in nodes.iter().enumerate() {
+            let cp = make_checkpoint_json(node, 1, Some(&run_oids[i].to_string()));
+            bs.write_entry("checkpoint.json", &cp, "checkpoint")
+                .unwrap();
+        }
+
+        run_oids
+    }
+
+    #[test]
+    fn fork_creates_new_run_and_metadata_branches() {
+        let (_dir, store) = temp_repo();
+        let source_run_id = "run-source";
+        let _run_oids = setup_source_run(&store, source_run_id, &["start", "build", "test"]);
+
+        let timeline = build_timeline(&store, source_run_id).unwrap();
+        let entry =
+            resolve_target(&timeline, &parse_target("@2").unwrap(), &HashMap::new()).unwrap();
+
+        let new_run_id = fork(&store, source_run_id, entry, false).unwrap();
+
+        let new_run_branch = format!("{}{new_run_id}", crate::git::RUN_BRANCH_PREFIX);
+        let new_meta_branch = MetadataStore::branch_name(&new_run_id);
+
+        assert!(store.resolve_ref(&new_run_branch).unwrap().is_some());
+        assert!(store.resolve_ref(&new_meta_branch).unwrap().is_some());
+
+        let sig = test_sig();
+        let bs = BranchStore::new(&store, &new_meta_branch, &sig);
+        let run_json = bs.read_entry("run.json").unwrap().unwrap();
+        let run_record: RunRecord = serde_json::from_slice(&run_json).unwrap();
+        assert_eq!(run_record.run_id, new_run_id);
+    }
+
+    #[test]
+    fn fork_rejects_checkpoint_without_run_sha() {
+        let (_dir, store) = temp_repo();
+        let sig = test_sig();
+        let run_id = "run-no-sha";
+        let meta_branch = MetadataStore::branch_name(run_id);
+        let bs = BranchStore::new(&store, &meta_branch, &sig);
+        bs.ensure_branch().unwrap();
+        bs.write_entry("run.json", &make_run_record_json(run_id), "init")
+            .unwrap();
+
+        let cp = make_checkpoint_json("start", 1, None);
+        let oid = bs
+            .write_entry("checkpoint.json", &cp, "checkpoint")
+            .unwrap();
+        let entry = TimelineEntry {
+            ordinal: 1,
+            node_name: "start".to_string(),
+            visit: 1,
+            metadata_commit_oid: oid,
+            run_commit_sha: None,
+        };
+
+        let err = fork(&store, run_id, &entry, false).unwrap_err().to_string();
+        assert!(err.contains("cannot fork"));
+    }
+
+    #[test]
+    fn fork_supports_prefix_resolved_source_run_ids() {
+        let (_dir, store) = temp_repo();
+        let source_run_id = "abc-123-long";
+        setup_source_run(&store, source_run_id, &["start", "build"]);
+
+        let resolved = find_run_id_by_prefix(store.repo(), "abc-123").unwrap();
+        assert_eq!(resolved, source_run_id);
+    }
+}

lib/crates/fabro-workflows/src/operations/rewind.rs

@@ -1,4 +1,517 @@
-pub use crate::run_rewind::{
-    build_timeline, execute_rewind as rewind, find_run_id_by_prefix, load_parallel_map,
-    parse_target, resolve_target, TimelineEntry,
-};
+use std::collections::HashMap;
+
+use anyhow::{bail, Context, Result};
+use fabro_git_storage::branchstore::{BranchStore, CommitInfo};
+use fabro_git_storage::gitobj::Store;
+use git2::{Oid, Repository, Signature};
+
+use crate::checkpoint::Checkpoint;
+use crate::git::MetadataStore;
+use fabro_graphviz::graph::Graph;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum RewindTarget {
+    Ordinal(usize),
+    LatestVisit(String),
+    SpecificVisit(String, usize),
+}
+
+#[derive(Debug, Clone)]
+pub struct TimelineEntry {
+    pub ordinal: usize,
+    pub node_name: String,
+    pub visit: usize,
+    pub metadata_commit_oid: Oid,
+    pub run_commit_sha: Option<String>,
+}
+
+pub fn parse_target(s: &str) -> Result<RewindTarget> {
+    if let Some(rest) = s.strip_prefix('@') {
+        let n: usize = rest
+            .parse()
+            .with_context(|| format!("invalid ordinal: @{rest}"))?;
+        if n == 0 {
+            bail!("ordinal must be >= 1");
+        }
+        return Ok(RewindTarget::Ordinal(n));
+    }
+    if let Some(at_pos) = s.rfind('@') {
+        let name = &s[..at_pos];
+        let visit_str = &s[at_pos + 1..];
+        if !name.is_empty() && !visit_str.is_empty() {
+            if let Ok(visit) = visit_str.parse::<usize>() {
+                if visit == 0 {
+                    bail!("visit number must be >= 1");
+                }
+                return Ok(RewindTarget::SpecificVisit(name.to_string(), visit));
+            }
+        }
+    }
+    Ok(RewindTarget::LatestVisit(s.to_string()))
+}
+
+pub fn build_timeline(store: &Store, run_id: &str) -> Result<Vec<TimelineEntry>> {
+    let branch = MetadataStore::branch_name(run_id);
+    let sig = Signature::now("Fabro", "noreply@fabro.sh")?;
+    let bs = BranchStore::new(store, &branch, &sig);
+
+    let commits = bs
+        .log(10_000)
+        .map_err(|e| anyhow::anyhow!("failed to read metadata branch log: {e}"))?;
+    let commits: Vec<&CommitInfo> = commits.iter().rev().collect();
+
+    let mut timeline = Vec::new();
+    let mut ordinal = 0usize;
+
+    for commit in &commits {
+        if !commit.message.starts_with("checkpoint") {
+            continue;
+        }
+        let blob = store
+            .read_blob_at(commit.oid, "checkpoint.json")
+            .map_err(|e| anyhow::anyhow!("failed to read checkpoint blob: {e}"))?;
+        let Some(bytes) = blob else { continue };
+        let cp: Checkpoint = serde_json::from_slice(&bytes)
+            .with_context(|| format!("failed to parse checkpoint at {}", commit.oid))?;
+
+        ordinal += 1;
+        let visit = cp.node_visits.get(&cp.current_node).copied().unwrap_or(1);
+
+        timeline.push(TimelineEntry {
+            ordinal,
+            node_name: cp.current_node.clone(),
+            visit,
+            metadata_commit_oid: commit.oid,
+            run_commit_sha: cp.git_commit_sha.clone(),
+        });
+    }
+
+    backfill_run_shas(store, run_id, &mut timeline);
+    Ok(timeline)
+}
+
+fn backfill_run_shas(store: &Store, run_id: &str, timeline: &mut [TimelineEntry]) {
+    if !timeline.iter().any(|e| e.run_commit_sha.is_none()) {
+        return;
+    }
+
+    let run_branch = format!("{}{run_id}", crate::git::RUN_BRANCH_PREFIX);
+    let sig = match Signature::now("Fabro", "noreply@fabro.sh") {
+        Ok(s) => s,
+        Err(_) => return,
+    };
+    let bs = BranchStore::new(store, &run_branch, &sig);
+    let run_commits = match bs.log(10_000) {
+        Ok(c) => c,
+        Err(_) => return,
+    };
+
+    let prefix = format!("fabro({run_id}): ");
+    let mut node_commits: HashMap<String, Vec<String>> = HashMap::new();
+    for commit in &run_commits {
+        if let Some(rest) = commit.message.strip_prefix(&prefix) {
+            if let Some(node_name) = rest.split_whitespace().next() {
+                node_commits
+                    .entry(node_name.to_string())
+                    .or_default()
+                    .push(commit.oid.to_string());
+            }
+        }
+    }
+
+    for shas in node_commits.values_mut() {
+        shas.reverse();
+    }
+    let mut node_indices: HashMap<String, usize> = HashMap::new();
+
+    for entry in timeline.iter_mut() {
+        if entry.run_commit_sha.is_some() {
+            continue;
+        }
+        if let Some(shas) = node_commits.get(&entry.node_name) {
+            let idx = node_indices.entry(entry.node_name.clone()).or_insert(0);
+            if *idx < shas.len() {
+                entry.run_commit_sha = Some(shas[*idx].clone());
+                *idx += 1;
+            }
+        }
+    }
+}
+
+pub fn detect_parallel_interior(graph: &Graph) -> HashMap<String, String> {
+    let mut interior_map = HashMap::new();
+
+    for node in graph.nodes.values() {
+        if node.handler_type() != Some("parallel") {
+            continue;
+        }
+        let parallel_id = &node.id;
+        let mut queue: Vec<String> = graph
+            .outgoing_edges(parallel_id)
+            .iter()
+            .map(|e| e.to.clone())
+            .collect();
+        let mut visited = std::collections::HashSet::new();
+
+        while let Some(current) = queue.pop() {
+            if !visited.insert(current.clone()) {
+                continue;
+            }
+            if let Some(n) = graph.nodes.get(&current) {
+                if n.handler_type() == Some("parallel.fan_in") {
+                    continue;
+                }
+            }
+            interior_map.insert(current.clone(), parallel_id.clone());
+            for edge in graph.outgoing_edges(&current) {
+                queue.push(edge.to.clone());
+            }
+        }
+    }
+
+    interior_map
+}
+
+pub fn resolve_target<'a>(
+    timeline: &'a [TimelineEntry],
+    target: &RewindTarget,
+    parallel_map: &HashMap<String, String>,
+) -> Result<&'a TimelineEntry> {
+    match target {
+        RewindTarget::Ordinal(n) => timeline
+            .iter()
+            .find(|e| e.ordinal == *n)
+            .ok_or_else(|| anyhow::anyhow!("ordinal @{n} out of range (max @{})", timeline.len())),
+        RewindTarget::LatestVisit(name) => {
+            let effective_name = parallel_map.get(name).unwrap_or(name);
+            timeline
+                .iter()
+                .rev()
+                .find(|e| e.node_name == *effective_name)
+                .ok_or_else(|| {
+                    if effective_name != name {
+                        anyhow::anyhow!(
+                            "node '{name}' is inside parallel '{effective_name}'; \
+                             no checkpoint found for '{effective_name}'"
+                        )
+                    } else {
+                        anyhow::anyhow!("no checkpoint found for node '{name}'")
+                    }
+                })
+        }
+        RewindTarget::SpecificVisit(name, visit) => {
+            let effective_name = parallel_map.get(name).unwrap_or(name);
+            timeline
+                .iter()
+                .find(|e| e.node_name == *effective_name && e.visit == *visit)
+                .ok_or_else(|| {
+                    if effective_name != name {
+                        anyhow::anyhow!(
+                            "node '{name}' is inside parallel '{effective_name}'; \
+                             no visit {visit} found for '{effective_name}'"
+                        )
+                    } else {
+                        anyhow::anyhow!("no visit {visit} found for node '{name}'")
+                    }
+                })
+        }
+    }
+}
+
+pub fn rewind(store: &Store, run_id: &str, entry: &TimelineEntry, push: bool) -> Result<()> {
+    let meta_branch = MetadataStore::branch_name(run_id);
+    store
+        .update_ref(&meta_branch, entry.metadata_commit_oid)
+        .map_err(|e| anyhow::anyhow!("failed to update metadata ref: {e}"))?;
+    eprintln!(
+        "Rewound metadata branch to @{} ({})",
+        entry.ordinal, entry.node_name
+    );
+
+    let run_branch = format!("{}{run_id}", crate::git::RUN_BRANCH_PREFIX);
+    match &entry.run_commit_sha {
+        Some(sha) => {
+            let oid =
+                Oid::from_str(sha).with_context(|| format!("invalid run commit SHA: {sha}"))?;
+            store
+                .update_ref(&run_branch, oid)
+                .map_err(|e| anyhow::anyhow!("failed to update run branch ref: {e}"))?;
+            eprintln!(
+                "Rewound run branch {}{run_id} to {}",
+                crate::git::RUN_BRANCH_PREFIX,
+                &sha[..8]
+            );
+        }
+        None => {
+            eprintln!(
+                "Warning: checkpoint @{} has no git_commit_sha; run branch not moved",
+                entry.ordinal
+            );
+        }
+    }
+
+    if push {
+        let repo_path = store
+            .repo()
+            .workdir()
+            .or_else(|| store.repo().path().parent())
+            .unwrap_or(store.repo().path());
+
+        let remote_ref = format!("refs/remotes/origin/{run_branch}");
+        let has_remote_tracking = store.repo().find_reference(&remote_ref).is_ok();
+
+        if has_remote_tracking {
+            eprintln!("Force-pushing rewound branches to origin...");
+
+            if entry.run_commit_sha.is_some() {
+                let refspec = format!("+refs/heads/{run_branch}:refs/heads/{run_branch}");
+                crate::git::push_branch(repo_path, "origin", &refspec)
+                    .map_err(|e| anyhow::anyhow!("failed to push run branch: {e}"))?;
+            }
+
+            let meta_refspec = format!("+refs/heads/{meta_branch}:refs/heads/{meta_branch}");
+            crate::git::push_branch(repo_path, "origin", &meta_refspec)
+                .map_err(|e| anyhow::anyhow!("failed to push metadata branch: {e}"))?;
+
+            eprintln!("Remote refs updated.");
+        }
+    }
+
+    Ok(())
+}
+
+pub fn find_run_id_by_prefix(repo: &Repository, prefix: &str) -> Result<String> {
+    let refs = repo.references()?;
+    let pattern = "refs/heads/fabro/meta/";
+    let mut matches = Vec::new();
+
+    for reference in refs.flatten() {
+        let name = match reference.name() {
+            Some(n) => n,
+            None => continue,
+        };
+        if let Some(run_id) = name.strip_prefix(pattern) {
+            if run_id == prefix {
+                return Ok(run_id.to_string());
+            }
+            if run_id.starts_with(prefix) {
+                matches.push(run_id.to_string());
+            }
+        }
+    }
+
+    match matches.len() {
+        0 => bail!("no run found matching '{prefix}'"),
+        1 => Ok(matches.into_iter().next().unwrap()),
+        _ => {
+            let mut msg = format!("ambiguous run ID prefix '{prefix}', matches:\n");
+            for m in &matches {
+                msg.push_str(&format!("  {m}\n"));
+            }
+            bail!("{msg}")
+        }
+    }
+}
+
+pub fn load_parallel_map(store: &Store, run_id: &str) -> HashMap<String, String> {
+    let branch = MetadataStore::branch_name(run_id);
+    let sig = match Signature::now("Fabro", "noreply@fabro.sh") {
+        Ok(s) => s,
+        Err(_) => return HashMap::new(),
+    };
+    let bs = BranchStore::new(store, &branch, &sig);
+
+    if let Ok(Some(run_bytes)) = bs.read_entry("run.json") {
+        if let Ok(record) = serde_json::from_slice::<crate::run_record::RunRecord>(&run_bytes) {
+            return detect_parallel_interior(&record.graph);
+        }
+    }
+
+    let graph_bytes = match bs.read_entry("graph.fabro") {
+        Ok(Some(bytes)) => bytes,
+        _ => return HashMap::new(),
+    };
+    let dot_source = String::from_utf8_lossy(&graph_bytes);
+    let graph = match fabro_graphviz::parser::parse(&dot_source) {
+        Ok(g) => g,
+        Err(_) => return HashMap::new(),
+    };
+    detect_parallel_interior(&graph)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn temp_repo() -> (tempfile::TempDir, Store) {
+        let dir = tempfile::TempDir::new().unwrap();
+        let repo = Repository::init(dir.path()).unwrap();
+        (dir, Store::new(repo))
+    }
+
+    fn test_sig() -> Signature<'static> {
+        Signature::now("Test", "test@example.com").unwrap()
+    }
+
+    fn make_checkpoint_json(current_node: &str, visit: usize, git_sha: Option<&str>) -> Vec<u8> {
+        let mut node_visits = HashMap::new();
+        node_visits.insert(current_node.to_string(), visit);
+        let cp = serde_json::json!({
+            "timestamp": "2025-01-01T00:00:00Z",
+            "current_node": current_node,
+            "completed_nodes": [current_node],
+            "node_retries": {},
+            "context_values": {},
+            "logs": [],
+            "node_visits": node_visits,
+            "git_commit_sha": git_sha,
+        });
+        serde_json::to_vec(&cp).unwrap()
+    }
+
+    #[test]
+    fn parse_target_ordinal() {
+        assert_eq!(parse_target("@4").unwrap(), RewindTarget::Ordinal(4));
+    }
+
+    #[test]
+    fn parse_target_latest_visit() {
+        assert_eq!(
+            parse_target("step2").unwrap(),
+            RewindTarget::LatestVisit("step2".to_string())
+        );
+    }
+
+    #[test]
+    fn build_timeline_simple() {
+        let (_dir, store) = temp_repo();
+        let sig = test_sig();
+        let branch = MetadataStore::branch_name("test-run-1");
+        let bs = BranchStore::new(&store, &branch, &sig);
+        bs.ensure_branch().unwrap();
+
+        bs.write_entry("run.json", b"{}", "init run").unwrap();
+        let cp1 = make_checkpoint_json("start", 1, Some("aaa"));
+        bs.write_entry("checkpoint.json", &cp1, "checkpoint")
+            .unwrap();
+        let cp2 = make_checkpoint_json("build", 1, Some("bbb"));
+        bs.write_entry("checkpoint.json", &cp2, "checkpoint")
+            .unwrap();
+
+        let timeline = build_timeline(&store, "test-run-1").unwrap();
+        assert_eq!(timeline.len(), 2);
+        assert_eq!(timeline[0].node_name, "start");
+        assert_eq!(timeline[1].node_name, "build");
+    }
+
+    #[test]
+    fn resolve_latest_visit() {
+        let timeline = vec![
+            TimelineEntry {
+                ordinal: 1,
+                node_name: "start".to_string(),
+                visit: 1,
+                metadata_commit_oid: Oid::zero(),
+                run_commit_sha: Some("aaa".to_string()),
+            },
+            TimelineEntry {
+                ordinal: 2,
+                node_name: "build".to_string(),
+                visit: 1,
+                metadata_commit_oid: Oid::zero(),
+                run_commit_sha: Some("bbb".to_string()),
+            },
+            TimelineEntry {
+                ordinal: 3,
+                node_name: "build".to_string(),
+                visit: 2,
+                metadata_commit_oid: Oid::zero(),
+                run_commit_sha: Some("ccc".to_string()),
+            },
+        ];
+
+        let entry = resolve_target(
+            &timeline,
+            &RewindTarget::LatestVisit("build".to_string()),
+            &HashMap::new(),
+        )
+        .unwrap();
+        assert_eq!(entry.ordinal, 3);
+    }
+
+    #[test]
+    fn parallel_interior_detection() {
+        let mut graph = Graph::new("test");
+        let mut parallel_node = fabro_graphviz::graph::Node::new("parallel1");
+        parallel_node.attrs.insert(
+            "shape".to_string(),
+            fabro_graphviz::graph::AttrValue::String("component".to_string()),
+        );
+        graph.nodes.insert("parallel1".to_string(), parallel_node);
+
+        let mut fan_in = fabro_graphviz::graph::Node::new("fan_in1");
+        fan_in.attrs.insert(
+            "shape".to_string(),
+            fabro_graphviz::graph::AttrValue::String("tripleoctagon".to_string()),
+        );
+        graph.nodes.insert("fan_in1".to_string(), fan_in);
+
+        let mut a = fabro_graphviz::graph::Node::new("a");
+        a.attrs.insert(
+            "shape".to_string(),
+            fabro_graphviz::graph::AttrValue::String("box".to_string()),
+        );
+        graph.nodes.insert("a".to_string(), a);
+
+        graph.edges.push(fabro_graphviz::graph::Edge {
+            from: "parallel1".to_string(),
+            to: "a".to_string(),
+            attrs: HashMap::new(),
+        });
+        graph.edges.push(fabro_graphviz::graph::Edge {
+            from: "a".to_string(),
+            to: "fan_in1".to_string(),
+            attrs: HashMap::new(),
+        });
+
+        let map = detect_parallel_interior(&graph);
+        assert_eq!(map.get("a"), Some(&"parallel1".to_string()));
+        assert!(!map.contains_key("parallel1"));
+    }
+
+    #[test]
+    fn rewind_moves_metadata_ref() {
+        let (_dir, store) = temp_repo();
+        let sig = test_sig();
+        let branch = MetadataStore::branch_name("run-1");
+        let bs = BranchStore::new(&store, &branch, &sig);
+        bs.ensure_branch().unwrap();
+
+        bs.write_entry("run.json", b"{}", "init run").unwrap();
+        let cp1 = make_checkpoint_json("start", 1, None);
+        let oid1 = bs
+            .write_entry("checkpoint.json", &cp1, "checkpoint")
+            .unwrap();
+        let cp2 = make_checkpoint_json("build", 1, None);
+        bs.write_entry("checkpoint.json", &cp2, "checkpoint")
+            .unwrap();
+
+        let timeline = build_timeline(&store, "run-1").unwrap();
+        rewind(&store, "run-1", &timeline[0], false).unwrap();
+
+        let resolved = store.resolve_ref(&branch).unwrap().unwrap();
+        assert_eq!(resolved, oid1);
+    }
+
+    #[test]
+    fn find_run_id_prefix_match() {
+        let (_dir, store) = temp_repo();
+        let sig = test_sig();
+        let branch = MetadataStore::branch_name("abc-123-long-id");
+        let bs = BranchStore::new(&store, &branch, &sig);
+        bs.ensure_branch().unwrap();
+
+        let result = find_run_id_by_prefix(store.repo(), "abc-123").unwrap();
+        assert_eq!(result, "abc-123-long-id");
+    }
+}

lib/crates/fabro-workflows/src/pipeline/execute.rs

@@ -395,3 +395,7 @@ mod tests {
         );
     }
 }
+
+#[cfg(test)]
+#[path = "execute_engine_compat_tests.rs"]
+mod engine_compat_tests;

lib/crates/fabro-workflows/src/pipeline/execute_engine_compat_tests.rs

@@ -0,0 +1,850 @@
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use fabro_agent::Sandbox;
+use fabro_config::config::FabroConfig;
+use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
+use fabro_hooks::HookConfig;
+
+use crate::checkpoint::Checkpoint;
+use crate::context::{self, Context};
+use crate::error::FabroError;
+use crate::event::{EventEmitter, WorkflowRunEvent};
+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;
+use crate::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings};
+use crate::test_support::run_graph;
+
+fn local_env() -> Arc<dyn Sandbox> {
+    Arc::new(fabro_agent::LocalSandbox::new(
+        std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")),
+    ))
+}
+
+fn simple_graph() -> Graph {
+    let mut g = Graph::new("test_pipeline");
+    g.attrs.insert(
+        "goal".to_string(),
+        AttrValue::String("Run tests".to_string()),
+    );
+
+    let mut start = Node::new("start");
+    start.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Mdiamond".to_string()),
+    );
+    g.nodes.insert("start".to_string(), start);
+
+    let mut exit = Node::new("exit");
+    exit.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Msquare".to_string()),
+    );
+    g.nodes.insert("exit".to_string(), exit);
+
+    g.edges.push(Edge::new("start", "exit"));
+    g
+}
+
+fn make_registry() -> HandlerRegistry {
+    use crate::handler::exit::ExitHandler;
+
+    let mut registry = HandlerRegistry::new(Box::new(StartHandler));
+    registry.register("start", Box::new(StartHandler));
+    registry.register("exit", Box::new(ExitHandler));
+    registry
+}
+
+fn test_settings(run_dir: &Path, run_id: &str) -> RunSettings {
+    RunSettings {
+        run_dir: run_dir.to_path_buf(),
+        cancel_token: None,
+        dry_run: false,
+        run_id: run_id.into(),
+        config: FabroConfig::default(),
+        git: None,
+        host_repo_path: None,
+        labels: HashMap::new(),
+        github_app: None,
+        git_author: crate::git::GitAuthor::default(),
+        base_branch: None,
+        workflow_slug: None,
+    }
+}
+
+fn test_lifecycle(setup_commands: Vec<String>) -> LifecycleConfig {
+    LifecycleConfig {
+        setup_commands,
+        setup_command_timeout_ms: 300_000,
+        devcontainer_phases: Vec::new(),
+    }
+}
+
+async fn run_with_lifecycle(
+    registry: HandlerRegistry,
+    emitter: Arc<EventEmitter>,
+    sandbox: Arc<dyn Sandbox>,
+    graph: &Graph,
+    settings: RunSettings,
+    lifecycle: LifecycleConfig,
+) -> 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());
+    let initialized = initialize(
+        validated,
+        InitOptions {
+            run_id,
+            run_dir,
+            dry_run: settings.dry_run,
+            emitter,
+            sandbox,
+            registry: Arc::new(registry),
+            lifecycle,
+            run_settings: settings,
+            hooks: HookConfig { hooks: vec![] },
+            sandbox_env: HashMap::new(),
+            checkpoint: None,
+            seed_context: None,
+        },
+    )
+    .await?;
+    super::execute(initialized).await.outcome
+}
+
+struct AlwaysFailHandler;
+
+#[async_trait]
+impl HandlerTrait for AlwaysFailHandler {
+    async fn execute(
+        &self,
+        _node: &Node,
+        _context: &Context,
+        _graph: &Graph,
+        _run_dir: &Path,
+        _services: &crate::handler::EngineServices,
+    ) -> std::result::Result<Outcome, FabroError> {
+        Ok(Outcome::fail_classify("always fails"))
+    }
+}
+
+struct SlowHandler {
+    sleep_ms: u64,
+}
+
+#[async_trait]
+impl HandlerTrait for SlowHandler {
+    async fn execute(
+        &self,
+        _node: &Node,
+        _context: &Context,
+        _graph: &Graph,
+        _run_dir: &Path,
+        _services: &crate::handler::EngineServices,
+    ) -> std::result::Result<Outcome, FabroError> {
+        tokio::time::sleep(Duration::from_millis(self.sleep_ms)).await;
+        Ok(Outcome::success())
+    }
+}
+
+struct PanickingHandler;
+
+#[async_trait]
+impl HandlerTrait for PanickingHandler {
+    async fn execute(
+        &self,
+        _node: &Node,
+        _context: &Context,
+        _graph: &Graph,
+        _run_dir: &Path,
+        _services: &crate::handler::EngineServices,
+    ) -> std::result::Result<Outcome, FabroError> {
+        panic!("test panic message");
+    }
+}
+
+struct FailOnceThenSucceedHandler {
+    call_count: AtomicU32,
+}
+
+#[async_trait]
+impl HandlerTrait for FailOnceThenSucceedHandler {
+    async fn execute(
+        &self,
+        _node: &Node,
+        _context: &Context,
+        _graph: &Graph,
+        _run_dir: &Path,
+        _services: &crate::handler::EngineServices,
+    ) -> std::result::Result<Outcome, FabroError> {
+        if self.call_count.fetch_add(1, Ordering::Relaxed) == 0 {
+            Err(FabroError::handler("transient failure"))
+        } else {
+            Ok(Outcome::success())
+        }
+    }
+}
+
+fn cyclic_graph() -> Graph {
+    let mut g = Graph::new("cyclic");
+    g.attrs
+        .insert("goal".to_string(), AttrValue::String("loop".to_string()));
+    g.attrs
+        .insert("default_max_retries".to_string(), AttrValue::Integer(0));
+
+    let mut start = Node::new("start");
+    start.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Mdiamond".to_string()),
+    );
+    g.nodes.insert("start".to_string(), start);
+    g.nodes.insert("work".to_string(), Node::new("work"));
+
+    let mut exit = Node::new("exit");
+    exit.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Msquare".to_string()),
+    );
+    g.nodes.insert("exit".to_string(), exit);
+
+    g.edges.push(Edge::new("start", "work"));
+    let mut cond_edge = Edge::new("work", "exit");
+    cond_edge.attrs.insert(
+        "condition".to_string(),
+        AttrValue::String("outcome=never_matches".to_string()),
+    );
+    g.edges.push(cond_edge);
+    g.edges.push(Edge::new("work", "work"));
+    g
+}
+
+fn looping_fail_graph() -> Graph {
+    let mut g = Graph::new("loop_fail");
+    g.attrs
+        .insert("goal".to_string(), AttrValue::String("test".to_string()));
+    g.attrs
+        .insert("default_max_retries".to_string(), AttrValue::Integer(0));
+
+    let mut start = Node::new("start");
+    start.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Mdiamond".to_string()),
+    );
+    g.nodes.insert("start".to_string(), start);
+
+    let mut work = Node::new("work");
+    work.attrs.insert(
+        "type".to_string(),
+        AttrValue::String("always_fail".to_string()),
+    );
+    work.attrs
+        .insert("max_retries".to_string(), AttrValue::Integer(0));
+    g.nodes.insert("work".to_string(), work);
+
+    let mut exit = Node::new("exit");
+    exit.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Msquare".to_string()),
+    );
+    g.nodes.insert("exit".to_string(), exit);
+
+    g.edges.push(Edge::new("start", "work"));
+    let mut fail_edge = Edge::new("work", "work");
+    fail_edge.attrs.insert(
+        "condition".to_string(),
+        AttrValue::String("outcome=fail".to_string()),
+    );
+    g.edges.push(fail_edge);
+    let mut ok_edge = Edge::new("work", "exit");
+    ok_edge.attrs.insert(
+        "condition".to_string(),
+        AttrValue::String("outcome=success".to_string()),
+    );
+    g.edges.push(ok_edge);
+    g
+}
+
+#[tokio::test]
+async fn execute_runs_simple_workflow() {
+    let dir = tempfile::tempdir().unwrap();
+    let outcome = run_graph(
+        make_registry(),
+        Arc::new(EventEmitter::new()),
+        local_env(),
+        &simple_graph(),
+        &test_settings(dir.path(), "test-run"),
+    )
+    .await
+    .unwrap();
+    assert_eq!(outcome.status, StageStatus::Success);
+}
+
+#[tokio::test]
+async fn execute_saves_checkpoint() {
+    let dir = tempfile::tempdir().unwrap();
+    run_graph(
+        make_registry(),
+        Arc::new(EventEmitter::new()),
+        local_env(),
+        &simple_graph(),
+        &test_settings(dir.path(), "test-run"),
+    )
+    .await
+    .unwrap();
+    assert!(dir.path().join("checkpoint.json").exists());
+}
+
+#[tokio::test]
+async fn execute_emits_events() {
+    let dir = tempfile::tempdir().unwrap();
+    let events = Arc::new(std::sync::Mutex::new(Vec::new()));
+    let events_clone = Arc::clone(&events);
+    let emitter = EventEmitter::new();
+    emitter.on_event(move |event| {
+        events_clone.lock().unwrap().push(format!("{event:?}"));
+    });
+
+    run_graph(
+        make_registry(),
+        Arc::new(emitter),
+        local_env(),
+        &simple_graph(),
+        &test_settings(dir.path(), "test-run"),
+    )
+    .await
+    .unwrap();
+
+    assert!(events.lock().unwrap().len() >= 4);
+}
+
+#[tokio::test]
+async fn execute_error_when_no_start_node() {
+    let dir = tempfile::tempdir().unwrap();
+    let result = run_graph(
+        make_registry(),
+        Arc::new(EventEmitter::new()),
+        local_env(),
+        &Graph::new("empty"),
+        &test_settings(dir.path(), "test-run"),
+    )
+    .await;
+    assert!(result.is_err());
+}
+
+#[tokio::test]
+async fn execute_mirrors_graph_goal_to_context() {
+    let dir = tempfile::tempdir().unwrap();
+    run_graph(
+        make_registry(),
+        Arc::new(EventEmitter::new()),
+        local_env(),
+        &simple_graph(),
+        &test_settings(dir.path(), "test-run"),
+    )
+    .await
+    .unwrap();
+
+    let cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap();
+    assert_eq!(
+        cp.context_values.get(context::keys::GRAPH_GOAL),
+        Some(&serde_json::json!("Run tests"))
+    );
+}
+
+#[tokio::test]
+async fn execute_conditional_routing_uses_unconditional_success_path() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut g = Graph::new("cond_test");
+
+    let mut start = Node::new("start");
+    start.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Mdiamond".to_string()),
+    );
+    g.nodes.insert("start".to_string(), start);
+
+    let mut exit = Node::new("exit");
+    exit.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Msquare".to_string()),
+    );
+    g.nodes.insert("exit".to_string(), exit);
+
+    g.nodes.insert("path_a".to_string(), Node::new("path_a"));
+    g.nodes.insert("path_b".to_string(), Node::new("path_b"));
+
+    let mut e1 = Edge::new("start", "path_a");
+    e1.attrs.insert(
+        "condition".to_string(),
+        AttrValue::String("outcome=fail".to_string()),
+    );
+    g.edges.push(e1);
+    g.edges.push(Edge::new("start", "path_b"));
+    g.edges.push(Edge::new("path_a", "exit"));
+    g.edges.push(Edge::new("path_b", "exit"));
+
+    run_graph(
+        make_registry(),
+        Arc::new(EventEmitter::new()),
+        local_env(),
+        &g,
+        &test_settings(dir.path(), "test-run"),
+    )
+    .await
+    .unwrap();
+
+    let cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap();
+    assert!(cp.completed_nodes.contains(&"path_b".to_string()));
+    assert!(!cp.completed_nodes.contains(&"path_a".to_string()));
+}
+
+#[tokio::test]
+async fn execute_writes_start_json_and_node_status() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut settings = test_settings(dir.path(), "test-run");
+    settings.git = Some(GitCheckpointSettings {
+        base_sha: Some("abc123".into()),
+        run_branch: Some("fabro/run/test-run".into()),
+        meta_branch: None,
+    });
+
+    run_graph(
+        make_registry(),
+        Arc::new(EventEmitter::new()),
+        local_env(),
+        &simple_graph(),
+        &settings,
+    )
+    .await
+    .unwrap();
+
+    let start = crate::start_record::StartRecord::load(dir.path()).unwrap();
+    assert_eq!(start.run_id, "test-run");
+    assert_eq!(start.run_branch.as_deref(), Some("fabro/run/test-run"));
+    assert_eq!(start.base_sha.as_deref(), Some("abc123"));
+
+    let status_path = dir.path().join("nodes").join("start").join("status.json");
+    assert!(status_path.exists());
+}
+
+#[tokio::test]
+async fn timeout_causes_fail_status_json() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut g = Graph::new("timeout_test");
+
+    let mut start = Node::new("start");
+    start.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Mdiamond".to_string()),
+    );
+    g.nodes.insert("start".to_string(), start);
+
+    let mut work = Node::new("work");
+    work.attrs.insert(
+        "timeout".to_string(),
+        AttrValue::Duration(Duration::from_millis(50)),
+    );
+    work.attrs
+        .insert("type".to_string(), AttrValue::String("slow".to_string()));
+    work.attrs
+        .insert("max_retries".to_string(), AttrValue::Integer(0));
+    g.nodes.insert("work".to_string(), work);
+
+    let mut exit = Node::new("exit");
+    exit.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Msquare".to_string()),
+    );
+    g.nodes.insert("exit".to_string(), exit);
+
+    g.edges.push(Edge::new("start", "work"));
+    let mut fail_edge = Edge::new("work", "exit");
+    fail_edge.attrs.insert(
+        "condition".to_string(),
+        AttrValue::String("outcome=fail".to_string()),
+    );
+    g.edges.push(fail_edge);
+
+    let mut registry = make_registry();
+    registry.register("slow", Box::new(SlowHandler { sleep_ms: 500 }));
+    run_graph(
+        registry,
+        Arc::new(EventEmitter::new()),
+        local_env(),
+        &g,
+        &test_settings(dir.path(), "test-run"),
+    )
+    .await
+    .unwrap();
+
+    let status_path = dir.path().join("nodes").join("work").join("status.json");
+    let status: serde_json::Value =
+        serde_json::from_str(&std::fs::read_to_string(&status_path).unwrap()).unwrap();
+    assert_eq!(status["status"], "fail");
+}
+
+#[tokio::test]
+async fn execute_cancelled_mid_run() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut g = simple_graph();
+    let mut work = Node::new("work");
+    work.attrs
+        .insert("type".to_string(), AttrValue::String("slow".to_string()));
+    work.attrs
+        .insert("max_retries".to_string(), AttrValue::Integer(0));
+    g.nodes.insert("work".to_string(), work);
+    g.edges.clear();
+    g.edges.push(Edge::new("start", "work"));
+    g.edges.push(Edge::new("work", "exit"));
+
+    let cancel_token = Arc::new(AtomicBool::new(false));
+    let cancel_token_clone = Arc::clone(&cancel_token);
+    let mut registry = make_registry();
+    registry.register("slow", Box::new(SlowHandler { sleep_ms: 200 }));
+    let mut settings = test_settings(dir.path(), "test-run");
+    settings.cancel_token = Some(cancel_token);
+
+    tokio::spawn(async move {
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        cancel_token_clone.store(true, Ordering::Relaxed);
+    });
+
+    let result = run_graph(
+        registry,
+        Arc::new(EventEmitter::new()),
+        local_env(),
+        &g,
+        &settings,
+    )
+    .await;
+    assert!(matches!(result, Err(FabroError::Cancelled)));
+}
+
+#[tokio::test]
+async fn max_node_visits_errors_on_cycle() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut g = cyclic_graph();
+    g.attrs
+        .insert("max_node_visits".to_string(), AttrValue::Integer(3));
+
+    let result = run_graph(
+        make_registry(),
+        Arc::new(EventEmitter::new()),
+        local_env(),
+        &g,
+        &test_settings(dir.path(), "test-run"),
+    )
+    .await;
+    let err = result.unwrap_err().to_string();
+    assert!(err.contains("stuck in a cycle"));
+}
+
+#[tokio::test]
+async fn panic_handler_writes_panic_txt() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut g = Graph::new("panic_test");
+    let mut start = Node::new("start");
+    start.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Mdiamond".to_string()),
+    );
+    g.nodes.insert("start".to_string(), start);
+    let mut panic_node = Node::new("boom");
+    panic_node.attrs.insert(
+        "type".to_string(),
+        AttrValue::String("panicker".to_string()),
+    );
+    panic_node
+        .attrs
+        .insert("max_retries".to_string(), AttrValue::Integer(0));
+    g.nodes.insert("boom".to_string(), panic_node);
+    g.edges.push(Edge::new("start", "boom"));
+
+    let mut registry = make_registry();
+    registry.register("panicker", Box::new(PanickingHandler));
+    let _ = run_graph(
+        registry,
+        Arc::new(EventEmitter::new()),
+        local_env(),
+        &g,
+        &test_settings(dir.path(), "test-run"),
+    )
+    .await;
+
+    let panic_path = dir.path().join("nodes").join("boom").join("panic.txt");
+    assert!(panic_path.exists());
+    let content = std::fs::read_to_string(&panic_path).unwrap();
+    assert!(content.contains("test panic message"));
+}
+
+#[tokio::test]
+async fn loop_circuit_breaker_aborts_on_repeated_failure() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut registry = make_registry();
+    registry.register("always_fail", Box::new(AlwaysFailHandler));
+
+    let result = run_graph(
+        registry,
+        Arc::new(EventEmitter::new()),
+        local_env(),
+        &looping_fail_graph(),
+        &test_settings(dir.path(), "test-run"),
+    )
+    .await;
+    let err = result.unwrap_err().to_string();
+    assert!(err.contains("deterministic failure cycle detected"));
+}
+
+#[tokio::test]
+async fn stall_watchdog_triggers_on_hung_handler() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut g = Graph::new("stall_test");
+    g.attrs
+        .insert("goal".to_string(), AttrValue::String("test".to_string()));
+    g.attrs.insert(
+        "stall_timeout".to_string(),
+        AttrValue::Duration(Duration::from_millis(50)),
+    );
+    g.attrs
+        .insert("default_max_retries".to_string(), AttrValue::Integer(0));
+
+    let mut start = Node::new("start");
+    start.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Mdiamond".to_string()),
+    );
+    g.nodes.insert("start".to_string(), start);
+
+    let mut work = Node::new("work");
+    work.attrs
+        .insert("type".to_string(), AttrValue::String("slow".to_string()));
+    g.nodes.insert("work".to_string(), work);
+
+    let mut exit = Node::new("exit");
+    exit.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Msquare".to_string()),
+    );
+    g.nodes.insert("exit".to_string(), exit);
+
+    g.edges.push(Edge::new("start", "work"));
+    g.edges.push(Edge::new("work", "exit"));
+
+    let mut registry = make_registry();
+    registry.register("slow", Box::new(SlowHandler { sleep_ms: 60_000 }));
+    let result = run_graph(
+        registry,
+        Arc::new(EventEmitter::new()),
+        local_env(),
+        &g,
+        &test_settings(dir.path(), "test-run"),
+    )
+    .await;
+    let err = result.unwrap_err().to_string();
+    assert!(err.contains("stall watchdog"));
+}
+
+#[tokio::test]
+async fn retry_emits_stage_started_per_attempt() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut g = Graph::new("retry_events");
+    g.attrs
+        .insert("goal".to_string(), AttrValue::String("test".to_string()));
+
+    let mut start = Node::new("start");
+    start.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Mdiamond".to_string()),
+    );
+    g.nodes.insert("start".to_string(), start);
+
+    let mut work = Node::new("work");
+    work.attrs.insert(
+        "type".to_string(),
+        AttrValue::String("fail_once".to_string()),
+    );
+    work.attrs
+        .insert("max_retries".to_string(), AttrValue::Integer(1));
+    work.attrs.insert(
+        "retry_policy".to_string(),
+        AttrValue::String("aggressive".to_string()),
+    );
+    g.nodes.insert("work".to_string(), work);
+
+    let mut exit = Node::new("exit");
+    exit.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Msquare".to_string()),
+    );
+    g.nodes.insert("exit".to_string(), exit);
+
+    g.edges.push(Edge::new("start", "work"));
+    g.edges.push(Edge::new("work", "exit"));
+
+    let events = Arc::new(std::sync::Mutex::new(Vec::<WorkflowRunEvent>::new()));
+    let events_clone = Arc::clone(&events);
+    let emitter = EventEmitter::new();
+    emitter.on_event(move |event| {
+        events_clone.lock().unwrap().push(event.clone());
+    });
+
+    let mut registry = make_registry();
+    registry.register(
+        "fail_once",
+        Box::new(FailOnceThenSucceedHandler {
+            call_count: AtomicU32::new(0),
+        }),
+    );
+
+    let outcome = run_graph(
+        registry,
+        Arc::new(emitter),
+        local_env(),
+        &g,
+        &test_settings(dir.path(), "retry-events-test"),
+    )
+    .await
+    .unwrap();
+    assert_eq!(outcome.status, StageStatus::Success);
+
+    let collected = events.lock().unwrap();
+    let work_started: Vec<_> = collected
+        .iter()
+        .filter_map(|e| match e {
+            WorkflowRunEvent::StageStarted {
+                node_id, attempt, ..
+            } if node_id == "work" => Some(*attempt),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(work_started, vec![1, 2]);
+}
+
+#[tokio::test]
+async fn run_with_lifecycle_emits_initialize_and_setup_events() {
+    let dir = tempfile::tempdir().unwrap();
+    let events = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
+    let events_clone = Arc::clone(&events);
+    let emitter = EventEmitter::new();
+    emitter.on_event(move |event| {
+        let name = match event {
+            WorkflowRunEvent::SandboxInitialized { .. } => "SandboxInitialized",
+            WorkflowRunEvent::SetupStarted { .. } => "SetupStarted",
+            WorkflowRunEvent::SetupCompleted { .. } => "SetupCompleted",
+            WorkflowRunEvent::WorkflowRunStarted { .. } => "WorkflowRunStarted",
+            _ => return,
+        };
+        events_clone.lock().unwrap().push(name.to_string());
+    });
+
+    let outcome = run_with_lifecycle(
+        make_registry(),
+        Arc::new(emitter),
+        local_env(),
+        &simple_graph(),
+        test_settings(dir.path(), "order-test"),
+        test_lifecycle(vec!["echo ok".to_string()]),
+    )
+    .await
+    .unwrap();
+    assert_eq!(outcome.status, StageStatus::Success);
+
+    let names = events.lock().unwrap();
+    let sandbox_idx = names
+        .iter()
+        .position(|n| n == "SandboxInitialized")
+        .unwrap();
+    let setup_idx = names.iter().position(|n| n == "SetupStarted").unwrap();
+    let run_started_idx = names
+        .iter()
+        .position(|n| n == "WorkflowRunStarted")
+        .unwrap();
+    assert!(sandbox_idx < setup_idx);
+    assert!(setup_idx < run_started_idx);
+}
+
+#[tokio::test]
+async fn git_checkpoint_skips_start_node() {
+    let repo_dir = tempfile::tempdir().unwrap();
+    let repo = repo_dir.path();
+    std::process::Command::new("git")
+        .args(["init"])
+        .current_dir(repo)
+        .output()
+        .unwrap();
+    std::process::Command::new("git")
+        .args([
+            "-c",
+            "user.name=Test",
+            "-c",
+            "user.email=test@test.com",
+            "commit",
+            "--allow-empty",
+            "-m",
+            "initial",
+        ])
+        .current_dir(repo)
+        .output()
+        .unwrap();
+    let base_sha = String::from_utf8(
+        std::process::Command::new("git")
+            .args(["rev-parse", "HEAD"])
+            .current_dir(repo)
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap()
+    .trim()
+    .to_string();
+
+    let run_tmp = tempfile::tempdir().unwrap();
+    let mut g = simple_graph();
+    g.nodes.insert("work".to_string(), Node::new("work"));
+    g.edges.clear();
+    g.edges.push(Edge::new("start", "work"));
+    g.edges.push(Edge::new("work", "exit"));
+
+    let events = Arc::new(std::sync::Mutex::new(Vec::<WorkflowRunEvent>::new()));
+    let events_clone = Arc::clone(&events);
+    let emitter = EventEmitter::new();
+    emitter.on_event(move |event| {
+        events_clone.lock().unwrap().push(event.clone());
+    });
+
+    let sandbox: Arc<dyn Sandbox> = Arc::new(fabro_agent::LocalSandbox::new(repo.to_path_buf()));
+    let mut settings = test_settings(run_tmp.path(), "git-cp-test");
+    settings.git = Some(GitCheckpointSettings {
+        base_sha: Some(base_sha),
+        run_branch: None,
+        meta_branch: Some(crate::git::MetadataStore::branch_name("git-cp-test")),
+    });
+    settings.host_repo_path = Some(repo.to_path_buf());
+
+    run_graph(make_registry(), Arc::new(emitter), sandbox, &g, &settings)
+        .await
+        .unwrap();
+
+    let collected = events.lock().unwrap();
+    let checkpoint_node_ids: Vec<&str> = collected
+        .iter()
+        .filter_map(|e| match e {
+            WorkflowRunEvent::CheckpointCompleted {
+                node_id,
+                git_commit_sha: Some(_),
+                ..
+            } => Some(node_id.as_str()),
+            _ => None,
+        })
+        .collect();
+    assert!(!checkpoint_node_ids.contains(&"start"));
+    assert!(checkpoint_node_ids.contains(&"work"));
+}

lib/crates/fabro-workflows/src/run_fork.rs

@@ -1,391 +0,0 @@
-use anyhow::{Context, Result};
-use fabro_git_storage::branchstore::BranchStore;
-use fabro_git_storage::gitobj::Store;
-use git2::{Oid, Signature};
-
-use crate::git::MetadataStore;
-use crate::run_record::RunRecord;
-use crate::start_record::StartRecord;
-
-use crate::run_rewind::TimelineEntry;
-
-/// Create a new run that branches from an existing run at a specific checkpoint.
-///
-/// Returns the new run ID.
-pub fn execute_fork(
-    store: &Store,
-    source_run_id: &str,
-    entry: &TimelineEntry,
-    push: bool,
-) -> Result<String> {
-    let new_run_id = ulid::Ulid::new().to_string();
-    let sig = Signature::now("Fabro", "noreply@fabro.sh")?;
-
-    // 1. Create new run branch pointing at the target checkpoint's run commit
-    let new_run_branch = format!("{}{new_run_id}", crate::git::RUN_BRANCH_PREFIX);
-    match &entry.run_commit_sha {
-        Some(sha) => {
-            let oid =
-                Oid::from_str(sha).with_context(|| format!("invalid run commit SHA: {sha}"))?;
-            store
-                .update_ref(&new_run_branch, oid)
-                .map_err(|e| anyhow::anyhow!("failed to create run branch ref: {e}"))?;
-        }
-        None => {
-            anyhow::bail!(
-                "checkpoint @{} has no git_commit_sha; cannot fork",
-                entry.ordinal
-            );
-        }
-    }
-
-    // 2. Create new metadata branch
-    let source_meta_branch = MetadataStore::branch_name(source_run_id);
-    let new_meta_branch = MetadataStore::branch_name(&new_run_id);
-    let source_bs = BranchStore::new(store, &source_meta_branch, &sig);
-    let new_bs = BranchStore::new(store, &new_meta_branch, &sig);
-
-    new_bs
-        .ensure_branch()
-        .map_err(|e| anyhow::anyhow!("failed to create metadata branch: {e}"))?;
-
-    // Read run record, start record, and sandbox from source metadata.
-    let source_entries = source_bs
-        .read_entries(&["run.json", "start.json", "sandbox.json"])
-        .map_err(|e| anyhow::anyhow!("failed to read source metadata: {e}"))?;
-
-    let mut run_record_bytes = None;
-    let mut start_record_bytes = None;
-    let mut sandbox_bytes = None;
-    for (path, data) in source_entries {
-        match path {
-            "run.json" => run_record_bytes = Some(data),
-            "start.json" => start_record_bytes = Some(data),
-            "sandbox.json" => sandbox_bytes = Some(data),
-            _ => {}
-        }
-    }
-    let run_record_bytes =
-        run_record_bytes.ok_or_else(|| anyhow::anyhow!("source run has no run.json"))?;
-
-    let now = chrono::Utc::now();
-
-    // Create new RunRecord for the forked run
-    let mut run_record: RunRecord =
-        serde_json::from_slice(&run_record_bytes).context("failed to parse source run.json")?;
-    run_record.run_id = new_run_id.clone();
-    run_record.created_at = now;
-    let new_run_record_bytes =
-        serde_json::to_vec_pretty(&run_record).context("failed to serialize new run.json")?;
-
-    // Create new StartRecord for the forked run
-    let new_start_record_bytes = if start_record_bytes.is_some() {
-        let start_record = StartRecord {
-            run_id: new_run_id.clone(),
-            start_time: now,
-            run_branch: Some(new_run_branch.clone()),
-            base_sha: None,
-        };
-        Some(
-            serde_json::to_vec_pretty(&start_record)
-                .context("failed to serialize new start.json")?,
-        )
-    } else {
-        None
-    };
-
-    // Read checkpoint from the target metadata commit (not branch tip)
-    let checkpoint_bytes = store
-        .read_blob_at(entry.metadata_commit_oid, "checkpoint.json")
-        .map_err(|e| anyhow::anyhow!("failed to read checkpoint blob: {e}"))?
-        .ok_or_else(|| {
-            anyhow::anyhow!(
-                "no checkpoint.json at metadata commit {}",
-                entry.metadata_commit_oid
-            )
-        })?;
-
-    // Write all entries to the new metadata branch in a single commit
-    let mut file_entries: Vec<(&str, &[u8])> = vec![
-        ("run.json", &new_run_record_bytes),
-        ("checkpoint.json", &checkpoint_bytes),
-    ];
-    if let Some(ref start_record) = new_start_record_bytes {
-        file_entries.push(("start.json", start_record));
-    }
-    if let Some(ref sandbox) = sandbox_bytes {
-        file_entries.push(("sandbox.json", sandbox));
-    }
-
-    let commit_msg = format!("fork from {} @{}", source_run_id, entry.ordinal);
-    new_bs
-        .write_entries(&file_entries, &commit_msg)
-        .map_err(|e| anyhow::anyhow!("failed to write metadata entries: {e}"))?;
-
-    // 3. Optionally push both new branches to origin
-    if push {
-        let repo_path = store
-            .repo()
-            .workdir()
-            .or_else(|| store.repo().path().parent())
-            .unwrap_or(store.repo().path());
-
-        // Check if the source run branch has a remote tracking ref (indicating we use a remote)
-        let source_run_branch = format!("{}{source_run_id}", crate::git::RUN_BRANCH_PREFIX);
-        let remote_ref = format!("refs/remotes/origin/{source_run_branch}");
-        let has_remote_tracking = store.repo().find_reference(&remote_ref).is_ok();
-
-        if has_remote_tracking {
-            eprintln!("Pushing new branches to origin...");
-
-            // Push run branch
-            let run_refspec = format!("refs/heads/{new_run_branch}:refs/heads/{new_run_branch}");
-            crate::git::push_branch(repo_path, "origin", &run_refspec)
-                .map_err(|e| anyhow::anyhow!("failed to push run branch: {e}"))?;
-
-            // Push metadata branch
-            let meta_refspec = format!("refs/heads/{new_meta_branch}:refs/heads/{new_meta_branch}");
-            crate::git::push_branch(repo_path, "origin", &meta_refspec)
-                .map_err(|e| anyhow::anyhow!("failed to push metadata branch: {e}"))?;
-
-            eprintln!("Remote refs updated.");
-        }
-    }
-
-    Ok(new_run_id)
-}
-
-#[cfg(test)]
-mod tests {
-    use std::collections::HashMap;
-
-    use super::*;
-    use crate::run_rewind::{build_timeline, find_run_id_by_prefix, parse_target, resolve_target};
-    use git2::Repository;
-
-    fn temp_repo() -> (tempfile::TempDir, Store) {
-        let dir = tempfile::TempDir::new().unwrap();
-        let repo = Repository::init(dir.path()).unwrap();
-        (dir, Store::new(repo))
-    }
-
-    fn test_sig() -> Signature<'static> {
-        Signature::now("Test", "test@example.com").unwrap()
-    }
-
-    fn make_checkpoint_json(current_node: &str, visit: usize, git_sha: Option<&str>) -> Vec<u8> {
-        let mut node_visits = HashMap::new();
-        node_visits.insert(current_node.to_string(), visit);
-        let cp = serde_json::json!({
-            "timestamp": "2025-01-01T00:00:00Z",
-            "current_node": current_node,
-            "completed_nodes": [current_node],
-            "node_retries": {},
-            "context_values": {},
-            "logs": [],
-            "node_visits": node_visits,
-            "git_commit_sha": git_sha,
-        });
-        serde_json::to_vec(&cp).unwrap()
-    }
-
-    fn make_run_record_json(run_id: &str) -> Vec<u8> {
-        let record = serde_json::json!({
-            "run_id": run_id,
-            "created_at": "2025-01-01T00:00:00Z",
-            "config": {},
-            "graph": {
-                "name": "test_workflow",
-                "nodes": {
-                    "start": {"id": "start", "attrs": {}},
-                    "build": {"id": "build", "attrs": {}},
-                    "test": {"id": "test", "attrs": {}}
-                },
-                "edges": [
-                    {"from": "start", "to": "build", "attrs": {}},
-                    {"from": "build", "to": "test", "attrs": {}}
-                ],
-                "attrs": {}
-            },
-            "working_directory": "/tmp/test",
-        });
-        serde_json::to_vec_pretty(&record).unwrap()
-    }
-
-    fn make_start_record_json(run_id: &str) -> Vec<u8> {
-        let record = serde_json::json!({
-            "run_id": run_id,
-            "start_time": "2025-01-01T00:00:00Z",
-            "run_branch": format!("{}{}",  crate::git::RUN_BRANCH_PREFIX, run_id),
-        });
-        serde_json::to_vec_pretty(&record).unwrap()
-    }
-
-    /// Set up a source run with the given number of checkpoints.
-    /// Returns (run_id, vec of run commit OIDs).
-    fn setup_source_run(store: &Store, run_id: &str, nodes: &[&str]) -> Vec<Oid> {
-        let sig = test_sig();
-
-        // Create run branch with commits
-        let run_branch = format!("{}{run_id}", crate::git::RUN_BRANCH_PREFIX);
-        let empty_tree = store.write_empty_tree().unwrap();
-        let mut run_oids = Vec::new();
-        let mut parent: Option<Oid> = None;
-
-        for node in nodes {
-            let parents = match parent {
-                Some(p) => vec![p],
-                None => vec![],
-            };
-            let oid = store
-                .write_commit(
-                    empty_tree,
-                    &parents,
-                    &format!("fabro({run_id}): {node} (completed)"),
-                    &sig,
-                )
-                .unwrap();
-            store.update_ref(&run_branch, oid).unwrap();
-            run_oids.push(oid);
-            parent = Some(oid);
-        }
-
-        // Create metadata branch
-        let meta_branch = MetadataStore::branch_name(run_id);
-        let bs = BranchStore::new(store, &meta_branch, &sig);
-        bs.ensure_branch().unwrap();
-
-        // Write run record and start record
-        let run_record = make_run_record_json(run_id);
-        let start_record = make_start_record_json(run_id);
-        bs.write_entries(
-            &[("run.json", &run_record), ("start.json", &start_record)],
-            "init run",
-        )
-        .unwrap();
-
-        // Write checkpoint commits
-        for (i, node) in nodes.iter().enumerate() {
-            let cp = make_checkpoint_json(node, 1, Some(&run_oids[i].to_string()));
-            bs.write_entry("checkpoint.json", &cp, "checkpoint")
-                .unwrap();
-        }
-
-        run_oids
-    }
-
-    #[test]
-    fn fork_creates_new_run_branch() {
-        let (_dir, store) = temp_repo();
-        let run_oids = setup_source_run(&store, "source-run", &["start", "build"]);
-
-        let timeline = build_timeline(&store, "source-run").unwrap();
-        // Fork at @1 (start)
-        let entry = &timeline[0];
-
-        let new_run_id = execute_fork(&store, "source-run", entry, false).unwrap();
-
-        // Verify new run branch exists and points at the target run commit
-        let new_run_branch = format!("{}{new_run_id}", crate::git::RUN_BRANCH_PREFIX);
-        let resolved = store.resolve_ref(&new_run_branch).unwrap().unwrap();
-        assert_eq!(resolved, run_oids[0]);
-    }
-
-    #[test]
-    fn fork_creates_new_metadata_branch() {
-        let (_dir, store) = temp_repo();
-        setup_source_run(&store, "source-run", &["start", "build"]);
-
-        let timeline = build_timeline(&store, "source-run").unwrap();
-        let entry = &timeline[0]; // @1
-
-        let new_run_id = execute_fork(&store, "source-run", entry, false).unwrap();
-
-        // Verify new metadata branch exists
-        let new_meta_branch = MetadataStore::branch_name(&new_run_id);
-        let sig = test_sig();
-        let bs = BranchStore::new(&store, &new_meta_branch, &sig);
-
-        // Check RunRecord has new run_id and updated created_at
-        let rr_bytes = bs.read_entry("run.json").unwrap().unwrap();
-        let run_record: RunRecord = serde_json::from_slice(&rr_bytes).unwrap();
-        assert_eq!(run_record.run_id, new_run_id);
-
-        // Check StartRecord has new run_id and updated run_branch
-        let sr_bytes = bs.read_entry("start.json").unwrap().unwrap();
-        let start_record: StartRecord = serde_json::from_slice(&sr_bytes).unwrap();
-        assert_eq!(start_record.run_id, new_run_id);
-        assert_eq!(
-            start_record.run_branch.as_deref(),
-            Some(format!("{}{new_run_id}", crate::git::RUN_BRANCH_PREFIX).as_str())
-        );
-
-        // Check checkpoint matches target (@1 = start)
-        let cp_bytes = bs.read_entry("checkpoint.json").unwrap().unwrap();
-        let cp: serde_json::Value = serde_json::from_slice(&cp_bytes).unwrap();
-        assert_eq!(cp["current_node"], "start");
-    }
-
-    #[test]
-    fn fork_preserves_original_run() {
-        let (_dir, store) = temp_repo();
-        let run_oids = setup_source_run(&store, "source-run", &["start", "build", "test"]);
-
-        // Record original refs
-        let source_run_branch = format!("{}source-run", crate::git::RUN_BRANCH_PREFIX);
-        let source_meta_branch = MetadataStore::branch_name("source-run");
-        let original_run_ref = store.resolve_ref(&source_run_branch).unwrap().unwrap();
-        let original_meta_ref = store.resolve_ref(&source_meta_branch).unwrap().unwrap();
-
-        let timeline = build_timeline(&store, "source-run").unwrap();
-        let entry = &timeline[0]; // @1
-
-        execute_fork(&store, "source-run", entry, false).unwrap();
-
-        // Verify source branches are untouched
-        let after_run_ref = store.resolve_ref(&source_run_branch).unwrap().unwrap();
-        let after_meta_ref = store.resolve_ref(&source_meta_branch).unwrap().unwrap();
-        assert_eq!(original_run_ref, after_run_ref);
-        assert_eq!(original_meta_ref, after_meta_ref);
-
-        // Verify source run branch still points at the last commit (test)
-        assert_eq!(after_run_ref, run_oids[2]);
-    }
-
-    #[test]
-    fn fork_defaults_to_latest_checkpoint() {
-        let (_dir, store) = temp_repo();
-        let run_oids = setup_source_run(&store, "source-run", &["start", "build", "test"]);
-
-        let repo = store.repo();
-        let run_id = find_run_id_by_prefix(repo, "source-run").unwrap();
-        let timeline = build_timeline(&store, &run_id).unwrap();
-
-        // Default: fork from the last checkpoint
-        let entry = timeline.last().unwrap();
-        let new_run_id = execute_fork(&store, &run_id, entry, false).unwrap();
-
-        // Verify new run branch points at the last run commit (test)
-        let new_run_branch = format!("{}{new_run_id}", crate::git::RUN_BRANCH_PREFIX);
-        let resolved = store.resolve_ref(&new_run_branch).unwrap().unwrap();
-        assert_eq!(resolved, run_oids[2]);
-    }
-
-    #[test]
-    fn fork_at_specific_ordinal() {
-        let (_dir, store) = temp_repo();
-        let run_oids = setup_source_run(&store, "source-run", &["start", "build", "test"]);
-
-        let timeline = build_timeline(&store, "source-run").unwrap();
-
-        // Fork at @2 (build)
-        let target = parse_target("@2").unwrap();
-        let entry = resolve_target(&timeline, &target, &HashMap::new()).unwrap();
-        let new_run_id = execute_fork(&store, "source-run", entry, false).unwrap();
-
-        // Verify new run branch points at the second run commit (build)
-        let new_run_branch = format!("{}{new_run_id}", crate::git::RUN_BRANCH_PREFIX);
-        let resolved = store.resolve_ref(&new_run_branch).unwrap().unwrap();
-        assert_eq!(resolved, run_oids[1]);
-    }
-}

lib/crates/fabro-workflows/src/run_rewind.rs

@@ -1,862 +0,0 @@
-use std::collections::HashMap;
-
-use anyhow::{bail, Context, Result};
-use fabro_git_storage::branchstore::{BranchStore, CommitInfo};
-use fabro_git_storage::gitobj::Store;
-use git2::{Oid, Repository, Signature};
-
-use crate::checkpoint::Checkpoint;
-use crate::git::MetadataStore;
-use fabro_graphviz::graph::Graph;
-
-/// Parsed rewind target.
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub enum RewindTarget {
-    /// @N — the Nth checkpoint (1-based)
-    Ordinal(usize),
-    /// node_name — most recent visit of the named node
-    LatestVisit(String),
-    /// node_name@N — the Nth visit of the named node
-    SpecificVisit(String, usize),
-}
-
-/// One row in the checkpoint timeline.
-#[derive(Debug, Clone)]
-pub struct TimelineEntry {
-    /// 1-based ordinal (checkpoint sequence number)
-    pub ordinal: usize,
-    /// The node that was just completed at this checkpoint
-    pub node_name: String,
-    /// Visit number for this node (from node_visits)
-    pub visit: usize,
-    /// OID of the metadata-branch commit that contains this checkpoint
-    pub metadata_commit_oid: Oid,
-    /// SHA of the run-branch commit captured at this checkpoint
-    pub run_commit_sha: Option<String>,
-}
-
-/// Parse a target string into a `RewindTarget`.
-pub fn parse_target(s: &str) -> Result<RewindTarget> {
-    if let Some(rest) = s.strip_prefix('@') {
-        let n: usize = rest
-            .parse()
-            .with_context(|| format!("invalid ordinal: @{rest}"))?;
-        if n == 0 {
-            bail!("ordinal must be >= 1");
-        }
-        return Ok(RewindTarget::Ordinal(n));
-    }
-    if let Some(at_pos) = s.rfind('@') {
-        let name = &s[..at_pos];
-        let visit_str = &s[at_pos + 1..];
-        if !name.is_empty() && !visit_str.is_empty() {
-            if let Ok(visit) = visit_str.parse::<usize>() {
-                if visit == 0 {
-                    bail!("visit number must be >= 1");
-                }
-                return Ok(RewindTarget::SpecificVisit(name.to_string(), visit));
-            }
-        }
-    }
-    Ok(RewindTarget::LatestVisit(s.to_string()))
-}
-
-/// Build the checkpoint timeline by walking the metadata branch oldest-first.
-///
-/// The metadata branch checkpoint.json may not contain `git_commit_sha` (the engine
-/// only writes it to the on-disk checkpoint). As a fallback, we walk the run branch
-/// and match commits by message pattern `fabro({run_id}): {node_name}`.
-pub fn build_timeline(store: &Store, run_id: &str) -> Result<Vec<TimelineEntry>> {
-    let branch = MetadataStore::branch_name(run_id);
-    let sig = Signature::now("Fabro", "noreply@fabro.sh")?;
-    let bs = BranchStore::new(store, &branch, &sig);
-
-    let commits = bs
-        .log(10_000)
-        .map_err(|e| anyhow::anyhow!("failed to read metadata branch log: {e}"))?;
-
-    // Reverse to oldest-first
-    let commits: Vec<&CommitInfo> = commits.iter().rev().collect();
-
-    let mut timeline = Vec::new();
-    let mut ordinal = 0usize;
-
-    for commit in &commits {
-        if !commit.message.starts_with("checkpoint") {
-            continue;
-        }
-        let blob = store
-            .read_blob_at(commit.oid, "checkpoint.json")
-            .map_err(|e| anyhow::anyhow!("failed to read checkpoint blob: {e}"))?;
-        let Some(bytes) = blob else { continue };
-        let cp: Checkpoint = serde_json::from_slice(&bytes)
-            .with_context(|| format!("failed to parse checkpoint at {}", commit.oid))?;
-
-        ordinal += 1;
-        let visit = cp.node_visits.get(&cp.current_node).copied().unwrap_or(1);
-
-        timeline.push(TimelineEntry {
-            ordinal,
-            node_name: cp.current_node.clone(),
-            visit,
-            metadata_commit_oid: commit.oid,
-            run_commit_sha: cp.git_commit_sha.clone(),
-        });
-    }
-
-    // Backfill missing git_commit_sha from run branch commit messages
-    backfill_run_shas(store, run_id, &mut timeline);
-
-    Ok(timeline)
-}
-
-/// Walk the run branch and match commits by message pattern to backfill missing SHAs.
-fn backfill_run_shas(store: &Store, run_id: &str, timeline: &mut [TimelineEntry]) {
-    let needs_backfill = timeline.iter().any(|e| e.run_commit_sha.is_none());
-    if !needs_backfill {
-        return;
-    }
-
-    let run_branch = format!("{}{run_id}", crate::git::RUN_BRANCH_PREFIX);
-    let sig = match Signature::now("Fabro", "noreply@fabro.sh") {
-        Ok(s) => s,
-        Err(_) => return,
-    };
-    let bs = BranchStore::new(store, &run_branch, &sig);
-    let run_commits = match bs.log(10_000) {
-        Ok(c) => c,
-        Err(_) => return,
-    };
-
-    // Build a map from node_name to Vec<commit SHA> (newest-first from log)
-    let prefix = format!("fabro({run_id}): ");
-    let mut node_commits: HashMap<String, Vec<String>> = HashMap::new();
-    for commit in &run_commits {
-        if let Some(rest) = commit.message.strip_prefix(&prefix) {
-            // Message format: "fabro({run_id}): {node_name} ({status})"
-            if let Some(node_name) = rest.split_whitespace().next() {
-                node_commits
-                    .entry(node_name.to_string())
-                    .or_default()
-                    .push(commit.oid.to_string());
-            }
-        }
-    }
-
-    // Assign SHAs to timeline entries that are missing them.
-    // For each node, pop from the end (oldest) to match visit order.
-    for (_, shas) in node_commits.iter_mut() {
-        shas.reverse(); // oldest-first
-    }
-    let mut node_indices: HashMap<String, usize> = HashMap::new();
-
-    for entry in timeline.iter_mut() {
-        if entry.run_commit_sha.is_some() {
-            continue;
-        }
-        if let Some(shas) = node_commits.get(&entry.node_name) {
-            let idx = node_indices.entry(entry.node_name.clone()).or_insert(0);
-            if *idx < shas.len() {
-                entry.run_commit_sha = Some(shas[*idx].clone());
-                *idx += 1;
-            }
-        }
-    }
-}
-
-/// Map interior parallel nodes to their fan-out parallel node ID.
-pub fn detect_parallel_interior(graph: &Graph) -> HashMap<String, String> {
-    let mut interior_map = HashMap::new();
-
-    for node in graph.nodes.values() {
-        if node.handler_type() != Some("parallel") {
-            continue;
-        }
-        let parallel_id = &node.id;
-        // BFS from parallel node to find interior nodes until we hit the fan_in
-        let mut queue: Vec<String> = graph
-            .outgoing_edges(parallel_id)
-            .iter()
-            .map(|e| e.to.clone())
-            .collect();
-        let mut visited = std::collections::HashSet::new();
-
-        while let Some(current) = queue.pop() {
-            if !visited.insert(current.clone()) {
-                continue;
-            }
-            if let Some(n) = graph.nodes.get(&current) {
-                if n.handler_type() == Some("parallel.fan_in") {
-                    continue; // don't traverse past fan_in
-                }
-            }
-            interior_map.insert(current.clone(), parallel_id.clone());
-            for edge in graph.outgoing_edges(&current) {
-                queue.push(edge.to.clone());
-            }
-        }
-    }
-
-    interior_map
-}
-
-/// Resolve a target to a timeline entry, with parallel snap-back.
-pub fn resolve_target<'a>(
-    timeline: &'a [TimelineEntry],
-    target: &RewindTarget,
-    parallel_map: &HashMap<String, String>,
-) -> Result<&'a TimelineEntry> {
-    match target {
-        RewindTarget::Ordinal(n) => timeline
-            .iter()
-            .find(|e| e.ordinal == *n)
-            .ok_or_else(|| anyhow::anyhow!("ordinal @{n} out of range (max @{})", timeline.len())),
-
-        RewindTarget::LatestVisit(name) => {
-            let effective_name = parallel_map.get(name).unwrap_or(name);
-            timeline
-                .iter()
-                .rev()
-                .find(|e| e.node_name == *effective_name)
-                .ok_or_else(|| {
-                    if effective_name != name {
-                        anyhow::anyhow!(
-                            "node '{name}' is inside parallel '{effective_name}'; \
-                             no checkpoint found for '{effective_name}'"
-                        )
-                    } else {
-                        anyhow::anyhow!("no checkpoint found for node '{name}'")
-                    }
-                })
-        }
-
-        RewindTarget::SpecificVisit(name, visit) => {
-            let effective_name = parallel_map.get(name).unwrap_or(name);
-            timeline
-                .iter()
-                .find(|e| e.node_name == *effective_name && e.visit == *visit)
-                .ok_or_else(|| {
-                    if effective_name != name {
-                        anyhow::anyhow!(
-                            "node '{name}' is inside parallel '{effective_name}'; \
-                             no visit {visit} found for '{effective_name}'"
-                        )
-                    } else {
-                        anyhow::anyhow!("no visit {visit} found for node '{name}'")
-                    }
-                })
-        }
-    }
-}
-
-/// Move both refs backward to the target checkpoint.
-pub fn execute_rewind(
-    store: &Store,
-    run_id: &str,
-    entry: &TimelineEntry,
-    push: bool,
-) -> Result<()> {
-    // Move metadata branch ref
-    let meta_branch = MetadataStore::branch_name(run_id);
-    store
-        .update_ref(&meta_branch, entry.metadata_commit_oid)
-        .map_err(|e| anyhow::anyhow!("failed to update metadata ref: {e}"))?;
-    eprintln!(
-        "Rewound metadata branch to @{} ({})",
-        entry.ordinal, entry.node_name
-    );
-
-    // Move run branch ref
-    let run_branch = format!("{}{run_id}", crate::git::RUN_BRANCH_PREFIX);
-    match &entry.run_commit_sha {
-        Some(sha) => {
-            let oid =
-                Oid::from_str(sha).with_context(|| format!("invalid run commit SHA: {sha}"))?;
-            store
-                .update_ref(&run_branch, oid)
-                .map_err(|e| anyhow::anyhow!("failed to update run branch ref: {e}"))?;
-            eprintln!(
-                "Rewound run branch {}{run_id} to {}",
-                crate::git::RUN_BRANCH_PREFIX,
-                &sha[..8]
-            );
-        }
-        None => {
-            eprintln!(
-                "Warning: checkpoint @{} has no git_commit_sha; run branch not moved",
-                entry.ordinal
-            );
-        }
-    }
-
-    // Optionally push to remote
-    if push {
-        let repo_path = store
-            .repo()
-            .workdir()
-            .or_else(|| store.repo().path().parent())
-            .unwrap_or(store.repo().path());
-
-        // Check if run branch has a remote tracking ref
-        let remote_ref = format!("refs/remotes/origin/{run_branch}");
-        let has_remote_tracking = store.repo().find_reference(&remote_ref).is_ok();
-
-        if has_remote_tracking {
-            eprintln!("Force-pushing rewound branches to origin...");
-
-            // Force-push run branch
-            if entry.run_commit_sha.is_some() {
-                let refspec = format!("+refs/heads/{run_branch}:refs/heads/{run_branch}");
-                crate::git::push_branch(repo_path, "origin", &refspec)
-                    .map_err(|e| anyhow::anyhow!("failed to push run branch: {e}"))?;
-            }
-
-            // Force-push metadata branch
-            let meta_refspec = format!("+refs/heads/{meta_branch}:refs/heads/{meta_branch}");
-            crate::git::push_branch(repo_path, "origin", &meta_refspec)
-                .map_err(|e| anyhow::anyhow!("failed to push metadata branch: {e}"))?;
-
-            eprintln!("Remote refs updated.");
-        }
-    }
-
-    Ok(())
-}
-
-/// Find a run ID by exact match or unambiguous prefix.
-pub fn find_run_id_by_prefix(repo: &Repository, prefix: &str) -> Result<String> {
-    let refs = repo.references()?;
-    let pattern = "refs/heads/fabro/meta/";
-    let mut matches = Vec::new();
-
-    for reference in refs.flatten() {
-        let name = match reference.name() {
-            Some(n) => n,
-            None => continue,
-        };
-        if let Some(run_id) = name.strip_prefix(pattern) {
-            if run_id == prefix {
-                return Ok(run_id.to_string());
-            }
-            if run_id.starts_with(prefix) {
-                matches.push(run_id.to_string());
-            }
-        }
-    }
-
-    match matches.len() {
-        0 => bail!("no run found matching '{prefix}'"),
-        1 => Ok(matches.into_iter().next().unwrap()),
-        _ => {
-            let mut msg = format!("ambiguous run ID prefix '{prefix}', matches:\n");
-            for m in &matches {
-                msg.push_str(&format!("  {m}\n"));
-            }
-            bail!("{msg}")
-        }
-    }
-}
-
-/// Load the graph from the metadata branch and build the parallel interior map.
-///
-/// Tries `run.json` (RunRecord with embedded Graph) first, then falls back to
-/// parsing `graph.fabro` DOT source for backward compatibility.
-pub fn load_parallel_map(store: &Store, run_id: &str) -> HashMap<String, String> {
-    let branch = MetadataStore::branch_name(run_id);
-    let sig = match Signature::now("Fabro", "noreply@fabro.sh") {
-        Ok(s) => s,
-        Err(_) => return HashMap::new(),
-    };
-    let bs = BranchStore::new(store, &branch, &sig);
-
-    // Try run.json first (RunRecord with embedded Graph)
-    if let Ok(Some(run_bytes)) = bs.read_entry("run.json") {
-        if let Ok(record) = serde_json::from_slice::<crate::run_record::RunRecord>(&run_bytes) {
-            return detect_parallel_interior(&record.graph);
-        }
-    }
-
-    // Fallback: parse graph.fabro DOT
-    let graph_bytes = match bs.read_entry("graph.fabro") {
-        Ok(Some(bytes)) => bytes,
-        _ => return HashMap::new(),
-    };
-    let dot_source = String::from_utf8_lossy(&graph_bytes);
-    let graph = match fabro_graphviz::parser::parse(&dot_source) {
-        Ok(g) => g,
-        Err(_) => return HashMap::new(),
-    };
-    detect_parallel_interior(&graph)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    // -- parse_target tests --
-
-    #[test]
-    fn parse_target_ordinal() {
-        assert_eq!(parse_target("@4").unwrap(), RewindTarget::Ordinal(4));
-    }
-
-    #[test]
-    fn parse_target_ordinal_one() {
-        assert_eq!(parse_target("@1").unwrap(), RewindTarget::Ordinal(1));
-    }
-
-    #[test]
-    fn parse_target_ordinal_zero_errors() {
-        assert!(parse_target("@0").is_err());
-    }
-
-    #[test]
-    fn parse_target_latest_visit() {
-        assert_eq!(
-            parse_target("step2").unwrap(),
-            RewindTarget::LatestVisit("step2".to_string())
-        );
-    }
-
-    #[test]
-    fn parse_target_specific_visit() {
-        assert_eq!(
-            parse_target("step3@2").unwrap(),
-            RewindTarget::SpecificVisit("step3".to_string(), 2)
-        );
-    }
-
-    #[test]
-    fn parse_target_specific_visit_zero_errors() {
-        assert!(parse_target("step3@0").is_err());
-    }
-
-    // -- build_timeline tests --
-
-    fn temp_repo() -> (tempfile::TempDir, Store) {
-        let dir = tempfile::TempDir::new().unwrap();
-        let repo = Repository::init(dir.path()).unwrap();
-        (dir, Store::new(repo))
-    }
-
-    fn test_sig() -> Signature<'static> {
-        Signature::now("Test", "test@example.com").unwrap()
-    }
-
-    fn make_checkpoint_json(current_node: &str, visit: usize, git_sha: Option<&str>) -> Vec<u8> {
-        let mut node_visits = HashMap::new();
-        node_visits.insert(current_node.to_string(), visit);
-        let cp = serde_json::json!({
-            "timestamp": "2025-01-01T00:00:00Z",
-            "current_node": current_node,
-            "completed_nodes": [current_node],
-            "node_retries": {},
-            "context_values": {},
-            "logs": [],
-            "node_visits": node_visits,
-            "git_commit_sha": git_sha,
-        });
-        serde_json::to_vec(&cp).unwrap()
-    }
-
-    #[test]
-    fn build_timeline_simple() {
-        let (_dir, store) = temp_repo();
-        let sig = test_sig();
-        let branch = MetadataStore::branch_name("test-run-1");
-        let bs = BranchStore::new(&store, &branch, &sig);
-        bs.ensure_branch().unwrap();
-
-        // init commit (should be skipped)
-        bs.write_entry("run.json", b"{}", "init run").unwrap();
-
-        // 3 checkpoint commits
-        let cp1 = make_checkpoint_json("start", 1, Some("aaa"));
-        bs.write_entry("checkpoint.json", &cp1, "checkpoint")
-            .unwrap();
-
-        let cp2 = make_checkpoint_json("build", 1, Some("bbb"));
-        bs.write_entry("checkpoint.json", &cp2, "checkpoint")
-            .unwrap();
-
-        let cp3 = make_checkpoint_json("test", 1, Some("ccc"));
-        bs.write_entry("checkpoint.json", &cp3, "checkpoint")
-            .unwrap();
-
-        let timeline = build_timeline(&store, "test-run-1").unwrap();
-        assert_eq!(timeline.len(), 3);
-        assert_eq!(timeline[0].ordinal, 1);
-        assert_eq!(timeline[0].node_name, "start");
-        assert_eq!(timeline[0].visit, 1);
-        assert_eq!(timeline[1].ordinal, 2);
-        assert_eq!(timeline[1].node_name, "build");
-        assert_eq!(timeline[2].ordinal, 3);
-        assert_eq!(timeline[2].node_name, "test");
-    }
-
-    #[test]
-    fn build_timeline_skips_non_checkpoint_commits() {
-        let (_dir, store) = temp_repo();
-        let sig = test_sig();
-        let branch = MetadataStore::branch_name("test-run-2");
-        let bs = BranchStore::new(&store, &branch, &sig);
-        bs.ensure_branch().unwrap();
-
-        bs.write_entry("run.json", b"{}", "init run").unwrap();
-
-        let cp1 = make_checkpoint_json("start", 1, None);
-        bs.write_entry("checkpoint.json", &cp1, "checkpoint")
-            .unwrap();
-
-        // finalize commit — should be skipped
-        bs.write_entry("retro.json", b"{}", "finalize").unwrap();
-
-        let timeline = build_timeline(&store, "test-run-2").unwrap();
-        assert_eq!(timeline.len(), 1);
-        assert_eq!(timeline[0].node_name, "start");
-    }
-
-    // -- resolve_target tests --
-
-    fn make_timeline() -> Vec<TimelineEntry> {
-        vec![
-            TimelineEntry {
-                ordinal: 1,
-                node_name: "start".to_string(),
-                visit: 1,
-                metadata_commit_oid: Oid::zero(),
-                run_commit_sha: Some("aaa".to_string()),
-            },
-            TimelineEntry {
-                ordinal: 2,
-                node_name: "build".to_string(),
-                visit: 1,
-                metadata_commit_oid: Oid::zero(),
-                run_commit_sha: Some("bbb".to_string()),
-            },
-            TimelineEntry {
-                ordinal: 3,
-                node_name: "build".to_string(),
-                visit: 2,
-                metadata_commit_oid: Oid::zero(),
-                run_commit_sha: Some("ccc".to_string()),
-            },
-        ]
-    }
-
-    #[test]
-    fn resolve_ordinal() {
-        let timeline = make_timeline();
-        let entry = resolve_target(&timeline, &RewindTarget::Ordinal(2), &HashMap::new()).unwrap();
-        assert_eq!(entry.ordinal, 2);
-        assert_eq!(entry.node_name, "build");
-    }
-
-    #[test]
-    fn resolve_latest_visit() {
-        let timeline = make_timeline();
-        let entry = resolve_target(
-            &timeline,
-            &RewindTarget::LatestVisit("build".to_string()),
-            &HashMap::new(),
-        )
-        .unwrap();
-        assert_eq!(entry.ordinal, 3);
-        assert_eq!(entry.visit, 2);
-    }
-
-    #[test]
-    fn resolve_specific_visit() {
-        let timeline = make_timeline();
-        let entry = resolve_target(
-            &timeline,
-            &RewindTarget::SpecificVisit("build".to_string(), 1),
-            &HashMap::new(),
-        )
-        .unwrap();
-        assert_eq!(entry.ordinal, 2);
-        assert_eq!(entry.visit, 1);
-    }
-
-    #[test]
-    fn resolve_ordinal_out_of_range() {
-        let timeline = make_timeline();
-        let result = resolve_target(&timeline, &RewindTarget::Ordinal(99), &HashMap::new());
-        assert!(result.is_err());
-        assert!(result.unwrap_err().to_string().contains("out of range"));
-    }
-
-    #[test]
-    fn resolve_unknown_node() {
-        let timeline = make_timeline();
-        let result = resolve_target(
-            &timeline,
-            &RewindTarget::LatestVisit("nonexistent".to_string()),
-            &HashMap::new(),
-        );
-        assert!(result.is_err());
-        assert!(result
-            .unwrap_err()
-            .to_string()
-            .contains("no checkpoint found"));
-    }
-
-    // -- detect_parallel_interior tests --
-
-    #[test]
-    fn parallel_interior_detection() {
-        let mut graph = Graph::new("test");
-        let mut parallel_node = fabro_graphviz::graph::Node::new("parallel1");
-        parallel_node.attrs.insert(
-            "shape".to_string(),
-            fabro_graphviz::graph::AttrValue::String("component".to_string()),
-        );
-        graph.nodes.insert("parallel1".to_string(), parallel_node);
-
-        let mut fan_in = fabro_graphviz::graph::Node::new("fan_in1");
-        fan_in.attrs.insert(
-            "shape".to_string(),
-            fabro_graphviz::graph::AttrValue::String("tripleoctagon".to_string()),
-        );
-        graph.nodes.insert("fan_in1".to_string(), fan_in);
-
-        let mut a = fabro_graphviz::graph::Node::new("a");
-        a.attrs.insert(
-            "shape".to_string(),
-            fabro_graphviz::graph::AttrValue::String("box".to_string()),
-        );
-        graph.nodes.insert("a".to_string(), a);
-
-        let mut b = fabro_graphviz::graph::Node::new("b");
-        b.attrs.insert(
-            "shape".to_string(),
-            fabro_graphviz::graph::AttrValue::String("box".to_string()),
-        );
-        graph.nodes.insert("b".to_string(), b);
-
-        graph.edges.push(fabro_graphviz::graph::Edge {
-            from: "parallel1".to_string(),
-            to: "a".to_string(),
-            attrs: HashMap::new(),
-        });
-        graph.edges.push(fabro_graphviz::graph::Edge {
-            from: "parallel1".to_string(),
-            to: "b".to_string(),
-            attrs: HashMap::new(),
-        });
-        graph.edges.push(fabro_graphviz::graph::Edge {
-            from: "a".to_string(),
-            to: "fan_in1".to_string(),
-            attrs: HashMap::new(),
-        });
-        graph.edges.push(fabro_graphviz::graph::Edge {
-            from: "b".to_string(),
-            to: "fan_in1".to_string(),
-            attrs: HashMap::new(),
-        });
-
-        let map = detect_parallel_interior(&graph);
-        assert_eq!(map.get("a"), Some(&"parallel1".to_string()));
-        assert_eq!(map.get("b"), Some(&"parallel1".to_string()));
-        assert!(!map.contains_key("parallel1"));
-        assert!(!map.contains_key("fan_in1"));
-    }
-
-    #[test]
-    fn parallel_snap_back() {
-        let timeline = vec![
-            TimelineEntry {
-                ordinal: 1,
-                node_name: "parallel1".to_string(),
-                visit: 1,
-                metadata_commit_oid: Oid::zero(),
-                run_commit_sha: Some("aaa".to_string()),
-            },
-            TimelineEntry {
-                ordinal: 2,
-                node_name: "a".to_string(),
-                visit: 1,
-                metadata_commit_oid: Oid::zero(),
-                run_commit_sha: Some("bbb".to_string()),
-            },
-        ];
-
-        let mut parallel_map = HashMap::new();
-        parallel_map.insert("a".to_string(), "parallel1".to_string());
-
-        // Targeting "a" should snap back to "parallel1"
-        let entry = resolve_target(
-            &timeline,
-            &RewindTarget::LatestVisit("a".to_string()),
-            &parallel_map,
-        )
-        .unwrap();
-        assert_eq!(entry.node_name, "parallel1");
-        assert_eq!(entry.ordinal, 1);
-    }
-
-    // -- execute_rewind tests --
-
-    #[test]
-    fn execute_rewind_moves_metadata_ref() {
-        let (_dir, store) = temp_repo();
-        let sig = test_sig();
-        let branch = MetadataStore::branch_name("run-1");
-        let bs = BranchStore::new(&store, &branch, &sig);
-        bs.ensure_branch().unwrap();
-
-        bs.write_entry("run.json", b"{}", "init run").unwrap();
-
-        let cp1 = make_checkpoint_json("start", 1, None);
-        let oid1 = bs
-            .write_entry("checkpoint.json", &cp1, "checkpoint")
-            .unwrap();
-
-        let cp2 = make_checkpoint_json("build", 1, None);
-        bs.write_entry("checkpoint.json", &cp2, "checkpoint")
-            .unwrap();
-
-        let cp3 = make_checkpoint_json("test", 1, None);
-        bs.write_entry("checkpoint.json", &cp3, "checkpoint")
-            .unwrap();
-
-        let timeline = build_timeline(&store, "run-1").unwrap();
-        let entry = &timeline[0]; // @1 = start
-
-        execute_rewind(&store, "run-1", entry, false).unwrap();
-
-        // Verify metadata ref points to the @1 commit
-        let resolved = store.resolve_ref(&branch).unwrap().unwrap();
-        assert_eq!(resolved, oid1);
-    }
-
-    #[test]
-    fn execute_rewind_moves_run_branch_ref() {
-        let (_dir, store) = temp_repo();
-        let sig = test_sig();
-
-        // Create a run branch with some commits
-        let run_branch = "fabro/run/run-2";
-        let empty_tree = store.write_empty_tree().unwrap();
-        let run_c1 = store
-            .write_commit(empty_tree, &[], "run commit 1", &sig)
-            .unwrap();
-        store.update_ref(run_branch, run_c1).unwrap();
-        let run_c2 = store
-            .write_commit(empty_tree, &[run_c1], "run commit 2", &sig)
-            .unwrap();
-        store.update_ref(run_branch, run_c2).unwrap();
-
-        // Create metadata branch with checkpoints pointing to run commits
-        let meta_branch = MetadataStore::branch_name("run-2");
-        let meta_bs = BranchStore::new(&store, &meta_branch, &sig);
-        meta_bs.ensure_branch().unwrap();
-        meta_bs.write_entry("run.json", b"{}", "init run").unwrap();
-
-        let cp1 = make_checkpoint_json("start", 1, Some(&run_c1.to_string()));
-        meta_bs
-            .write_entry("checkpoint.json", &cp1, "checkpoint")
-            .unwrap();
-
-        let cp2 = make_checkpoint_json("build", 1, Some(&run_c2.to_string()));
-        meta_bs
-            .write_entry("checkpoint.json", &cp2, "checkpoint")
-            .unwrap();
-
-        let timeline = build_timeline(&store, "run-2").unwrap();
-        let entry = &timeline[0]; // @1
-
-        execute_rewind(&store, "run-2", entry, false).unwrap();
-
-        // Verify run branch ref moved to run_c1
-        let resolved = store.resolve_ref(run_branch).unwrap().unwrap();
-        assert_eq!(resolved, run_c1);
-    }
-
-    #[test]
-    fn execute_rewind_warns_on_missing_run_sha() {
-        let (_dir, store) = temp_repo();
-        let sig = test_sig();
-        let branch = MetadataStore::branch_name("run-3");
-        let bs = BranchStore::new(&store, &branch, &sig);
-        bs.ensure_branch().unwrap();
-
-        bs.write_entry("run.json", b"{}", "init run").unwrap();
-
-        let cp1 = make_checkpoint_json("start", 1, None);
-        let oid1 = bs
-            .write_entry("checkpoint.json", &cp1, "checkpoint")
-            .unwrap();
-
-        let timeline = build_timeline(&store, "run-3").unwrap();
-
-        // Should not panic even though run_commit_sha is None
-        execute_rewind(&store, "run-3", &timeline[0], false).unwrap();
-
-        // Metadata ref should still be moved
-        let resolved = store.resolve_ref(&branch).unwrap().unwrap();
-        assert_eq!(resolved, oid1);
-    }
-
-    // -- find_run_id_by_prefix tests --
-
-    #[test]
-    fn find_run_id_exact_match() {
-        let (_dir, store) = temp_repo();
-        let sig = test_sig();
-        let branch = MetadataStore::branch_name("abc-123");
-        let bs = BranchStore::new(&store, &branch, &sig);
-        bs.ensure_branch().unwrap();
-
-        let result = find_run_id_by_prefix(store.repo(), "abc-123").unwrap();
-        assert_eq!(result, "abc-123");
-    }
-
-    #[test]
-    fn find_run_id_prefix_match() {
-        let (_dir, store) = temp_repo();
-        let sig = test_sig();
-        let branch = MetadataStore::branch_name("abc-123-long-id");
-        let bs = BranchStore::new(&store, &branch, &sig);
-        bs.ensure_branch().unwrap();
-
-        let result = find_run_id_by_prefix(store.repo(), "abc-123").unwrap();
-        assert_eq!(result, "abc-123-long-id");
-    }
-
-    #[test]
-    fn find_run_id_ambiguous() {
-        let (_dir, store) = temp_repo();
-        let sig = test_sig();
-
-        let b1 = MetadataStore::branch_name("abc-111");
-        BranchStore::new(&store, &b1, &sig).ensure_branch().unwrap();
-
-        let b2 = MetadataStore::branch_name("abc-222");
-        BranchStore::new(&store, &b2, &sig).ensure_branch().unwrap();
-
-        let result = find_run_id_by_prefix(store.repo(), "abc");
-        assert!(result.is_err());
-        assert!(result.unwrap_err().to_string().contains("ambiguous"));
-    }
-
-    #[test]
-    fn find_run_id_not_found() {
-        let (_dir, store) = temp_repo();
-        let result = find_run_id_by_prefix(store.repo(), "nonexistent");
-        assert!(result.is_err());
-        assert!(result.unwrap_err().to_string().contains("no run found"));
-    }
-
-    #[test]
-    fn rewind_push_refspec_uses_same_name_on_both_sides() {
-        // The meta branch name should work directly as a refspec
-        // without needing strip_prefix translation
-        let meta_branch = MetadataStore::branch_name("run-1");
-        let refspec = format!("+refs/heads/{meta_branch}:refs/heads/{meta_branch}");
-        assert_eq!(
-            refspec,
-            "+refs/heads/fabro/meta/run-1:refs/heads/fabro/meta/run-1"
-        );
-    }
-}

lib/crates/fabro-workflows/src/test_support.rs

@@ -4,13 +4,44 @@ use std::sync::Arc;
 use fabro_agent::Sandbox;
 
 use crate::checkpoint::Checkpoint;
-use crate::engine::WorkflowRunEngine;
 use crate::error::Result;
 use crate::event::EventEmitter;
 use crate::handler::HandlerRegistry;
 use crate::outcome::Outcome;
+use crate::pipeline;
+use crate::pipeline::types::Initialized;
 use crate::run_settings::RunSettings;
 
+struct InitializedOptions {
+    hook_runner: Option<Arc<fabro_hooks::HookRunner>>,
+    env: HashMap<String, String>,
+    checkpoint: Option<Checkpoint>,
+}
+
+fn initialized(
+    registry: HandlerRegistry,
+    emitter: Arc<EventEmitter>,
+    sandbox: Arc<dyn Sandbox>,
+    graph: &fabro_graphviz::graph::Graph,
+    settings: &RunSettings,
+    options: InitializedOptions,
+) -> Initialized {
+    std::fs::create_dir_all(&settings.run_dir).expect("failed to create run dir");
+    Initialized {
+        graph: graph.clone(),
+        source: String::new(),
+        settings: settings.clone(),
+        checkpoint: options.checkpoint,
+        seed_context: None,
+        emitter,
+        sandbox,
+        registry: Arc::new(registry),
+        hook_runner: options.hook_runner,
+        env: options.env,
+        dry_run: settings.dry_run,
+    }
+}
+
 pub async fn run_graph(
     registry: HandlerRegistry,
     emitter: Arc<EventEmitter>,
@@ -18,8 +49,20 @@ pub async fn run_graph(
     graph: &fabro_graphviz::graph::Graph,
     settings: &RunSettings,
 ) -> Result<Outcome> {
-    let engine = WorkflowRunEngine::new(registry, emitter, sandbox);
-    engine.run(graph, settings).await
+    let executed = pipeline::execute(initialized(
+        registry,
+        emitter,
+        sandbox,
+        graph,
+        settings,
+        InitializedOptions {
+            hook_runner: None,
+            env: HashMap::new(),
+            checkpoint: None,
+        },
+    ))
+    .await;
+    executed.outcome
 }
 
 pub async fn run_graph_with_hooks(
@@ -31,12 +74,20 @@ pub async fn run_graph_with_hooks(
     hook_runner: Arc<fabro_hooks::HookRunner>,
     env: Option<HashMap<String, String>>,
 ) -> Result<Outcome> {
-    let mut engine = WorkflowRunEngine::new(registry, emitter, sandbox);
-    engine.set_hook_runner(hook_runner);
-    if let Some(env) = env {
-        engine.set_env(env);
-    }
-    engine.run(graph, settings).await
+    let executed = pipeline::execute(initialized(
+        registry,
+        emitter,
+        sandbox,
+        graph,
+        settings,
+        InitializedOptions {
+            hook_runner: Some(hook_runner),
+            env: env.unwrap_or_default(),
+            checkpoint: None,
+        },
+    ))
+    .await;
+    executed.outcome
 }
 
 pub async fn run_graph_from_checkpoint(
@@ -47,10 +98,20 @@ pub async fn run_graph_from_checkpoint(
     settings: &RunSettings,
     checkpoint: &Checkpoint,
 ) -> Result<Outcome> {
-    let engine = WorkflowRunEngine::new(registry, emitter, sandbox);
-    engine
-        .run_from_checkpoint(graph, settings, checkpoint)
-        .await
+    let executed = pipeline::execute(initialized(
+        registry,
+        emitter,
+        sandbox,
+        graph,
+        settings,
+        InitializedOptions {
+            hook_runner: None,
+            env: HashMap::new(),
+            checkpoint: Some(checkpoint.clone()),
+        },
+    ))
+    .await;
+    executed.outcome
 }
 
 pub struct WorkflowRunner {

lib/crates/fabro-workflows/src/workflow.rs

@@ -1,207 +0,0 @@
-use std::path::Path;
-
-use crate::error::FabroError;
-use crate::pipeline;
-use crate::pipeline::types::TransformOptions;
-use crate::transform::Transform;
-use fabro_graphviz::graph::Graph;
-use fabro_validate::Diagnostic;
-
-/// Builder for configuring and executing a workflow preparation.
-/// Collects custom transforms that run after the built-in ones.
-pub struct WorkflowBuilder {
-    transforms: Vec<Box<dyn Transform>>,
-}
-
-impl WorkflowBuilder {
-    #[must_use]
-    pub fn new() -> Self {
-        Self {
-            transforms: Vec::new(),
-        }
-    }
-
-    /// Register a custom transform. Custom transforms run after built-in transforms,
-    /// in registration order.
-    pub fn register_transform(&mut self, transform: Box<dyn Transform>) {
-        self.transforms.push(transform);
-    }
-
-    /// Prepare a workflow: parse DOT, apply built-in and custom transforms, validate.
-    ///
-    /// # Errors
-    ///
-    /// Returns an error if parsing or validation fails.
-    pub fn prepare(&self, dot_source: &str) -> Result<(Graph, Vec<Diagnostic>), FabroError> {
-        self.prepare_inner(dot_source, None)
-    }
-
-    /// Prepare a workflow with file inlining: parse DOT, apply built-in transforms
-    /// including `FileInliningTransform`, then custom transforms, then validate.
-    ///
-    /// # Errors
-    ///
-    /// Returns an error if parsing or validation fails.
-    pub fn prepare_with_file_inlining(
-        &self,
-        dot_source: &str,
-        base_dir: &Path,
-    ) -> Result<(Graph, Vec<Diagnostic>), FabroError> {
-        self.prepare_inner(dot_source, Some(base_dir))
-    }
-
-    fn prepare_inner(
-        &self,
-        dot_source: &str,
-        base_dir: Option<&Path>,
-    ) -> Result<(Graph, Vec<Diagnostic>), FabroError> {
-        let parsed = pipeline::parse(dot_source)?;
-        let mut transformed = pipeline::transform(
-            parsed,
-            &TransformOptions {
-                base_dir: base_dir.map(Path::to_path_buf),
-                custom_transforms: vec![],
-            },
-        );
-
-        // Apply WorkflowBuilder's own custom transforms
-        for t in &self.transforms {
-            t.apply(&mut transformed.graph);
-        }
-
-        let validated = pipeline::validate(transformed, &[]);
-        let (graph, _source, diagnostics) = validated.into_parts();
-        Ok((graph, diagnostics))
-    }
-}
-
-impl Default for WorkflowBuilder {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-/// Convenience: read a DOT file, apply built-in transforms including file inlining, validate.
-///
-/// # Errors
-///
-/// Returns an error if the file cannot be read, parsed, or validated.
-pub fn prepare_from_file(path: &Path) -> Result<(Graph, Vec<Diagnostic>), FabroError> {
-    let source = std::fs::read_to_string(path)
-        .map_err(|e| FabroError::Parse(format!("Failed to read {}: {e}", path.display())))?;
-    let dot_dir = path.parent().unwrap_or(Path::new("."));
-    WorkflowBuilder::new().prepare_with_file_inlining(&source, dot_dir)
-}
-
-/// Convenience: parse DOT source (no file inlining), apply built-in transforms, validate.
-/// Returns the graph or an error if validation produces Error-severity diagnostics.
-///
-/// # Errors
-///
-/// Returns an error if parsing fails or if validation produces Error-severity diagnostics.
-pub fn prepare_from_source(dot_source: &str) -> Result<Graph, FabroError> {
-    let builder = WorkflowBuilder::new();
-    let (graph, diagnostics) = builder.prepare(dot_source)?;
-    fabro_validate::raise_on_errors(&diagnostics)?;
-    Ok(graph)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use fabro_graphviz::graph::AttrValue;
-
-    const MINIMAL_DOT: &str = r#"digraph Test {
-        graph [goal="Build feature"]
-        start [shape=Mdiamond]
-        exit  [shape=Msquare]
-        start -> exit
-    }"#;
-
-    #[test]
-    fn prepare_from_source_minimal() {
-        let graph = prepare_from_source(MINIMAL_DOT).unwrap();
-        assert_eq!(graph.name, "Test");
-        assert!(graph.find_start_node().is_some());
-        assert!(graph.find_exit_node().is_some());
-    }
-
-    #[test]
-    fn prepare_from_source_applies_variable_expansion() {
-        let dot = r#"digraph Test {
-            graph [goal="Fix bugs"]
-            start [shape=Mdiamond]
-            work  [prompt="Goal: $goal"]
-            exit  [shape=Msquare]
-            start -> work -> exit
-        }"#;
-        let graph = prepare_from_source(dot).unwrap();
-        let prompt = graph.nodes["work"]
-            .attrs
-            .get("prompt")
-            .and_then(AttrValue::as_str)
-            .unwrap();
-        assert_eq!(prompt, "Goal: Fix bugs");
-    }
-
-    #[test]
-    fn prepare_from_source_applies_stylesheet() {
-        let dot = r#"digraph Test {
-            graph [goal="Test", model_stylesheet="* { model: sonnet; }"]
-            start [shape=Mdiamond]
-            work  [label="Work"]
-            exit  [shape=Msquare]
-            start -> work -> exit
-        }"#;
-        let graph = prepare_from_source(dot).unwrap();
-        // "sonnet" alias is resolved to canonical ID "claude-sonnet-4-6"
-        assert_eq!(
-            graph.nodes["work"].attrs.get("model"),
-            Some(&AttrValue::String("claude-sonnet-4-6".into()))
-        );
-    }
-
-    #[test]
-    fn prepare_from_source_returns_error_on_invalid_dot() {
-        let result = prepare_from_source("not a graph");
-        assert!(result.is_err());
-    }
-
-    #[test]
-    fn prepare_from_source_returns_error_on_validation_failure() {
-        let dot = r#"digraph Test {
-            graph [goal="Test"]
-            work [label="Work"]
-        }"#;
-        let result = prepare_from_source(dot);
-        assert!(result.is_err());
-    }
-
-    #[test]
-    fn pipeline_builder_custom_transform() {
-        struct TagTransform;
-        impl Transform for TagTransform {
-            fn apply(&self, graph: &mut fabro_graphviz::graph::Graph) {
-                for node in graph.nodes.values_mut() {
-                    node.attrs
-                        .insert("tagged".to_string(), AttrValue::Boolean(true));
-                }
-            }
-        }
-
-        let mut builder = WorkflowBuilder::new();
-        builder.register_transform(Box::new(TagTransform));
-        let (graph, _) = builder.prepare(MINIMAL_DOT).unwrap();
-        assert_eq!(
-            graph.nodes["start"].attrs.get("tagged"),
-            Some(&AttrValue::Boolean(true))
-        );
-    }
-
-    #[test]
-    fn pipeline_builder_default() {
-        let builder = WorkflowBuilder::default();
-        let (graph, _) = builder.prepare(MINIMAL_DOT).unwrap();
-        assert_eq!(graph.name, "Test");
-    }
-}