diff --git a/lib/crates/fabro-api/src/server.rs b/lib/crates/fabro-api/src/server.rs
index c91a1d7b7..1d9652cb8 100644
--- a/lib/crates/fabro-api/src/server.rs
+++ b/lib/crates/fabro-api/src/server.rs
@@ -25,7 +25,7 @@ use fabro_workflows::context::Context;
 use fabro_workflows::event::{EventEmitter, WorkflowRunEvent};
 use fabro_workflows::handler::HandlerRegistry;
 use fabro_workflows::operations::{self, CreateOptions};
-use fabro_workflows::pipeline::{self, InitOptions};
+use fabro_workflows::pipeline::{self, InitOptions, PersistOptions, Persisted};
 use fabro_workflows::records::Checkpoint;
 use fabro_workflows::run_settings::LifecycleConfig;
 use fabro_workflows::run_settings::RunSettings;
@@ -67,7 +67,6 @@ impl<T: serde::Serialize> ListResponse<T> {
 /// Snapshot of a managed run.
 struct ManagedRun {
     dot_source: String,
-    graph: fabro_graphviz::graph::Graph,
     status: RunStatus,
     error: Option<String>,
     created_at: chrono::DateTime<chrono::Utc>,
@@ -476,14 +475,13 @@ async fn start_run(
     State(state): State<Arc<AppState>>,
     Json(req): Json<StartRunRequest>,
 ) -> Response {
-    // Parse the DOT source
-    let graph = match operations::create(&req.dot_source, CreateOptions::default()) {
+    // Parse and persist the DOT source.
+    let validated = match operations::create(&req.dot_source, CreateOptions::default()) {
         Ok(validated) => {
             if let Err(e) = validated.raise_on_errors() {
                 return ApiError::bad_request(e.to_string()).into_response();
             }
-            let (graph, _, _) = validated.into_parts();
-            graph
+            validated
         }
         Err(e) => {
             return ApiError::bad_request(e.to_string()).into_response();
@@ -495,13 +493,6 @@ async fn start_run(
 
     let created_at = chrono::Utc::now();
     let run_dir = std::env::temp_dir().join(format!("fabro-{}", uuid::Uuid::new_v4()));
-    if let Err(err) = std::fs::create_dir_all(&run_dir) {
-        return ApiError::new(
-            StatusCode::INTERNAL_SERVER_ERROR,
-            format!("Failed to create run directory: {err}"),
-        )
-        .into_response();
-    }
 
     let run_record = fabro_workflows::records::RunRecord {
         run_id: run_id.clone(),
@@ -515,7 +506,7 @@ async fn start_run(
             }),
             ..Default::default()
         },
-        graph: graph.clone(),
+        graph: validated.graph().clone(),
         workflow_slug: None,
         working_directory: std::env::current_dir()
             .unwrap_or_else(|_| std::path::PathBuf::from(".")),
@@ -523,10 +514,16 @@ async fn start_run(
         base_branch: None,
         labels: std::collections::HashMap::new(),
     };
-    if let Err(err) = run_record.save(&run_dir) {
+    if let Err(err) = fabro_workflows::pipeline::persist(
+        validated,
+        PersistOptions {
+            run_dir: run_dir.clone(),
+            run_record,
+        },
+    ) {
         return ApiError::new(
             StatusCode::INTERNAL_SERVER_ERROR,
-            format!("Failed to persist run record: {err}"),
+            format!("Failed to persist run state: {err}"),
         )
         .into_response();
     }
@@ -537,7 +534,6 @@ async fn start_run(
             run_id.clone(),
             ManagedRun {
                 dot_source: req.dot_source,
-                graph,
                 status: RunStatus::Queued,
                 error: None,
                 created_at,
@@ -570,7 +566,7 @@ async fn start_run(
 /// Execute a single run: transitions queued → starting → running → completed/failed/cancelled.
 async fn execute_run(state: Arc<AppState>, run_id: String) {
     // Transition to Starting and set up cancel infrastructure
-    let (cancel_rx, graph, run_dir) = {
+    let (cancel_rx, run_dir) = {
         let mut runs = state.runs.lock().expect("runs lock poisoned");
         let managed_run = match runs.get_mut(&run_id) {
             Some(r) if r.status == RunStatus::Queued => r,
@@ -590,7 +586,7 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
         managed_run.cancel_token = Some(Arc::clone(&cancel_token));
         managed_run.event_tx = Some(event_tx);
 
-        (cancel_rx, managed_run.graph.clone(), run_dir)
+        (cancel_rx, run_dir)
     };
 
     // Create interviewer, sandbox, engine (this is the "provisioning" phase)
@@ -640,20 +636,21 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
         }
     }
 
-    let run_record = match fabro_workflows::records::RunRecord::load(&run_dir) {
-        Ok(r) => r,
+    let persisted = match Persisted::load(&run_dir) {
+        Ok(persisted) => persisted,
         Err(e) => {
-            tracing::error!(run_id = %run_id, error = %e, "Failed to load RunRecord");
+            tracing::error!(run_id = %run_id, error = %e, "Failed to load persisted run");
             let mut runs = state.runs.lock().expect("runs lock poisoned");
             if let Some(managed_run) = runs.get_mut(&run_id) {
                 managed_run.status = RunStatus::Failed;
-                managed_run.error = Some(format!("Failed to load run record: {e}"));
+                managed_run.error = Some(format!("Failed to load persisted run: {e}"));
                 managed_run.event_tx = None;
             }
             state.scheduler_notify.notify_one();
             return;
         }
     };
+    let run_record = persisted.run_record().clone();
     let config = RunSettings {
         config: run_record.config,
         run_dir: run_dir.clone(),
@@ -662,10 +659,10 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
         run_id: run_id.clone(),
         labels: run_record.labels,
         git_author: state.git_author.clone(),
-        workflow_slug: None,
+        workflow_slug: run_record.workflow_slug,
         github_app: None,
-        base_branch: None,
-        host_repo_path: None,
+        base_branch: run_record.base_branch,
+        host_repo_path: run_record.host_repo_path.map(Into::into),
         git: None,
     };
 
@@ -673,19 +670,15 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
         let emitter = Arc::clone(&emitter);
         let sandbox = Arc::clone(&sandbox);
         let registry = Arc::new(registry);
-        let graph = graph.clone();
-        let run_dir = run_dir.clone();
         let run_id = run_id.clone();
         let config = config.clone();
         let hooks = state.hooks.clone();
         let dry_run = state.dry_run;
         async move {
-            let validated = operations::create_from_graph(graph, String::new());
             let initialized = pipeline::initialize(
-                validated,
+                persisted,
                 InitOptions {
                     run_id,
-                    run_dir,
                     dry_run,
                     emitter,
                     sandbox,
diff --git a/lib/crates/fabro-cli/src/commands/create.rs b/lib/crates/fabro-cli/src/commands/create.rs
index 3bf033a8a..399f03b6b 100644
--- a/lib/crates/fabro-cli/src/commands/create.rs
+++ b/lib/crates/fabro-cli/src/commands/create.rs
@@ -3,6 +3,7 @@ use std::path::PathBuf;
 use chrono::Utc;
 use fabro_config::config::FabroConfig;
 use fabro_sandbox::SandboxProvider;
+use fabro_workflows::pipeline::PersistOptions;
 use fabro_workflows::records::RunRecord;
 
 use super::run::{
@@ -77,7 +78,6 @@ pub async fn create_run(
     let dot_source = prep.source().to_string();
     let graph = prep.graph().clone();
 
-    // Create run directory
     let run_id = args
         .run_id
         .clone()
@@ -86,20 +86,6 @@ pub async fn create_run(
         .run_dir
         .clone()
         .unwrap_or_else(|| default_run_dir(&run_id, args.dry_run));
-    tokio::fs::create_dir_all(&run_dir).await?;
-
-    // Write essential files
-    tokio::fs::write(cached_graph_path(&run_dir), &dot_source).await?;
-    tokio::fs::write(run_dir.join("id.txt"), &run_id).await?;
-    std::fs::File::create(run_dir.join("progress.jsonl"))?;
-    fabro_workflows::run_status::write_run_status(
-        &run_dir,
-        fabro_workflows::run_status::RunStatus::Submitted,
-        None,
-    );
-
-    // Copy the original workflow TOML into the run dir as a debug artifact.
-    write_run_config_snapshot(&run_dir, prep.workflow_toml_path.as_deref()).await?;
 
     // Build normalized config and RunRecord
     let working_directory = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
@@ -134,7 +120,24 @@ pub async fn create_run(
         base_branch,
         labels,
     };
-    record.save(&run_dir)?;
+    fabro_workflows::pipeline::persist(
+        prep.validated,
+        PersistOptions {
+            run_dir: run_dir.clone(),
+            run_record: record,
+        },
+    )?;
+
+    // Write CLI-owned debug and status artifacts after the run has been persisted.
+    tokio::fs::write(cached_graph_path(&run_dir), &dot_source).await?;
+    tokio::fs::write(run_dir.join("id.txt"), &run_id).await?;
+    std::fs::File::create(run_dir.join("progress.jsonl"))?;
+    fabro_workflows::run_status::write_run_status(
+        &run_dir,
+        fabro_workflows::run_status::RunStatus::Submitted,
+        None,
+    );
+    write_run_config_snapshot(&run_dir, prep.workflow_toml_path.as_deref()).await?;
 
     Ok((run_id, run_dir))
 }
diff --git a/lib/crates/fabro-cli/src/commands/resume.rs b/lib/crates/fabro-cli/src/commands/resume.rs
index 0be1768a8..3042d1b50 100644
--- a/lib/crates/fabro-cli/src/commands/resume.rs
+++ b/lib/crates/fabro-cli/src/commands/resume.rs
@@ -18,7 +18,7 @@ use fabro_workflows::operations::{
 };
 use fabro_workflows::outcome::StageStatus;
 use fabro_workflows::pipeline::{
-    build_conclusion, classify_engine_result, persist_terminal_outcome,
+    build_conclusion, classify_engine_result, persist_terminal_outcome, PersistOptions, Persisted,
 };
 use fabro_workflows::records::Checkpoint;
 use fabro_workflows::records::RunRecord;
@@ -97,7 +97,7 @@ pub struct ResumeArgs {
 /// Intermediate state produced by the two resolution paths (checkpoint-file vs. git-branch).
 struct ResumeContext {
     checkpoint: Checkpoint,
-    validated: fabro_workflows::pipeline::Validated,
+    persisted: Persisted,
     run_id: String,
     run_dir: PathBuf,
     run_cfg: Option<FabroConfig>,
@@ -227,48 +227,56 @@ async fn prepare_from_checkpoint(
         .run_dir
         .clone()
         .unwrap_or_else(|| default_run_dir(&run_id, args.dry_run));
+    let labels = args
+        .label
+        .iter()
+        .filter_map(|s| s.split_once('='))
+        .map(|(k, v)| (k.to_string(), v.to_string()))
+        .collect();
+    let working_directory = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
+    let cli_flags = super::create::CliFlags {
+        dry_run: args.dry_run,
+        auto_approve: args.auto_approve,
+        no_retro: args.no_retro,
+        verbose: args.verbose,
+        preserve_sandbox: args.preserve_sandbox,
+    };
+    let normalized = super::create::normalize_config(
+        run_cfg.as_ref(),
+        &prepared_run_defaults,
+        &prepared_model,
+        prepared_provider.as_deref(),
+        sandbox_provider,
+        &graph,
+        cli_flags,
+    );
+    let persisted = fabro_workflows::pipeline::persist(
+        validated,
+        PersistOptions {
+            run_dir: run_dir.clone(),
+            run_record: fabro_workflows::records::RunRecord {
+                run_id: run_id.clone(),
+                created_at: chrono::Utc::now(),
+                config: normalized,
+                graph: graph.clone(),
+                workflow_slug: workflow_slug.clone(),
+                working_directory: working_directory.clone(),
+                host_repo_path: Some(working_directory.to_string_lossy().to_string()),
+                base_branch: None,
+                labels,
+            },
+        },
+    )?;
+    let run_cfg: Option<FabroConfig> = Some(persisted.run_record().config.clone());
+    let settings_config = persisted.run_record().config.clone();
+
     tokio::fs::create_dir_all(&run_dir).await?;
     fabro_util::run_log::activate(&run_dir.join("cli.log"))
         .context("Failed to activate per-run log")?;
     let status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?;
     tokio::fs::write(cached_graph_path(&run_dir), &source).await?;
-    let run_cfg: Option<FabroConfig> = run_cfg;
     write_run_config_snapshot(&run_dir, workflow_toml_path.as_deref()).await?;
 
-    // Write RunRecord for the resumed run
-    let settings_config = {
-        let working_directory = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
-        let cli_flags = super::create::CliFlags {
-            dry_run: args.dry_run,
-            auto_approve: args.auto_approve,
-            no_retro: args.no_retro,
-            verbose: args.verbose,
-            preserve_sandbox: args.preserve_sandbox,
-        };
-        let normalized = super::create::normalize_config(
-            run_cfg.as_ref(),
-            &prepared_run_defaults,
-            &prepared_model,
-            prepared_provider.as_deref(),
-            sandbox_provider,
-            &graph,
-            cli_flags,
-        );
-        let record = fabro_workflows::records::RunRecord {
-            run_id: run_id.clone(),
-            created_at: chrono::Utc::now(),
-            config: normalized.clone(),
-            graph: graph.clone(),
-            workflow_slug: workflow_slug.clone(),
-            working_directory: working_directory.clone(),
-            host_repo_path: Some(working_directory.to_string_lossy().to_string()),
-            base_branch: None,
-            labels: std::collections::HashMap::new(),
-        };
-        let _ = record.save(&run_dir);
-        normalized
-    };
-
     let original_cwd = std::env::current_dir()?;
     let emitter = Arc::new(EventEmitter::new());
 
@@ -482,7 +490,7 @@ async fn prepare_from_checkpoint(
 
     Ok(ResumeContext {
         checkpoint,
-        validated,
+        persisted,
         run_id,
         run_dir,
         run_cfg,
@@ -553,7 +561,7 @@ async fn prepare_from_branch(
         .or_else(|| repo_info.as_ref().and_then(|(_, branch)| branch.clone()));
     let base_sha = start_record.as_ref().and_then(|s| s.base_sha.clone());
 
-    let (validated, graph_source, run_cfg, mut sandbox_provider, workflow_slug) =
+    let (mut validated, mut graph_source, mut run_cfg, mut sandbox_provider, mut workflow_slug) =
         if let Some(ref workflow_path) = args.workflow {
             let prepared = prepare_workflow_with_project_config(
                 &resume_as_run_args(args, workflow_path.clone()),
@@ -596,15 +604,6 @@ async fn prepare_from_branch(
         } else {
             bail!("no run.json found on metadata branch for run {run_id}");
         };
-    let graph = validated.graph().clone();
-
-    eprintln!(
-        "{} {} from branch {} ({})",
-        styles.bold.apply_to("Resuming workflow:"),
-        graph.name,
-        styles.dim.apply_to(&run_branch),
-        run_id,
-    );
 
     // Set up logs directory — reuse existing run dir for this run_id to avoid
     // "ambiguous prefix" errors when the resume happens on a different day.
@@ -618,56 +617,94 @@ async fn prepare_from_branch(
     };
     tokio::fs::create_dir_all(&run_dir).await?;
     let run_dir = tokio::fs::canonicalize(&run_dir).await.unwrap_or(run_dir);
+    if args.workflow.is_none() {
+        if let Ok(loaded) = Persisted::load(&run_dir) {
+            graph_source = loaded.source().to_string();
+            run_cfg = Some(loaded.run_record().config.clone());
+            workflow_slug = loaded.run_record().workflow_slug.clone();
+            validated = create_from_graph(loaded.graph().clone(), loaded.source().to_string());
+            if !args.dry_run {
+                if let Some(provider) = loaded
+                    .run_record()
+                    .config
+                    .sandbox
+                    .as_ref()
+                    .and_then(|s| s.provider.as_deref())
+                    .and_then(|s| s.parse::<SandboxProvider>().ok())
+                {
+                    sandbox_provider = args.sandbox.map(Into::into).unwrap_or(provider);
+                }
+            }
+        }
+    }
+
+    let graph = validated.graph().clone();
+
+    eprintln!(
+        "{} {} from branch {} ({})",
+        styles.bold.apply_to("Resuming workflow:"),
+        graph.name,
+        styles.dim.apply_to(&run_branch),
+        run_id,
+    );
+
+    let (model_str, provider_str) = resolve_model_provider(
+        args.model.as_deref(),
+        args.provider.as_deref(),
+        run_cfg.as_ref(),
+        run_defaults,
+        &graph,
+    );
+    let cli_flags = super::create::CliFlags {
+        dry_run: args.dry_run,
+        auto_approve: args.auto_approve,
+        no_retro: args.no_retro,
+        verbose: args.verbose,
+        preserve_sandbox: args.preserve_sandbox,
+    };
+    let normalized = super::create::normalize_config(
+        run_cfg.as_ref(),
+        run_defaults,
+        &model_str,
+        provider_str.as_deref(),
+        sandbox_provider,
+        &graph,
+        cli_flags,
+    );
+    let persisted = fabro_workflows::pipeline::persist(
+        validated,
+        PersistOptions {
+            run_dir: run_dir.clone(),
+            run_record: fabro_workflows::records::RunRecord {
+                run_id: run_id.clone(),
+                created_at: chrono::Utc::now(),
+                config: normalized,
+                graph: graph.clone(),
+                workflow_slug: workflow_slug.clone(),
+                working_directory: resume_repo_path.clone(),
+                host_repo_path: Some(resume_repo_path.to_string_lossy().to_string()),
+                base_branch: detected_base_branch.clone(),
+                labels: args
+                    .label
+                    .iter()
+                    .filter_map(|s| s.split_once('='))
+                    .map(|(k, v)| (k.to_string(), v.to_string()))
+                    .collect(),
+            },
+        },
+    )?;
+    let run_cfg: Option<FabroConfig> = Some(persisted.run_record().config.clone());
+    let settings_config = persisted.run_record().config.clone();
+
     fabro_util::run_log::activate(&run_dir.join("cli.log"))
         .context("Failed to activate per-run log")?;
     let status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?;
     if !graph_source.is_empty() {
         tokio::fs::write(cached_graph_path(&run_dir), &graph_source).await?;
     }
-    let run_cfg: Option<FabroConfig> = run_cfg;
     // Git-branch resume: no original TOML available, skip debug snapshot.
     write_run_config_snapshot(&run_dir, None).await?;
 
-    // Write RunRecord for the resumed run
-    let settings_config = {
-        let (model_str, provider_str) = resolve_model_provider(
-            args.model.as_deref(),
-            args.provider.as_deref(),
-            run_cfg.as_ref(),
-            run_defaults,
-            &graph,
-        );
-        let cli_flags = super::create::CliFlags {
-            dry_run: args.dry_run,
-            auto_approve: args.auto_approve,
-            no_retro: args.no_retro,
-            verbose: args.verbose,
-            preserve_sandbox: args.preserve_sandbox,
-        };
-        let normalized = super::create::normalize_config(
-            run_cfg.as_ref(),
-            run_defaults,
-            &model_str,
-            provider_str.as_deref(),
-            sandbox_provider,
-            &graph,
-            cli_flags,
-        );
-        let record = fabro_workflows::records::RunRecord {
-            run_id: run_id.clone(),
-            created_at: chrono::Utc::now(),
-            config: normalized.clone(),
-            graph: graph.clone(),
-            workflow_slug: workflow_slug.clone(),
-            working_directory: resume_repo_path.clone(),
-            host_repo_path: Some(resume_repo_path.to_string_lossy().to_string()),
-            base_branch: detected_base_branch.clone(),
-            labels: std::collections::HashMap::new(),
-        };
-        let _ = record.save(&run_dir);
-        normalized
-    };
-
     let emitter = Arc::new(EventEmitter::new());
 
     // Resolve devcontainer BEFORE sandbox creation (mirrors run_command) so that
@@ -914,7 +951,7 @@ async fn prepare_from_branch(
 
     Ok(ResumeContext {
         checkpoint,
-        validated,
+        persisted,
         run_id,
         run_dir,
         run_cfg,
@@ -942,7 +979,7 @@ async fn run_resumed(
 ) -> anyhow::Result<()> {
     let ResumeContext {
         checkpoint,
-        validated,
+        persisted,
         run_id,
         run_dir,
         mut run_cfg,
@@ -959,7 +996,7 @@ async fn run_resumed(
         github_app,
         mut status_guard,
     } = ctx;
-    let graph = validated.graph().clone();
+    let graph = persisted.graph().clone();
 
     // Create progress UI (verbose mode shows detailed turn/tool counts and token usage)
     let is_tty = std::io::stderr().is_terminal();
@@ -1256,11 +1293,10 @@ async fn run_resumed(
         settings.pull_request().cloned()
     };
     let started = start(
-        validated,
+        persisted,
         StartOptions {
             init: fabro_workflows::pipeline::InitOptions {
                 run_id: run_id.clone(),
-                run_dir: run_dir.clone(),
                 dry_run: dry_run_mode,
                 emitter: Arc::clone(&emitter),
                 sandbox: Arc::clone(&sandbox),
diff --git a/lib/crates/fabro-cli/src/commands/run.rs b/lib/crates/fabro-cli/src/commands/run.rs
index 380a56e81..822a9af17 100644
--- a/lib/crates/fabro-cli/src/commands/run.rs
+++ b/lib/crates/fabro-cli/src/commands/run.rs
@@ -21,13 +21,12 @@ use fabro_workflows::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
 use fabro_workflows::git::GitSyncStatus;
 use fabro_workflows::handler::default_registry;
 use fabro_workflows::handler::llm::{AgentApiBackend, AgentCliBackend, BackendRouter};
-use fabro_workflows::operations::{
-    create_from_graph, start, StartFinalizeConfig, StartOptions, StartRetroConfig,
-};
+use fabro_workflows::operations::{start, StartFinalizeConfig, StartOptions, StartRetroConfig};
 use fabro_workflows::outcome::StageStatus;
 use fabro_workflows::outcome::{compute_stage_cost, format_cost};
 use fabro_workflows::pipeline::{
-    build_conclusion, classify_engine_result, persist_terminal_outcome,
+    build_conclusion, classify_engine_result, persist_terminal_outcome, PersistOptions, Persisted,
+    Validated,
 };
 use fabro_workflows::records::Checkpoint;
 use fabro_workflows::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings};
@@ -571,6 +570,11 @@ impl PreparedWorkflow {
     }
 }
 
+enum WorkflowState {
+    Validated(Validated),
+    Persisted(Box<Persisted>),
+}
+
 /// Resolve config, parse/validate the workflow graph, and resolve sandbox + model.
 ///
 /// Shared between `create_run` (which only persists the spec) and
@@ -747,7 +751,7 @@ pub(crate) fn prepare_workflow_with_project_config(
 
 /// Pre-prepared run state, used to skip workflow preparation in `run_command_impl`.
 struct RecordBasedRun {
-    validated: fabro_workflows::pipeline::Validated,
+    workflow: WorkflowState,
     raw_source: String,
     run_cfg: Option<FabroConfig>,
     sandbox_provider: SandboxProvider,
@@ -763,13 +767,14 @@ struct RecordBasedRun {
 ///
 /// Used by `run_engine_entrypoint` for detached runs that already have a RunRecord on disk.
 pub async fn run_from_record(
-    record: fabro_workflows::records::RunRecord,
+    persisted: Persisted,
     run_dir: PathBuf,
     run_defaults: FabroConfig,
     styles: &'static Styles,
     github_app: Option<fabro_github::GitHubAppCredentials>,
     git_author: fabro_workflows::git::GitAuthor,
 ) -> anyhow::Result<()> {
+    let record = persisted.run_record().clone();
     let sandbox_provider = record
         .config
         .sandbox
@@ -792,8 +797,8 @@ pub async fn run_from_record(
         .filter(|s| !s.is_empty());
 
     let record_run = RecordBasedRun {
+        workflow: WorkflowState::Persisted(Box::new(persisted)),
         raw_source: String::new(), // Raw DOT provenance is best-effort for record-based runs
-        validated: create_from_graph(record.graph.clone(), String::new()),
         run_cfg: Some(record.config.clone()),
         sandbox_provider,
         model: model.clone(),
@@ -817,7 +822,7 @@ pub async fn run_from_record(
         sandbox: Some(CliSandboxProvider::from(sandbox_provider)),
         label: record
             .labels
-            .into_iter()
+            .iter()
             .map(|(k, v)| format!("{k}={v}"))
             .collect(),
         no_retro: record.config.no_retro_enabled(),
@@ -828,7 +833,7 @@ pub async fn run_from_record(
             .and_then(|s| s.preserve)
             .unwrap_or(false),
         detach: false,
-        run_id: Some(record.run_id),
+        run_id: Some(record.run_id.clone()),
     };
 
     run_command_impl(args, styles, github_app, git_author, Some(record_run)).await
@@ -859,7 +864,7 @@ pub async fn run_command(
     } = prepare_workflow(&args, run_defaults, styles, false)?;
 
     let record_run = RecordBasedRun {
-        validated,
+        workflow: WorkflowState::Validated(validated),
         raw_source,
         run_cfg,
         sandbox_provider,
@@ -881,7 +886,7 @@ async fn run_command_impl(
     record_run: Option<RecordBasedRun>,
 ) -> anyhow::Result<()> {
     let (
-        validated,
+        workflow,
         raw_source,
         mut run_cfg,
         sandbox_provider,
@@ -892,7 +897,7 @@ async fn run_command_impl(
         workflow_toml_path,
     ) = match record_run {
         Some(rr) => (
-            rr.validated,
+            rr.workflow,
             rr.raw_source,
             rr.run_cfg,
             rr.sandbox_provider,
@@ -904,10 +909,13 @@ async fn run_command_impl(
         ),
         None => unreachable!("run_command_impl always receives a RecordBasedRun"),
     };
-    let graph = validated.graph().clone();
+    let graph = match &workflow {
+        WorkflowState::Validated(validated) => validated.graph().clone(),
+        WorkflowState::Persisted(persisted) => persisted.graph().clone(),
+    };
 
-    // For record-based runs from run_from_record, workflow is None (preparation was skipped).
-    let from_record = args.workflow.is_none();
+    // For record-based runs from run_from_record, the workflow has already been persisted.
+    let from_record = matches!(&workflow, WorkflowState::Persisted(_)) && args.workflow.is_none();
 
     // Collect setup commands — they'll be run inside the sandbox
     let setup_commands: Vec<String> = run_cfg
@@ -955,26 +963,61 @@ async fn run_command_impl(
     let run_dir = args
         .run_dir
         .unwrap_or_else(|| default_run_dir(&run_id, dry_run_flag));
-    tokio::fs::create_dir_all(&run_dir).await?;
     let cached_run_restart = if from_record {
-        // Record-based runs already have RunRecord on disk — skip re-writing.
-        true
+        false
     } else {
         let workflow_path = args.workflow.as_ref().unwrap();
         is_cached_run_restart(workflow_path, &run_dir)
     };
-    let existing_record = if cached_run_restart {
-        fabro_workflows::records::RunRecord::load(&run_dir).ok()
-    } else {
-        None
-    };
-    let workflow_slug = if cached_run_restart {
-        existing_record
-            .as_ref()
-            .and_then(|r| r.workflow_slug.clone())
-    } else {
-        prepared_workflow_slug
+
+    let persisted = match (cached_run_restart, workflow) {
+        (true, _) => Persisted::load(&run_dir)?,
+        (false, WorkflowState::Persisted(persisted)) => *persisted,
+        (false, WorkflowState::Validated(validated)) => {
+            let cli_flags = super::create::CliFlags {
+                dry_run: dry_run_flag,
+                auto_approve: auto_approve_flag,
+                no_retro: no_retro_flag,
+                verbose: verbose_flag,
+                preserve_sandbox: preserve_sandbox_flag,
+            };
+            let normalized_config = super::create::normalize_config(
+                run_cfg.as_ref(),
+                &run_defaults,
+                &model,
+                provider.as_deref(),
+                sandbox_provider,
+                validated.graph(),
+                cli_flags,
+            );
+            let run_record = fabro_workflows::records::RunRecord {
+                run_id: run_id.clone(),
+                created_at: chrono::Utc::now(),
+                config: normalized_config,
+                graph: validated.graph().clone(),
+                workflow_slug: prepared_workflow_slug.clone(),
+                working_directory: original_cwd.clone(),
+                host_repo_path: Some(original_cwd.to_string_lossy().to_string()),
+                base_branch: detected_base_branch.clone(),
+                labels: label_vec
+                    .iter()
+                    .filter_map(|s| s.split_once('='))
+                    .map(|(k, v)| (k.to_string(), v.to_string()))
+                    .collect(),
+            };
+            fabro_workflows::pipeline::persist(
+                validated,
+                PersistOptions {
+                    run_dir: run_dir.clone(),
+                    run_record,
+                },
+            )?
+        }
     };
+    run_cfg = Some(persisted.run_record().config.clone());
+    let workflow_slug = persisted.run_record().workflow_slug.clone();
+
+    tokio::fs::create_dir_all(&run_dir).await?;
     fabro_util::run_log::activate(&run_dir.join("cli.log"))
         .context("Failed to activate per-run log")?;
     if !from_record && !raw_source.is_empty() {
@@ -988,47 +1031,7 @@ async fn run_command_impl(
         write_run_config_snapshot(&run_dir, workflow_toml_path.as_deref()).await?;
     }
 
-    // Write RunRecord and compute normalized config for RunSettings
-    let settings_config = if !cached_run_restart {
-        let cli_flags = super::create::CliFlags {
-            dry_run: dry_run_flag,
-            auto_approve: auto_approve_flag,
-            no_retro: no_retro_flag,
-            verbose: verbose_flag,
-            preserve_sandbox: preserve_sandbox_flag,
-        };
-        let normalized_config = super::create::normalize_config(
-            run_cfg.as_ref(),
-            &run_defaults,
-            &model,
-            provider.as_deref(),
-            sandbox_provider,
-            &graph,
-            cli_flags,
-        );
-        let record = fabro_workflows::records::RunRecord {
-            run_id: run_id.clone(),
-            created_at: chrono::Utc::now(),
-            config: normalized_config.clone(),
-            graph: graph.clone(),
-            workflow_slug: workflow_slug.clone(),
-            working_directory: original_cwd.clone(),
-            host_repo_path: Some(original_cwd.to_string_lossy().to_string()),
-            base_branch: detected_base_branch.clone(),
-            labels: label_vec
-                .iter()
-                .filter_map(|s| s.split_once('='))
-                .map(|(k, v)| (k.to_string(), v.to_string()))
-                .collect(),
-        };
-        record.save(&run_dir)?;
-        normalized_config
-    } else {
-        existing_record
-            .as_ref()
-            .map(|r| r.config.clone())
-            .unwrap_or_default()
-    };
+    let settings_config = persisted.run_record().config.clone();
 
     // Now resolve ${env.VARNAME} references for runtime use.
     if let Some(ref mut cfg) = run_cfg {
@@ -1701,21 +1704,20 @@ async fn run_command_impl(
         cancel_token: None,
         dry_run: dry_run_mode,
         run_id: run_id.clone(),
-        labels: label_vec
-            .iter()
-            .filter_map(|s| s.split_once('='))
-            .map(|(k, v)| (k.to_string(), v.to_string()))
-            .collect(),
+        labels: persisted.run_record().labels.clone(),
         git_author: git_author.clone(),
         workflow_slug: workflow_slug.clone(),
         github_app: github_app.clone(),
-        base_branch: existing_record
-            .as_ref()
-            .and_then(|r| r.base_branch.clone())
+        base_branch: persisted
+            .run_record()
+            .base_branch
+            .clone()
             .or(detected_base_branch),
-        host_repo_path: existing_record
-            .as_ref()
-            .and_then(|r| r.host_repo_path.as_deref().map(PathBuf::from))
+        host_repo_path: persisted
+            .run_record()
+            .host_repo_path
+            .as_deref()
+            .map(PathBuf::from)
             .or_else(|| Some(original_cwd.clone())),
         git,
     };
@@ -1745,11 +1747,10 @@ async fn run_command_impl(
         config.pull_request().cloned()
     };
     let started = start(
-        validated,
+        persisted,
         StartOptions {
             init: fabro_workflows::pipeline::InitOptions {
                 run_id: run_id.clone(),
-                run_dir: run_dir.clone(),
                 dry_run: dry_run_mode,
                 emitter: Arc::clone(&emitter),
                 sandbox: Arc::clone(&sandbox),
diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs
index ed6d536fb..bf3eb0eec 100644
--- a/lib/crates/fabro-cli/src/main.rs
+++ b/lib/crates/fabro-cli/src/main.rs
@@ -324,10 +324,10 @@ async fn run_engine_entrypoint(
         cli_config.git_author().and_then(|a| a.email.clone()),
     );
 
-    let record = match fabro_workflows::records::RunRecord::load(&run_dir) {
-        Ok(record) => record,
+    let persisted = match fabro_workflows::pipeline::Persisted::load(&run_dir) {
+        Ok(persisted) => persisted,
         Err(err) => {
-            let anyhow_err: anyhow::Error = anyhow::anyhow!("Failed to load run record: {err}");
+            let anyhow_err: anyhow::Error = anyhow::anyhow!("Failed to load persisted run: {err}");
             let _ = commands::detached_support::persist_detached_failure(
                 &run_dir,
                 "bootstrap",
@@ -338,12 +338,14 @@ async fn run_engine_entrypoint(
         }
     };
 
-    if let Err(err) = std::env::set_current_dir(&record.working_directory).map_err(|e| {
-        anyhow::anyhow!(
-            "Failed to set working directory to {}: {e}",
-            record.working_directory.display()
-        )
-    }) {
+    if let Err(err) =
+        std::env::set_current_dir(&persisted.run_record().working_directory).map_err(|e| {
+            anyhow::anyhow!(
+                "Failed to set working directory to {}: {e}",
+                persisted.run_record().working_directory.display()
+            )
+        })
+    {
         let _ = commands::detached_support::persist_detached_failure(
             &run_dir,
             "bootstrap",
@@ -353,10 +355,10 @@ async fn run_engine_entrypoint(
         return Err(err);
     }
 
-    // Use run_from_record: loads config + graph directly from RunRecord,
+    // Use run_from_record: loads config + graph directly from persisted state,
     // skipping prepare_workflow() entirely. No TOML/DOT re-parsing needed.
     match commands::run::run_from_record(
-        record,
+        persisted,
         run_dir.clone(),
         cli_config,
         styles,
diff --git a/lib/crates/fabro-workflows/src/operations/start.rs b/lib/crates/fabro-workflows/src/operations/start.rs
index 9d9fa5312..0956bb11e 100644
--- a/lib/crates/fabro-workflows/src/operations/start.rs
+++ b/lib/crates/fabro-workflows/src/operations/start.rs
@@ -4,7 +4,7 @@ use std::time::{Duration, Instant};
 use crate::error::FabroError;
 use crate::event::WorkflowRunEvent;
 use crate::outcome::StageStatus;
-use crate::pipeline::{self, FinalizeOptions, Finalized, InitOptions, RetroOptions, Validated};
+use crate::pipeline::{self, FinalizeOptions, Finalized, InitOptions, Persisted, RetroOptions};
 
 pub struct StartRetroConfig {
     pub enabled: bool,
@@ -34,8 +34,8 @@ pub struct Started {
     pub retro_duration: Duration,
 }
 
-/// Run a validated workflow through initialize, execute, retro, and finalize.
-pub async fn start(validated: Validated, options: StartOptions) -> Result<Started, FabroError> {
+/// Run a persisted workflow through initialize, execute, retro, and finalize.
+pub async fn start(persisted: Persisted, options: StartOptions) -> Result<Started, FabroError> {
     let preserve_sandbox = options.finalize.preserve_sandbox;
     let sandbox_for_cleanup = Arc::clone(&options.init.sandbox);
     let cleanup_guard = scopeguard::guard((), move |()| {
@@ -49,7 +49,7 @@ pub async fn start(validated: Validated, options: StartOptions) -> Result<Starte
         }
     });
 
-    let initialized = pipeline::initialize(validated, options.init).await?;
+    let initialized = pipeline::initialize(persisted, options.init).await?;
 
     let last_git_sha: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
     {
@@ -123,6 +123,7 @@ mod tests {
     use std::sync::atomic::{AtomicUsize, Ordering};
 
     use async_trait::async_trait;
+    use chrono::Utc;
     use fabro_agent::{DirEntry, ExecResult, GrepOptions, LocalSandbox, Sandbox};
     use fabro_config::config::FabroConfig;
     use fabro_graphviz::graph::{Graph, Node};
@@ -135,6 +136,8 @@ mod tests {
     use crate::handler::start::StartHandler;
     use crate::handler::{Handler, HandlerRegistry};
     use crate::outcome::Outcome;
+    use crate::pipeline::PersistOptions;
+    use crate::records::RunRecord;
     use crate::run_settings::{LifecycleConfig, RunSettings};
 
     const MINIMAL_DOT: &str = r#"digraph Test {
@@ -340,11 +343,29 @@ mod tests {
         }
     }
 
-    fn validated_workflow(dot: &str) -> Validated {
+    fn persisted_workflow(dot: &str, run_dir: &std::path::Path) -> Persisted {
         let validated =
             crate::operations::create(dot, crate::operations::CreateOptions::default()).unwrap();
         validated.raise_on_errors().unwrap();
-        validated
+        let graph = validated.graph().clone();
+        crate::pipeline::persist(
+            validated,
+            PersistOptions {
+                run_dir: run_dir.to_path_buf(),
+                run_record: RunRecord {
+                    run_id: "run-test".to_string(),
+                    created_at: Utc::now(),
+                    config: FabroConfig::default(),
+                    graph,
+                    workflow_slug: Some("test".to_string()),
+                    working_directory: std::env::current_dir().unwrap(),
+                    host_repo_path: Some(std::env::current_dir().unwrap().display().to_string()),
+                    base_branch: Some("main".to_string()),
+                    labels: HashMap::new(),
+                },
+            },
+        )
+        .unwrap()
     }
 
     fn test_settings(run_dir: &std::path::Path) -> RunSettings {
@@ -383,7 +404,6 @@ mod tests {
         StartOptions {
             init: InitOptions {
                 run_id: "run-test".to_string(),
-                run_dir: run_dir.to_path_buf(),
                 dry_run: false,
                 emitter,
                 sandbox,
@@ -433,7 +453,7 @@ mod tests {
         let (sandbox, cleanup_count) = counting_sandbox();
 
         let result = start(
-            validated_workflow(MINIMAL_DOT),
+            persisted_workflow(MINIMAL_DOT, &run_dir),
             test_start_options(
                 &run_dir,
                 sandbox,
@@ -465,7 +485,7 @@ mod tests {
             Arc::new(LocalSandbox::new(std::env::current_dir().unwrap()));
 
         let started = start(
-            validated_workflow(EMIT_DOT),
+            persisted_workflow(EMIT_DOT, &run_dir),
             test_start_options(
                 &run_dir,
                 sandbox,
diff --git a/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs
index a552872a0..42b32e9aa 100644
--- a/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs
+++ b/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs
@@ -5,6 +5,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use async_trait::async_trait;
+use chrono::Utc;
 use fabro_agent::Sandbox;
 use fabro_config::config::FabroConfig;
 use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
@@ -18,11 +19,10 @@ use crate::event::{EventEmitter, WorkflowRunEvent};
 use crate::handler::default_registry;
 use crate::handler::start::StartHandler;
 use crate::handler::{Handler as HandlerTrait, HandlerRegistry};
-use crate::operations::create_from_graph;
 use crate::outcome::{Outcome, OutcomeExt, StageStatus};
 use crate::pipeline::initialize;
-use crate::pipeline::types::{InitOptions, Validated};
-use crate::records::Checkpoint;
+use crate::pipeline::types::{InitOptions, Persisted};
+use crate::records::{Checkpoint, RunRecord};
 use crate::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings};
 use crate::test_support::run_graph;
 
@@ -106,6 +106,31 @@ fn simple_validated_graph() -> (Graph, String) {
     (graph, source)
 }
 
+fn persisted_workflow(graph: Graph, source: String, run_dir: &Path, run_id: &str) -> Persisted {
+    Persisted::new(
+        graph.clone(),
+        source,
+        vec![],
+        run_dir.to_path_buf(),
+        RunRecord {
+            run_id: run_id.to_string(),
+            created_at: Utc::now(),
+            config: FabroConfig::default(),
+            graph,
+            workflow_slug: Some("test".to_string()),
+            working_directory: std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")),
+            host_repo_path: Some(
+                std::env::current_dir()
+                    .unwrap_or_else(|_| PathBuf::from("."))
+                    .display()
+                    .to_string(),
+            ),
+            base_branch: Some("main".to_string()),
+            labels: HashMap::new(),
+        },
+    )
+}
+
 fn test_lifecycle(setup_commands: Vec<String>) -> LifecycleConfig {
     LifecycleConfig {
         setup_commands,
@@ -118,12 +143,12 @@ fn test_lifecycle(setup_commands: Vec<String>) -> LifecycleConfig {
 async fn execute_runs_start_to_exit_and_returns_final_context() {
     let temp = tempfile::tempdir().unwrap();
     let run_dir = temp.path().join("run");
+    std::fs::create_dir_all(&run_dir).unwrap();
     let (graph, source) = simple_validated_graph();
     let initialized = initialize(
-        Validated::new(graph, source, vec![]),
+        persisted_workflow(graph, source, &run_dir, "run-test"),
         InitOptions {
             run_id: "run-test".to_string(),
-            run_dir: run_dir.clone(),
             dry_run: false,
             emitter: Arc::new(crate::event::EventEmitter::new()),
             sandbox: Arc::new(fabro_agent::LocalSandbox::new(
@@ -169,12 +194,11 @@ async fn run_with_lifecycle(
 ) -> Result<Outcome, FabroError> {
     let run_dir = settings.run_dir.clone();
     let run_id = settings.run_id.clone();
-    let validated = create_from_graph(graph.clone(), String::new());
+    std::fs::create_dir_all(&run_dir).unwrap();
     let initialized = initialize(
-        validated,
+        persisted_workflow(graph.clone(), String::new(), &run_dir, &run_id),
         InitOptions {
             run_id,
-            run_dir,
             dry_run: settings.dry_run,
             emitter,
             sandbox,
diff --git a/lib/crates/fabro-workflows/src/pipeline/initialize.rs b/lib/crates/fabro-workflows/src/pipeline/initialize.rs
index 2802a4eca..27fef880a 100644
--- a/lib/crates/fabro-workflows/src/pipeline/initialize.rs
+++ b/lib/crates/fabro-workflows/src/pipeline/initialize.rs
@@ -8,7 +8,7 @@ use crate::error::FabroError;
 use crate::event::WorkflowRunEvent;
 use crate::run_settings::GitCheckpointSettings;
 
-use super::types::{InitOptions, Initialized, Validated};
+use super::types::{InitOptions, Initialized, Persisted};
 
 async fn run_hooks(
     hook_runner: Option<&HookRunner>,
@@ -28,17 +28,11 @@ async fn run_hooks(
 ///
 /// Returns `FabroError` if sandbox preparation fails.
 pub async fn initialize(
-    validated: Validated,
+    persisted: Persisted,
     mut options: InitOptions,
 ) -> Result<Initialized, FabroError> {
-    let (graph, source, _diagnostics) = validated.into_parts();
-
-    // Create run directory and write graph
-    std::fs::create_dir_all(&options.run_dir)?;
-    if !source.is_empty() {
-        let graph_path = options.run_dir.join("graph.fabro");
-        std::fs::write(&graph_path, &source)?;
-    }
+    let (graph, source, _diagnostics, run_dir, _run_record) = persisted.into_parts();
+    options.run_settings.run_dir = run_dir;
 
     let hook_runner = if options.hooks.hooks.is_empty() {
         None
@@ -196,13 +190,14 @@ mod tests {
     use std::collections::HashMap;
     use std::sync::Arc;
 
+    use chrono::Utc;
     use fabro_config::config::FabroConfig;
     use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
     use fabro_interview::AutoApproveInterviewer;
 
     use super::*;
     use crate::handler::default_registry;
-    use crate::pipeline::types::Validated;
+    use crate::records::RunRecord;
     use crate::run_settings::RunSettings;
 
     fn simple_graph() -> (Graph, String) {
@@ -246,12 +241,33 @@ mod tests {
         }
     }
 
+    fn test_persisted(graph: Graph, source: String, run_dir: &std::path::Path) -> Persisted {
+        Persisted::new(
+            graph.clone(),
+            source,
+            vec![],
+            run_dir.to_path_buf(),
+            RunRecord {
+                run_id: "run-test".to_string(),
+                created_at: Utc::now(),
+                config: FabroConfig::default(),
+                graph,
+                workflow_slug: Some("test".to_string()),
+                working_directory: std::env::current_dir().unwrap(),
+                host_repo_path: Some(std::env::current_dir().unwrap().display().to_string()),
+                base_branch: Some("main".to_string()),
+                labels: HashMap::new(),
+            },
+        )
+    }
+
     #[tokio::test]
-    async fn initialize_prepares_sandbox_and_writes_graph() {
+    async fn initialize_prepares_sandbox_and_uses_persisted_run_dir() {
         let temp = tempfile::tempdir().unwrap();
         let run_dir = temp.path().join("run");
+        std::fs::create_dir_all(&run_dir).unwrap();
         let (graph, source) = simple_graph();
-        let validated = Validated::new(graph, source.clone(), vec![]);
+        let persisted = test_persisted(graph, source.clone(), &run_dir);
         let emitter = Arc::new(crate::event::EventEmitter::new());
         let sandbox = Arc::new(fabro_agent::LocalSandbox::new(
             std::env::current_dir().unwrap(),
@@ -259,10 +275,9 @@ mod tests {
         let registry = Arc::new(default_registry(Arc::new(AutoApproveInterviewer), || None));
 
         let initialized = initialize(
-            validated,
+            persisted,
             InitOptions {
                 run_id: "run-test".to_string(),
-                run_dir: run_dir.clone(),
                 dry_run: false,
                 emitter,
                 sandbox,
@@ -282,11 +297,8 @@ mod tests {
         .await
         .unwrap();
 
-        assert!(run_dir.join("graph.fabro").exists());
-        assert_eq!(
-            std::fs::read_to_string(run_dir.join("graph.fabro")).unwrap(),
-            source
-        );
+        assert_eq!(initialized.settings.run_dir, run_dir);
+        assert_eq!(initialized.source, source);
         assert!(initialized.hook_runner.is_none());
         assert_eq!(
             initialized.env.get("TEST_KEY").map(String::as_str),
@@ -298,8 +310,9 @@ mod tests {
     async fn initialize_skips_empty_graph_source() {
         let temp = tempfile::tempdir().unwrap();
         let run_dir = temp.path().join("run");
+        std::fs::create_dir_all(&run_dir).unwrap();
         let (graph, _source) = simple_graph();
-        let validated = Validated::new(graph, String::new(), vec![]);
+        let persisted = test_persisted(graph, String::new(), &run_dir);
         let emitter = Arc::new(crate::event::EventEmitter::new());
         let sandbox = Arc::new(fabro_agent::LocalSandbox::new(
             std::env::current_dir().unwrap(),
@@ -307,10 +320,9 @@ mod tests {
         let registry = Arc::new(default_registry(Arc::new(AutoApproveInterviewer), || None));
 
         let initialized = initialize(
-            validated,
+            persisted,
             InitOptions {
                 run_id: "run-test".to_string(),
-                run_dir: run_dir.clone(),
                 dry_run: false,
                 emitter,
                 sandbox,
@@ -330,7 +342,6 @@ mod tests {
         .await
         .unwrap();
 
-        assert!(!run_dir.join("graph.fabro").exists());
         assert!(initialized.source.is_empty());
     }
 }
diff --git a/lib/crates/fabro-workflows/src/pipeline/mod.rs b/lib/crates/fabro-workflows/src/pipeline/mod.rs
index ec100aa35..6853818c3 100644
--- a/lib/crates/fabro-workflows/src/pipeline/mod.rs
+++ b/lib/crates/fabro-workflows/src/pipeline/mod.rs
@@ -2,6 +2,7 @@ mod execute;
 mod finalize;
 mod initialize;
 mod parse;
+mod persist;
 mod retro;
 mod transform;
 pub mod types;
@@ -14,6 +15,7 @@ pub use finalize::{
 };
 pub use initialize::initialize;
 pub use parse::parse;
+pub use persist::persist;
 pub use retro::{retro, run_retro};
 pub use transform::transform;
 pub use types::*;
diff --git a/lib/crates/fabro-workflows/src/pipeline/persist.rs b/lib/crates/fabro-workflows/src/pipeline/persist.rs
new file mode 100644
index 000000000..0098e0d11
--- /dev/null
+++ b/lib/crates/fabro-workflows/src/pipeline/persist.rs
@@ -0,0 +1,333 @@
+use std::path::Path;
+
+use crate::error::FabroError;
+
+use super::types::{PersistOptions, Persisted, Validated};
+
+const GRAPH_FILE_NAME: &str = "graph.fabro";
+
+/// PERSIST phase: create run directory, write graph.fabro and run.json to disk.
+///
+/// Overwrites `run_record.graph` with the validated graph before saving.
+pub fn persist(validated: Validated, mut options: PersistOptions) -> Result<Persisted, FabroError> {
+    let (graph, source, diagnostics) = validated.into_parts();
+    options.run_record.graph = graph.clone();
+
+    std::fs::create_dir_all(&options.run_dir)?;
+    if !source.is_empty() {
+        std::fs::write(options.run_dir.join(GRAPH_FILE_NAME), &source)?;
+    }
+    options.run_record.save(&options.run_dir)?;
+
+    Ok(Persisted::new(
+        graph,
+        source,
+        diagnostics,
+        options.run_dir,
+        options.run_record,
+    ))
+}
+
+/// Load a previously persisted run from disk.
+///
+/// `run.json` is authoritative for graph + config; `graph.fabro` provides the
+/// original DOT source string when present.
+pub(crate) fn load(run_dir: &Path) -> Result<Persisted, FabroError> {
+    let run_record = crate::records::RunRecord::load(run_dir)?;
+    let graph = run_record.graph.clone();
+    let source = match std::fs::read_to_string(run_dir.join(GRAPH_FILE_NAME)) {
+        Ok(source) => source,
+        Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
+        Err(err) => return Err(err.into()),
+    };
+
+    Ok(Persisted::new(
+        graph,
+        source,
+        Vec::new(),
+        run_dir.to_path_buf(),
+        run_record,
+    ))
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+    use std::path::PathBuf;
+
+    use chrono::Utc;
+    use fabro_config::config::FabroConfig;
+    use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
+
+    use super::*;
+    use crate::records::RunRecord;
+
+    fn graph_and_source() -> (Graph, String) {
+        let source = r#"digraph test {
+  graph [goal="Ship feature"];
+  start [shape=Mdiamond];
+  exit [shape=Msquare];
+  start -> exit;
+}"#
+        .to_string();
+
+        let mut graph = Graph::new("test");
+        graph.attrs.insert(
+            "goal".to_string(),
+            AttrValue::String("Ship feature".to_string()),
+        );
+
+        let mut start = Node::new("start");
+        start.attrs.insert(
+            "shape".to_string(),
+            AttrValue::String("Mdiamond".to_string()),
+        );
+        graph.nodes.insert("start".to_string(), start);
+
+        let mut exit = Node::new("exit");
+        exit.attrs.insert(
+            "shape".to_string(),
+            AttrValue::String("Msquare".to_string()),
+        );
+        graph.nodes.insert("exit".to_string(), exit);
+
+        graph.edges.push(Edge::new("start", "exit"));
+        (graph, source)
+    }
+
+    fn different_graph() -> Graph {
+        let mut graph = Graph::new("different");
+        let mut start = Node::new("start");
+        start.attrs.insert(
+            "shape".to_string(),
+            AttrValue::String("Mdiamond".to_string()),
+        );
+        graph.nodes.insert("start".to_string(), start);
+        graph
+    }
+
+    fn sample_record(graph: Graph) -> RunRecord {
+        RunRecord {
+            run_id: "run-123".to_string(),
+            created_at: Utc::now(),
+            config: FabroConfig {
+                dry_run: Some(true),
+                verbose: Some(true),
+                ..Default::default()
+            },
+            graph,
+            workflow_slug: Some("ship".to_string()),
+            working_directory: PathBuf::from("/tmp/project"),
+            host_repo_path: Some("/tmp/project".to_string()),
+            base_branch: Some("main".to_string()),
+            labels: HashMap::from([
+                ("env".to_string(), "test".to_string()),
+                ("team".to_string(), "workflow".to_string()),
+            ]),
+        }
+    }
+
+    #[test]
+    fn persist_creates_run_dir_and_writes_graph_and_record() {
+        let temp = tempfile::tempdir().unwrap();
+        let run_dir = temp.path().join("run");
+        let (graph, source) = graph_and_source();
+        let persisted = persist(
+            Validated::new(graph.clone(), source.clone(), vec![]),
+            PersistOptions {
+                run_dir: run_dir.clone(),
+                run_record: sample_record(different_graph()),
+            },
+        )
+        .unwrap();
+
+        assert!(run_dir.is_dir());
+        assert_eq!(
+            std::fs::read_to_string(run_dir.join(GRAPH_FILE_NAME)).unwrap(),
+            source
+        );
+        assert!(run_dir.join(RunRecord::file_name()).exists());
+        assert_eq!(persisted.run_dir(), run_dir.as_path());
+        assert_eq!(
+            serde_json::to_value(persisted.run_record().graph.clone()).unwrap(),
+            serde_json::to_value(graph).unwrap()
+        );
+    }
+
+    #[test]
+    fn persist_skips_graph_file_when_source_is_empty() {
+        let temp = tempfile::tempdir().unwrap();
+        let run_dir = temp.path().join("run");
+        let (graph, _source) = graph_and_source();
+
+        persist(
+            Validated::new(graph, String::new(), vec![]),
+            PersistOptions {
+                run_dir: run_dir.clone(),
+                run_record: sample_record(different_graph()),
+            },
+        )
+        .unwrap();
+
+        assert!(!run_dir.join(GRAPH_FILE_NAME).exists());
+        assert!(run_dir.join(RunRecord::file_name()).exists());
+    }
+
+    #[test]
+    fn persist_overwrites_run_record_graph_with_validated_graph() {
+        let temp = tempfile::tempdir().unwrap();
+        let run_dir = temp.path().join("run");
+        let (graph, source) = graph_and_source();
+
+        persist(
+            Validated::new(graph.clone(), source, vec![]),
+            PersistOptions {
+                run_dir: run_dir.clone(),
+                run_record: sample_record(different_graph()),
+            },
+        )
+        .unwrap();
+
+        let saved = RunRecord::load(&run_dir).unwrap();
+        assert_eq!(saved.graph.name, graph.name);
+        assert!(saved.graph.nodes.contains_key("exit"));
+        assert_eq!(
+            serde_json::to_value(saved.graph).unwrap(),
+            serde_json::to_value(graph).unwrap()
+        );
+    }
+
+    #[test]
+    fn persist_roundtrips_full_run_record_fields_through_load() {
+        let temp = tempfile::tempdir().unwrap();
+        let run_dir = temp.path().join("run");
+        let (graph, source) = graph_and_source();
+        let mut expected = sample_record(different_graph());
+        expected.graph = graph.clone();
+
+        persist(
+            Validated::new(graph, source, vec![]),
+            PersistOptions {
+                run_dir: run_dir.clone(),
+                run_record: expected.clone(),
+            },
+        )
+        .unwrap();
+
+        let loaded = Persisted::load(&run_dir).unwrap();
+
+        assert_eq!(
+            serde_json::to_value(loaded.run_record()).unwrap(),
+            serde_json::to_value(expected).unwrap()
+        );
+        assert!(loaded.diagnostics().is_empty());
+    }
+
+    #[test]
+    fn persist_returns_error_on_io_failure() {
+        let temp = tempfile::tempdir().unwrap();
+        let run_dir = temp.path().join("run");
+        std::fs::write(&run_dir, "not a directory").unwrap();
+        let (graph, source) = graph_and_source();
+
+        let err = persist(
+            Validated::new(graph, source, vec![]),
+            PersistOptions {
+                run_dir,
+                run_record: sample_record(different_graph()),
+            },
+        )
+        .unwrap_err();
+
+        assert!(matches!(err, FabroError::Io(_)));
+    }
+
+    #[test]
+    fn load_roundtrips_persisted_workflow() {
+        let temp = tempfile::tempdir().unwrap();
+        let run_dir = temp.path().join("run");
+        let (graph, source) = graph_and_source();
+        let mut expected = sample_record(different_graph());
+        expected.graph = graph.clone();
+
+        let persisted = persist(
+            Validated::new(graph, source.clone(), vec![]),
+            PersistOptions {
+                run_dir: run_dir.clone(),
+                run_record: expected.clone(),
+            },
+        )
+        .unwrap();
+        let loaded = Persisted::load(&run_dir).unwrap();
+
+        assert_eq!(loaded.source(), source);
+        assert_eq!(loaded.run_dir(), run_dir.as_path());
+        assert_eq!(
+            serde_json::to_value(loaded.run_record()).unwrap(),
+            serde_json::to_value(expected).unwrap()
+        );
+        assert_eq!(
+            serde_json::to_value(loaded.graph()).unwrap(),
+            serde_json::to_value(persisted.graph()).unwrap()
+        );
+    }
+
+    #[test]
+    fn load_uses_empty_source_when_graph_file_is_missing() {
+        let temp = tempfile::tempdir().unwrap();
+        let run_dir = temp.path().join("run");
+        std::fs::create_dir_all(&run_dir).unwrap();
+        let (graph, _source) = graph_and_source();
+        let mut record = sample_record(different_graph());
+        record.graph = graph;
+        record.save(&run_dir).unwrap();
+
+        let loaded = Persisted::load(&run_dir).unwrap();
+
+        assert!(loaded.source().is_empty());
+    }
+
+    #[test]
+    fn load_reads_graph_from_run_json_and_source_from_graph_file() {
+        let temp = tempfile::tempdir().unwrap();
+        let run_dir = temp.path().join("run");
+        std::fs::create_dir_all(&run_dir).unwrap();
+
+        let (graph, _) = graph_and_source();
+        let mut record = sample_record(different_graph());
+        record.graph = graph.clone();
+        record.save(&run_dir).unwrap();
+        std::fs::write(
+            run_dir.join(GRAPH_FILE_NAME),
+            "digraph mismatch { a -> b; }",
+        )
+        .unwrap();
+
+        let loaded = Persisted::load(&run_dir).unwrap();
+
+        assert_eq!(
+            serde_json::to_value(loaded.graph()).unwrap(),
+            serde_json::to_value(graph).unwrap()
+        );
+        assert_eq!(loaded.source(), "digraph mismatch { a -> b; }");
+    }
+
+    #[test]
+    fn load_reads_graph_source_from_graph_file() {
+        let temp = tempfile::tempdir().unwrap();
+        let run_dir = temp.path().join("run");
+        let (graph, source) = graph_and_source();
+
+        persist(
+            Validated::new(graph, source.clone(), vec![]),
+            PersistOptions {
+                run_dir: run_dir.clone(),
+                run_record: sample_record(different_graph()),
+            },
+        )
+        .unwrap();
+
+        let loaded = Persisted::load(&run_dir).unwrap();
+        assert_eq!(loaded.source(), source);
+    }
+}
diff --git a/lib/crates/fabro-workflows/src/pipeline/types.rs b/lib/crates/fabro-workflows/src/pipeline/types.rs
index c339686e1..e5422c115 100644
--- a/lib/crates/fabro-workflows/src/pipeline/types.rs
+++ b/lib/crates/fabro-workflows/src/pipeline/types.rs
@@ -1,5 +1,5 @@
 use std::collections::HashMap;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
 use fabro_agent::Sandbox;
@@ -12,8 +12,7 @@ use crate::error::FabroError;
 use crate::event::EventEmitter;
 use crate::handler::HandlerRegistry;
 use crate::outcome::Outcome;
-use crate::records::Checkpoint;
-use crate::records::Conclusion;
+use crate::records::{Checkpoint, Conclusion, RunRecord};
 use crate::run_settings::{LifecycleConfig, RunSettings};
 use fabro_validate::Severity;
 
@@ -94,10 +93,104 @@ impl Validated {
     }
 }
 
+/// Options for the PERSIST phase.
+pub struct PersistOptions {
+    pub run_dir: PathBuf,
+    pub run_record: RunRecord,
+}
+
+/// Output of the PERSIST phase. Run directory created, run.json and graph.fabro written.
+#[derive(Debug)]
+#[non_exhaustive]
+pub struct Persisted {
+    graph: Graph,
+    source: String,
+    diagnostics: Vec<Diagnostic>,
+    run_dir: PathBuf,
+    run_record: RunRecord,
+}
+
+impl Persisted {
+    /// Create a new `Persisted` from its parts.
+    pub(crate) fn new(
+        graph: Graph,
+        source: String,
+        diagnostics: Vec<Diagnostic>,
+        run_dir: PathBuf,
+        run_record: RunRecord,
+    ) -> Self {
+        Self {
+            graph,
+            source,
+            diagnostics,
+            run_dir,
+            run_record,
+        }
+    }
+
+    pub fn graph(&self) -> &Graph {
+        &self.graph
+    }
+
+    pub fn source(&self) -> &str {
+        &self.source
+    }
+
+    pub fn diagnostics(&self) -> &[Diagnostic] {
+        &self.diagnostics
+    }
+
+    pub fn run_dir(&self) -> &Path {
+        &self.run_dir
+    }
+
+    pub fn run_record(&self) -> &RunRecord {
+        &self.run_record
+    }
+
+    /// True if any diagnostic has Error severity.
+    #[must_use]
+    pub fn has_errors(&self) -> bool {
+        self.diagnostics
+            .iter()
+            .any(|d| d.severity == Severity::Error)
+    }
+
+    /// Returns `Err(FabroError::Validation)` if any Error-severity diagnostics exist.
+    pub fn raise_on_errors(&self) -> Result<(), FabroError> {
+        if self.has_errors() {
+            let message = self
+                .diagnostics
+                .iter()
+                .filter(|d| d.severity == Severity::Error)
+                .map(|d| d.message.as_str())
+                .collect::<Vec<_>>()
+                .join("; ");
+            return Err(FabroError::Validation(message));
+        }
+        Ok(())
+    }
+
+    /// Consume into owned graph, source, diagnostics, run dir, and run record.
+    pub fn into_parts(self) -> (Graph, String, Vec<Diagnostic>, PathBuf, RunRecord) {
+        (
+            self.graph,
+            self.source,
+            self.diagnostics,
+            self.run_dir,
+            self.run_record,
+        )
+    }
+
+    /// Load a previously persisted run from disk.
+    pub fn load(run_dir: &Path) -> Result<Self, FabroError> {
+        super::persist::load(run_dir)
+    }
+}
+
 /// Options for the INITIALIZE phase.
 pub struct InitOptions {
     pub run_id: String,
-    pub run_dir: PathBuf,
     pub dry_run: bool,
     pub emitter: Arc<EventEmitter>,
     pub sandbox: Arc<dyn Sandbox>,