Add explicit 7-phase pipeline module with typestate lifecycle

Introduce `fabro_workflows::pipeline` module defining typed phases:
PARSE → TRANSFORM → VALIDATE → INITIALIZE → EXECUTE → RETRO → FINALIZE.

Each phase is a standalone function with `#[non_exhaustive]` input/output
types so the compiler enforces ordering. `Validated` uses private fields
with read-only accessors to guarantee immutability post-validation.

Split `engine.run_with_lifecycle()` into `prepare_sandbox()` +
`execute_graph()` (backward-compatible wrapper preserved). Rewrite
`WorkflowBuilder::prepare_inner()` and CLI `prepare_workflow()` to use
pipeline functions. `PreparedWorkflow` now carries a `Validated` with
accessor methods instead of raw `graph`/`source` fields.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

lib/crates/fabro-cli/src/commands/create.rs

@@ -26,7 +26,16 @@ pub async fn create_run(
 
     let mut prep = prepare_workflow(args, run_defaults, styles, quiet)?;
 
-    let goal = prep.graph.goal();
+    // Collect graph-derived data before moving fields out of prep
+    let goal = prep.graph().goal().to_string();
+    let workflow_name = if prep.graph().name.is_empty() {
+        "unnamed".to_string()
+    } else {
+        prep.graph().name.clone()
+    };
+    let node_count = prep.graph().nodes.len();
+    let edge_count = prep.graph().edges.len();
+    let dot_source = prep.source().to_string();
 
     // Create run directory
     let run_id = args
@@ -40,7 +49,7 @@ pub async fn create_run(
     tokio::fs::create_dir_all(&run_dir).await?;
 
     // Write essential files
-    tokio::fs::write(cached_graph_path(&run_dir), &prep.source).await?;
+    tokio::fs::write(cached_graph_path(&run_dir), &dot_source).await?;
     tokio::fs::write(run_dir.join("id.txt"), &run_id).await?;
     std::fs::File::create(run_dir.join("progress.jsonl"))?;
     fabro_workflows::run_status::write_run_status(
@@ -63,12 +72,12 @@ pub async fn create_run(
     let spec = RunSpec {
         run_id: run_id.clone(),
         workflow_path: std::fs::canonicalize(workflow_path).unwrap_or(workflow_path.clone()),
-        dot_source: prep.source,
+        dot_source,
         working_directory: working_directory.clone(),
         goal: if goal.is_empty() {
             None
         } else {
-            Some(goal.to_string())
+            Some(goal.clone())
         },
         model: prep.model,
         provider: prep.provider,
@@ -82,21 +91,16 @@ pub async fn create_run(
     };
     spec.save(&run_dir)?;
 
-    let workflow_name = if prep.graph.name.is_empty() {
-        "unnamed".to_string()
-    } else {
-        prep.graph.name.clone()
-    };
     let base_branch = fabro_sandbox::daytona::detect_repo_info(&working_directory)
         .ok()
         .and_then(|(_, branch)| branch);
     let manifest = Manifest {
         run_id: run_id.clone(),
         workflow_name,
-        goal: goal.to_string(),
+        goal,
         start_time: Utc::now(),
-        node_count: prep.graph.nodes.len(),
-        edge_count: prep.graph.edges.len(),
+        node_count,
+        edge_count,
         run_branch: None,
         base_sha: None,
         labels,

lib/crates/fabro-cli/src/commands/resume.rs

@@ -202,8 +202,7 @@ async fn prepare_from_checkpoint(
         true,
         false,
     )?;
-    let source = prepared.source;
-    let graph = prepared.graph;
+    let (graph, source, _diagnostics) = prepared.validated.into_parts();
     let run_cfg = prepared.run_cfg;
     let sandbox_provider = prepared.sandbox_provider;
     let workflow_slug = prepared.workflow_slug;
@@ -531,13 +530,16 @@ async fn prepare_from_branch(
                 true,
                 false,
             )?;
-            (
-                prepared.graph,
-                prepared.source,
-                prepared.run_cfg,
-                prepared.sandbox_provider,
-                prepared.workflow_slug,
-            )
+            {
+                let (graph, source, _diagnostics) = prepared.validated.into_parts();
+                (
+                    graph,
+                    source,
+                    prepared.run_cfg,
+                    prepared.sandbox_provider,
+                    prepared.workflow_slug,
+                )
+            }
         } else {
             let (graph, diagnostics) =
                 fabro_workflows::workflow::WorkflowBuilder::new().prepare(&source)?;

lib/crates/fabro-cli/src/commands/run.rs

@@ -15,7 +15,6 @@ use fabro_config::{project as project_config, run as run_config, sandbox as sand
 use fabro_interview::{AutoApproveInterviewer, ConsoleInterviewer, FileInterviewer, Interviewer};
 use fabro_model::{Catalog, FallbackTarget, Provider};
 use fabro_util::terminal::Styles;
-use fabro_validate::Severity;
 use fabro_workflows::backend::{AgentApiBackend, AgentCliBackend, BackendRouter};
 use fabro_workflows::checkpoint::Checkpoint;
 use fabro_workflows::conclusion::Conclusion;
@@ -29,7 +28,6 @@ use fabro_workflows::manifest::Manifest;
 use fabro_workflows::outcome::{Outcome, OutcomeExt, StageStatus};
 use fabro_workflows::run_status::{RunStatus, StatusReason};
 use fabro_workflows::sandbox_provider::SandboxProvider;
-use fabro_workflows::workflow::WorkflowBuilder;
 use indicatif::HumanDuration;
 use std::time::Duration;
 use tracing::debug;
@@ -545,8 +543,7 @@ pub(crate) fn resolve_workflow_source(
 
 /// Result of workflow preparation (shared between `create` and `run` commands).
 pub(crate) struct PreparedWorkflow {
-    pub source: String,
-    pub graph: fabro_graphviz::graph::Graph,
+    pub validated: fabro_workflows::pipeline::Validated,
     pub run_cfg: Option<FabroConfig>,
     pub sandbox_provider: SandboxProvider,
     pub model: String,
@@ -555,6 +552,17 @@ pub(crate) struct PreparedWorkflow {
     pub run_defaults: FabroConfig,
 }
 
+impl PreparedWorkflow {
+    /// Read-through to validated graph.
+    pub fn graph(&self) -> &fabro_graphviz::graph::Graph {
+        self.validated.graph()
+    }
+    /// Read-through to validated source.
+    pub fn source(&self) -> &str {
+        self.validated.source()
+    }
+}
+
 /// Resolve config, parse/validate the workflow graph, and resolve sandbox + model.
 ///
 /// Shared between `create_run` (which only persists the spec) and
@@ -614,7 +622,7 @@ pub(crate) fn prepare_workflow_with_project_config(
             .map_err(|e| anyhow::anyhow!("Failed to set working directory to {dir}: {e}"))?;
     }
 
-    // Parse and validate workflow
+    // Parse and transform workflow using pipeline functions
     let source = read_workflow_file(&dot_path)?;
     let vars = run_cfg
         .as_ref()
@@ -625,34 +633,47 @@ pub(crate) fn prepare_workflow_with_project_config(
         None => source,
     };
     let dot_dir = dot_path.parent().unwrap_or(std::path::Path::new("."));
-    let (mut graph, diagnostics) =
-        WorkflowBuilder::new().prepare_with_file_inlining(&source, dot_dir)?;
+
+    let parsed = fabro_workflows::pipeline::parse(&source)?;
+    let mut transformed = fabro_workflows::pipeline::transform(
+        parsed,
+        &fabro_workflows::pipeline::TransformOptions {
+            base_dir: Some(dot_dir.to_path_buf()),
+            custom_transforms: vec![],
+        },
+    );
+
+    // Apply goal override on the mutable transformed graph
     let cli_goal = resolve_cli_goal(&args.goal, &args.goal_file)?;
     let toml_goal = run_cfg.as_ref().and_then(|c| c.goal.as_deref());
-    apply_goal_override(&mut graph, cli_goal.as_deref(), toml_goal);
+    apply_goal_override(&mut transformed.graph, cli_goal.as_deref(), toml_goal);
 
     // Inline @file references in the (possibly overridden) goal
-    if let Some(fabro_graphviz::graph::AttrValue::String(goal)) = graph.attrs.get("goal") {
+    if let Some(fabro_graphviz::graph::AttrValue::String(goal)) =
+        transformed.graph.attrs.get("goal")
+    {
         let fallback = dirs::home_dir().map(|h| h.join(".fabro"));
         let resolved =
             fabro_workflows::transform::resolve_file_ref(goal, dot_dir, fallback.as_deref());
         if resolved != *goal {
-            graph.attrs.insert(
+            transformed.graph.attrs.insert(
                 "goal".to_string(),
                 fabro_graphviz::graph::AttrValue::String(resolved),
             );
         }
     }
 
+    let validated = fabro_workflows::pipeline::validate(transformed, &[]);
+
     if !quiet {
         eprintln!(
             "{} {} {}",
             styles.bold.apply_to("Workflow:"),
-            graph.name,
+            validated.graph().name,
             styles.dim.apply_to(format!(
                 "({} nodes, {} edges)",
-                graph.nodes.len(),
-                graph.edges.len()
+                validated.graph().nodes.len(),
+                validated.graph().edges.len()
             )),
         );
         eprintln!(
@@ -661,16 +682,16 @@ pub(crate) fn prepare_workflow_with_project_config(
             styles.dim.apply_to(relative_path(&dot_path)),
         );
 
-        let goal = graph.goal();
+        let goal = validated.graph().goal();
         if !goal.is_empty() {
             let stripped = fabro_util::text::strip_goal_decoration(goal);
             eprintln!("{} {stripped}\n", styles.bold.apply_to("Goal:"));
         }
 
-        print_diagnostics(&diagnostics, styles);
+        print_diagnostics(validated.diagnostics(), styles);
     }
 
-    if diagnostics.iter().any(|d| d.severity == Severity::Error) {
+    if validated.has_errors() {
         bail!("Validation failed");
     }
 
@@ -691,12 +712,11 @@ pub(crate) fn prepare_workflow_with_project_config(
         args.provider.as_deref(),
         run_cfg.as_ref(),
         &run_defaults,
-        &graph,
+        validated.graph(),
     );
 
     Ok(PreparedWorkflow {
-        source,
-        graph,
+        validated,
         run_cfg,
         sandbox_provider,
         model,
@@ -719,8 +739,7 @@ pub async fn run_command(
     git_author: fabro_workflows::git::GitAuthor,
 ) -> anyhow::Result<()> {
     let PreparedWorkflow {
-        source,
-        graph,
+        validated,
         mut run_cfg,
         sandbox_provider,
         model,
@@ -728,6 +747,7 @@ pub async fn run_command(
         workflow_slug: prepared_workflow_slug,
         run_defaults,
     } = prepare_workflow(&args, run_defaults, styles, false)?;
+    let (graph, source, _diagnostics) = validated.into_parts();
 
     let workflow_path = args.workflow.as_ref().unwrap(); // safe: prepare_workflow validated
 
@@ -2869,8 +2889,8 @@ include = ["*.md"]
         )
         .unwrap();
 
-        assert_eq!(prepared.graph.name, "smoke");
-        assert_eq!(prepared.graph.goal(), "toml goal");
+        assert_eq!(prepared.graph().name, "smoke");
+        assert_eq!(prepared.graph().goal(), "toml goal");
         assert_eq!(prepared.sandbox_provider, SandboxProvider::Docker);
         assert_eq!(prepared.model, "gpt-5.2");
         assert_eq!(prepared.provider.as_deref(), Some("openai"));

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

@@ -1219,6 +1219,18 @@ impl WorkflowRunEngine {
         lifecycle: LifecycleConfig,
         checkpoint: Option<&Checkpoint>,
     ) -> Result<Outcome> {
+        self.prepare_sandbox(graph, config, lifecycle).await?;
+        self.execute_graph(graph, config, checkpoint).await
+    }
+
+    /// INITIALIZE: sandbox setup, git, setup commands, devcontainer.
+    /// Mutates config (fills base_sha, run_branch from sandbox git setup).
+    pub async fn prepare_sandbox(
+        &self,
+        graph: &Graph,
+        config: &mut RunConfig,
+        lifecycle: LifecycleConfig,
+    ) -> Result<()> {
         // 1. Initialize sandbox
         self.services
             .sandbox
@@ -1338,7 +1350,16 @@ impl WorkflowRunEngine {
             .map_err(|e| FabroError::engine(e.to_string()))?;
         }
 
-        // 7. Execute the workflow graph
+        Ok(())
+    }
+
+    /// EXECUTE: pure graph traversal. No sandbox setup.
+    pub async fn execute_graph(
+        &self,
+        graph: &Graph,
+        config: &RunConfig,
+        checkpoint: Option<&Checkpoint>,
+    ) -> Result<Outcome> {
         if let Some(cp) = checkpoint {
             self.run_from_checkpoint(graph, config, cp).await
         } else {

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

@@ -112,6 +112,7 @@ pub mod graph_render;
 pub mod handler;
 pub mod manifest;
 pub mod outcome;
+pub mod pipeline;
 pub mod preamble;
 pub mod pull_request;
 pub mod run_fork;

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

@@ -0,0 +1,36 @@
+use std::time::Instant;
+
+use super::types::{Executed, Initialized};
+
+/// EXECUTE phase: run the workflow graph.
+///
+/// Infallible at the function level — engine errors are captured in `outcome`.
+pub async fn execute(init: Initialized) -> Executed {
+    let Initialized {
+        graph,
+        source: _,
+        engine,
+        config,
+        checkpoint,
+        emitter,
+        sandbox,
+    } = init;
+
+    let start = Instant::now();
+
+    let outcome = engine
+        .execute_graph(&graph, &config, checkpoint.as_ref())
+        .await;
+
+    let duration_ms = crate::millis_u64(start.elapsed());
+
+    Executed {
+        graph,
+        outcome,
+        config,
+        engine,
+        emitter,
+        sandbox,
+        duration_ms,
+    }
+}

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

@@ -0,0 +1,53 @@
+use crate::error::FabroError;
+
+use super::types::{FinalizeOptions, Finalized, Retroed};
+
+/// FINALIZE phase: classify outcome, build conclusion, persist terminal state.
+///
+/// # Errors
+///
+/// Returns `FabroError` if persisting terminal state fails.
+pub async fn finalize(
+    retroed: Retroed,
+    _options: &FinalizeOptions,
+) -> Result<Finalized, FabroError> {
+    let Retroed {
+        graph: _,
+        outcome,
+        config,
+        engine: _,
+        emitter: _,
+        sandbox: _,
+        duration_ms: _,
+        retro: _,
+    } = retroed;
+
+    // TODO: Extract finalize logic from CLI run.rs in Step 5.
+    // For now, return a minimal Finalized.
+    let conclusion = crate::conclusion::Conclusion {
+        timestamp: chrono::Utc::now(),
+        status: match &outcome {
+            Ok(o) => o.status.clone(),
+            Err(_) => crate::outcome::StageStatus::Fail,
+        },
+        duration_ms: 0,
+        failure_reason: outcome.as_ref().err().map(|e| e.to_string()),
+        final_git_commit_sha: None,
+        stages: vec![],
+        total_cost: None,
+        total_retries: 0,
+        total_input_tokens: 0,
+        total_output_tokens: 0,
+        total_cache_read_tokens: 0,
+        total_cache_write_tokens: 0,
+        total_reasoning_tokens: 0,
+        has_pricing: false,
+    };
+
+    Ok(Finalized {
+        run_id: config.run_id,
+        outcome,
+        conclusion,
+        pr_url: None,
+    })
+}

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

@@ -0,0 +1,64 @@
+use std::sync::Arc;
+
+use fabro_hooks::HookRunner;
+
+use crate::engine::WorkflowRunEngine;
+use crate::error::FabroError;
+
+use super::types::{InitOptions, Initialized, Validated};
+
+/// INITIALIZE phase: set up the engine and prepare the sandbox for execution.
+///
+/// - Creates run directory, writes `graph.fabro`
+/// - Builds `WorkflowRunEngine` from components
+/// - Wires hooks, env, dry_run onto engine
+/// - Calls `engine.prepare_sandbox()` (sandbox init, git setup, setup commands, devcontainer)
+///
+/// # Errors
+///
+/// Returns `FabroError` if sandbox preparation fails.
+pub async fn initialize(
+    validated: Validated,
+    mut options: InitOptions,
+) -> Result<Initialized, FabroError> {
+    let (graph, source, _diagnostics) = validated.into_parts();
+
+    // Create run directory and write graph
+    std::fs::create_dir_all(&options.run_dir)?;
+    let graph_path = options.run_dir.join("graph.fabro");
+    std::fs::write(&graph_path, &source)?;
+
+    // Build engine
+    let mut engine = WorkflowRunEngine::with_interviewer(
+        options.registry,
+        Arc::clone(&options.emitter),
+        options.interviewer,
+        Arc::clone(&options.sandbox),
+    );
+
+    // Wire hooks
+    if !options.hooks.hooks.is_empty() {
+        engine.set_hook_runner(Arc::new(HookRunner::new(options.hooks)));
+    }
+
+    // Wire env and dry_run
+    engine.set_env(options.sandbox_env);
+    engine.set_dry_run(options.dry_run);
+
+    // Prepare sandbox (initialize, git setup, setup commands, devcontainer)
+    engine
+        .prepare_sandbox(&graph, &mut options.run_config, options.lifecycle)
+        .await?;
+
+    // At this point run_config may have been mutated by prepare_sandbox (base_sha, run_branch, etc.)
+
+    Ok(Initialized {
+        graph,
+        source,
+        engine,
+        config: options.run_config,
+        checkpoint: None,
+        emitter: options.emitter,
+        sandbox: options.sandbox,
+    })
+}

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

@@ -0,0 +1,17 @@
+mod execute;
+mod finalize;
+mod initialize;
+mod parse;
+mod retro;
+mod transform;
+pub mod types;
+mod validate;
+
+pub use execute::execute;
+pub use finalize::finalize;
+pub use initialize::initialize;
+pub use parse::parse;
+pub use retro::retro;
+pub use transform::transform;
+pub use types::*;
+pub use validate::validate;

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

@@ -0,0 +1,41 @@
+use super::types::Parsed;
+use crate::error::FabroError;
+
+/// PARSE phase: parse DOT source into a `Parsed` graph.
+///
+/// # Errors
+///
+/// Returns `FabroError::Parse` if the DOT source is invalid.
+pub fn parse(dot_source: &str) -> Result<Parsed, FabroError> {
+    let graph = fabro_graphviz::parser::parse(dot_source)?;
+    Ok(Parsed {
+        graph,
+        source: dot_source.to_string(),
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_minimal_dot() {
+        let dot = r#"digraph Test {
+            graph [goal="Build feature"]
+            start [shape=Mdiamond]
+            exit  [shape=Msquare]
+            start -> exit
+        }"#;
+        let parsed = parse(dot).unwrap();
+        assert_eq!(parsed.graph.name, "Test");
+        assert!(parsed.graph.find_start_node().is_some());
+        assert!(parsed.graph.find_exit_node().is_some());
+        assert_eq!(parsed.source, dot);
+    }
+
+    #[test]
+    fn parse_invalid_dot() {
+        let result = parse("not a graph");
+        assert!(result.is_err());
+    }
+}

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

@@ -0,0 +1,30 @@
+use super::types::{Executed, RetroOptions, Retroed};
+
+/// RETRO phase: generate a retrospective for the workflow run.
+///
+/// Infallible — errors are logged, not propagated. If disabled, passes through
+/// with `retro: None`.
+pub async fn retro(executed: Executed, _options: &RetroOptions) -> Retroed {
+    let Executed {
+        graph,
+        outcome,
+        config,
+        engine,
+        emitter,
+        sandbox,
+        duration_ms,
+    } = executed;
+
+    // TODO: Extract core retro logic from CLI run.rs in Step 5.
+    // For now, pass through with no retro.
+    Retroed {
+        graph,
+        outcome,
+        config,
+        engine,
+        emitter,
+        sandbox,
+        duration_ms,
+        retro: None,
+    }
+}

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

@@ -0,0 +1,87 @@
+use crate::transform::{
+    FileInliningTransform, ModelResolutionTransform, StylesheetApplicationTransform, Transform,
+    VariableExpansionTransform,
+};
+
+use super::types::{Parsed, TransformOptions, Transformed};
+
+/// TRANSFORM phase: apply built-in and custom transforms to a parsed graph.
+///
+/// Infallible. Returns `Transformed` with a mutable `graph` for post-transform
+/// adjustments (e.g. goal override) before validation.
+pub fn transform(parsed: Parsed, options: &TransformOptions) -> Transformed {
+    let Parsed { mut graph, source } = parsed;
+
+    // Built-in transforms (PreambleTransform moved to engine execution time)
+    VariableExpansionTransform.apply(&mut graph);
+    StylesheetApplicationTransform.apply(&mut graph);
+    ModelResolutionTransform.apply(&mut graph);
+
+    // File inlining when base_dir is provided
+    if let Some(ref dir) = options.base_dir {
+        let fallback = dirs::home_dir().map(|h| h.join(".fabro"));
+        FileInliningTransform::new(dir.clone(), fallback).apply(&mut graph);
+    }
+
+    // Custom transforms
+    for t in &options.custom_transforms {
+        t.apply(&mut graph);
+    }
+
+    Transformed { graph, source }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::pipeline::parse::parse;
+    use fabro_graphviz::graph::AttrValue;
+
+    #[test]
+    fn transform_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 parsed = parse(dot).unwrap();
+        let transformed = transform(
+            parsed,
+            &TransformOptions {
+                base_dir: None,
+                custom_transforms: vec![],
+            },
+        );
+        let prompt = transformed.graph.nodes["work"]
+            .attrs
+            .get("prompt")
+            .and_then(AttrValue::as_str)
+            .unwrap();
+        assert_eq!(prompt, "Goal: Fix bugs");
+    }
+
+    #[test]
+    fn transform_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 parsed = parse(dot).unwrap();
+        let transformed = transform(
+            parsed,
+            &TransformOptions {
+                base_dir: None,
+                custom_transforms: vec![],
+            },
+        );
+        assert_eq!(
+            transformed.graph.nodes["work"].attrs.get("model"),
+            Some(&AttrValue::String("claude-sonnet-4-6".into()))
+        );
+    }
+}

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

@@ -0,0 +1,180 @@
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use fabro_agent::Sandbox;
+use fabro_graphviz::graph::Graph;
+use fabro_validate::Diagnostic;
+
+use crate::checkpoint::Checkpoint;
+use crate::conclusion::Conclusion;
+use crate::engine::{LifecycleConfig, RunConfig, WorkflowRunEngine};
+use crate::error::FabroError;
+use crate::event::EventEmitter;
+use crate::handler::HandlerRegistry;
+use crate::outcome::Outcome;
+use fabro_interview::Interviewer;
+use fabro_validate::Severity;
+
+/// Output of the PARSE phase.
+#[non_exhaustive]
+pub struct Parsed {
+    pub graph: Graph,
+    pub source: String,
+}
+
+/// Output of the TRANSFORM phase. Graph is mutable — callers may apply
+/// post-transform adjustments (e.g. goal override) before validation.
+#[non_exhaustive]
+pub struct Transformed {
+    pub graph: Graph,
+    pub source: String,
+}
+
+/// Output of the VALIDATE phase. Always produced (even with errors).
+/// Caller inspects diagnostics and decides whether to proceed.
+/// Graph is read-only — use accessors, not direct field access.
+#[non_exhaustive]
+pub struct Validated {
+    graph: Graph,
+    source: String,
+    diagnostics: Vec<Diagnostic>,
+}
+
+impl Validated {
+    /// Create a new `Validated` from its parts.
+    pub(crate) fn new(graph: Graph, source: String, diagnostics: Vec<Diagnostic>) -> Self {
+        Self {
+            graph,
+            source,
+            diagnostics,
+        }
+    }
+
+    pub fn graph(&self) -> &Graph {
+        &self.graph
+    }
+
+    pub fn source(&self) -> &str {
+        &self.source
+    }
+
+    pub fn diagnostics(&self) -> &[Diagnostic] {
+        &self.diagnostics
+    }
+
+    /// True if any diagnostic has Error severity.
+    #[must_use]
+    pub fn has_errors(&self) -> bool {
+        self.diagnostics
+            .iter()
+            .any(|d| d.severity == Severity::Error)
+    }
+
+    /// Returns `Err(FabroError::Validation)` if any Error-severity diagnostics exist.
+    /// Diagnostics remain accessible via `diagnostics()` for printing before this call.
+    pub fn raise_on_errors(&self) -> Result<(), FabroError> {
+        if self.has_errors() {
+            let message = self
+                .diagnostics
+                .iter()
+                .filter(|d| d.severity == Severity::Error)
+                .map(|d| d.message.as_str())
+                .collect::<Vec<_>>()
+                .join("; ");
+            return Err(FabroError::Validation(message));
+        }
+        Ok(())
+    }
+
+    /// Consume into owned graph, source, and diagnostics (used by initialize).
+    pub fn into_parts(self) -> (Graph, String, Vec<Diagnostic>) {
+        (self.graph, self.source, self.diagnostics)
+    }
+}
+
+/// Options for the INITIALIZE phase.
+pub struct InitOptions {
+    pub run_id: String,
+    pub run_dir: PathBuf,
+    pub dry_run: bool,
+    pub emitter: Arc<EventEmitter>,
+    pub interviewer: Arc<dyn Interviewer>,
+    pub sandbox: Arc<dyn Sandbox>,
+    pub registry: HandlerRegistry,
+    pub lifecycle: LifecycleConfig,
+    pub run_config: RunConfig,
+    pub hooks: fabro_hooks::HookConfig,
+    pub sandbox_env: HashMap<String, String>,
+}
+
+/// Output of the INITIALIZE phase.
+#[non_exhaustive]
+pub struct Initialized {
+    pub graph: Graph,
+    pub source: String,
+    pub engine: WorkflowRunEngine,
+    pub config: RunConfig,
+    pub(crate) checkpoint: Option<Checkpoint>,
+    pub emitter: Arc<EventEmitter>,
+    pub sandbox: Arc<dyn Sandbox>,
+}
+
+/// Output of the EXECUTE phase.
+#[non_exhaustive]
+pub struct Executed {
+    pub graph: Graph,
+    pub outcome: Result<Outcome, FabroError>,
+    pub config: RunConfig,
+    pub engine: WorkflowRunEngine,
+    pub emitter: Arc<EventEmitter>,
+    pub sandbox: Arc<dyn Sandbox>,
+    pub duration_ms: u64,
+}
+
+/// Output of the RETRO phase.
+#[non_exhaustive]
+pub struct Retroed {
+    pub graph: Graph,
+    pub outcome: Result<Outcome, FabroError>,
+    pub config: RunConfig,
+    pub engine: WorkflowRunEngine,
+    pub emitter: Arc<EventEmitter>,
+    pub sandbox: Arc<dyn Sandbox>,
+    pub duration_ms: u64,
+    pub retro: Option<fabro_retro::retro::Retro>,
+}
+
+/// Output of the FINALIZE phase.
+#[non_exhaustive]
+pub struct Finalized {
+    pub run_id: String,
+    pub outcome: Result<Outcome, FabroError>,
+    pub conclusion: Conclusion,
+    pub pr_url: Option<String>,
+}
+
+/// Options for the TRANSFORM phase.
+pub struct TransformOptions {
+    pub base_dir: Option<PathBuf>,
+    pub custom_transforms: Vec<Box<dyn crate::transform::Transform>>,
+}
+
+/// Options for the RETRO phase.
+pub struct RetroOptions {
+    pub enabled: bool,
+    pub dry_run: bool,
+    pub llm_client: Option<fabro_llm::client::Client>,
+    pub provider: fabro_llm::Provider,
+    pub model: String,
+}
+
+/// Options for the FINALIZE phase.
+pub struct FinalizeOptions {
+    pub preserve_sandbox: bool,
+    pub pr_config: Option<fabro_config::run::PullRequestConfig>,
+    pub github_app: Option<fabro_github::GitHubAppCredentials>,
+    pub origin_url: Option<String>,
+    pub model: String,
+    pub last_git_sha: Option<String>,
+}

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

@@ -0,0 +1,88 @@
+use fabro_validate::LintRule;
+
+use super::types::{Transformed, Validated};
+
+/// VALIDATE phase: run lint rules against the transformed graph.
+///
+/// **Infallible.** Always returns `Validated` with diagnostics. Caller decides
+/// whether to fail via `validated.raise_on_errors()`.
+pub fn validate(transformed: Transformed, extra_rules: &[&dyn LintRule]) -> Validated {
+    let Transformed { graph, source } = transformed;
+    let diagnostics = fabro_validate::validate(&graph, extra_rules);
+    Validated::new(graph, source, diagnostics)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::pipeline::parse::parse;
+    use crate::pipeline::transform;
+    use crate::pipeline::types::TransformOptions;
+
+    fn run_pipeline(dot: &str) -> Validated {
+        let parsed = parse(dot).unwrap();
+        let transformed = transform::transform(
+            parsed,
+            &TransformOptions {
+                base_dir: None,
+                custom_transforms: vec![],
+            },
+        );
+        validate(transformed, &[])
+    }
+
+    #[test]
+    fn validate_valid_graph() {
+        let dot = r#"digraph Test {
+            graph [goal="Build feature"]
+            start [shape=Mdiamond]
+            exit  [shape=Msquare]
+            start -> exit
+        }"#;
+        let validated = run_pipeline(dot);
+        assert!(!validated.has_errors());
+        assert!(validated.raise_on_errors().is_ok());
+    }
+
+    #[test]
+    fn validate_missing_start_node() {
+        let dot = r#"digraph Test {
+            graph [goal="Test"]
+            work [label="Work"]
+        }"#;
+        let validated = run_pipeline(dot);
+        assert!(validated.has_errors());
+        assert!(validated.raise_on_errors().is_err());
+    }
+
+    #[test]
+    fn validate_into_parts() {
+        let dot = r#"digraph Test {
+            graph [goal="Build feature"]
+            start [shape=Mdiamond]
+            exit  [shape=Msquare]
+            start -> exit
+        }"#;
+        let validated = run_pipeline(dot);
+        let (graph, source, diagnostics) = validated.into_parts();
+        assert_eq!(graph.name, "Test");
+        assert_eq!(source, dot);
+        assert!(diagnostics
+            .iter()
+            .all(|d| d.severity != fabro_validate::Severity::Error));
+    }
+
+    #[test]
+    fn validate_diagnostics_accessible_before_raise() {
+        let dot = r#"digraph Test {
+            graph [goal="Test"]
+            work [label="Work"]
+        }"#;
+        let validated = run_pipeline(dot);
+        // Can read diagnostics before raising
+        let diags = validated.diagnostics();
+        assert!(!diags.is_empty());
+        // Then raise
+        assert!(validated.raise_on_errors().is_err());
+    }
+}

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

@@ -1,10 +1,9 @@
 use std::path::Path;
 
 use crate::error::FabroError;
-use crate::transform::{
-    FileInliningTransform, ModelResolutionTransform, StylesheetApplicationTransform, Transform,
-    VariableExpansionTransform,
-};
+use crate::pipeline;
+use crate::pipeline::types::TransformOptions;
+use crate::transform::Transform;
 use fabro_graphviz::graph::Graph;
 use fabro_validate::Diagnostic;
 
@@ -56,25 +55,22 @@ impl WorkflowBuilder {
         dot_source: &str,
         base_dir: Option<&Path>,
     ) -> Result<(Graph, Vec<Diagnostic>), FabroError> {
-        let mut graph = fabro_graphviz::parser::parse(dot_source)?;
+        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![],
+            },
+        );
 
-        // Built-in transforms (PreambleTransform moved to engine execution time)
-        VariableExpansionTransform.apply(&mut graph);
-        StylesheetApplicationTransform.apply(&mut graph);
-        ModelResolutionTransform.apply(&mut graph);
-
-        // File inlining when base_dir is provided
-        if let Some(dir) = base_dir {
-            let fallback = dirs::home_dir().map(|h| h.join(".fabro"));
-            FileInliningTransform::new(dir.to_path_buf(), fallback).apply(&mut graph);
+        // Apply WorkflowBuilder's own custom transforms
+        for t in &self.transforms {
+            t.apply(&mut transformed.graph);
         }
 
-        // Custom transforms
-        for transform in &self.transforms {
-            transform.apply(&mut graph);
-        }
-
-        let diagnostics = fabro_validate::validate(&graph, &[]);
+        let validated = pipeline::validate(transformed, &[]);
+        let (graph, _source, diagnostics) = validated.into_parts();
         Ok((graph, diagnostics))
     }
 }