Make metadata rebuild recovery robust

lib/crates/fabro-cli/tests/it/cli.rs

@@ -1,8 +1,20 @@
+use std::collections::{BTreeSet, HashMap};
+use std::path::Path;
+use std::sync::Arc;
+use std::time::Duration;
+
 use assert_cmd::Command;
+use chrono::TimeZone;
 use fabro_config::FabroSettings;
 use fabro_config::mcp::McpTransport;
-use fabro_store::RuntimeState;
+use fabro_git_storage::branchstore::BranchStore;
+use fabro_git_storage::gitobj::Store as GitStore;
+use fabro_store::{NodeVisitRef, RuntimeState, SlateStore, Store as _};
+use fabro_types::{Checkpoint, Graph, RunRecord, StartRecord};
+use git2::{Repository, Signature};
+use object_store::local::LocalFileSystem;
 use predicates::prelude::*;
+use tokio::runtime::Runtime;
 
 #[allow(deprecated)]
 fn arc() -> Command {
@@ -205,6 +217,193 @@ commands = ["workflow-setup"]
     (home, project, storage_dir)
 }
 
+fn init_cli_home(storage_dir: &Path) -> tempfile::TempDir {
+    let home = tempfile::tempdir().unwrap();
+    let home_fabro = home.path().join(".fabro");
+    std::fs::create_dir_all(&home_fabro).unwrap();
+    let storage_dir = serde_json::to_string(&storage_dir.to_string_lossy().into_owned()).unwrap();
+    std::fs::write(
+        home_fabro.join("cli.toml"),
+        format!("storage_dir = {storage_dir}\n"),
+    )
+    .unwrap();
+    home
+}
+
+fn list_metadata_run_ids(repo_dir: &Path) -> BTreeSet<String> {
+    let repo = Repository::discover(repo_dir).unwrap();
+    repo.references()
+        .unwrap()
+        .flatten()
+        .filter_map(|reference| reference.name().map(ToOwned::to_owned))
+        .filter_map(|name| {
+            name.strip_prefix("refs/heads/fabro/meta/")
+                .map(ToOwned::to_owned)
+        })
+        .collect()
+}
+
+fn seed_run_branch(repo_dir: &Path, run_id: &str, nodes: &[&str]) -> Vec<String> {
+    let repo = Repository::discover(repo_dir).unwrap();
+    let store = GitStore::new(repo);
+    let sig = Signature::now("Fabro", "noreply@fabro.sh").unwrap();
+    let run_branch = format!("fabro/run/{run_id}");
+    let empty_tree = store.write_empty_tree().unwrap();
+    let mut shas = Vec::new();
+    let mut parent = None;
+
+    for node in nodes {
+        let parents = parent.into_iter().collect::<Vec<_>>();
+        let oid = store
+            .write_commit(
+                empty_tree,
+                &parents,
+                &format!("fabro({run_id}): {node} (completed)"),
+                &sig,
+            )
+            .unwrap();
+        store.update_ref(&run_branch, oid).unwrap();
+        shas.push(oid.to_string());
+        parent = Some(oid);
+    }
+
+    shas
+}
+
+fn checkpoint_record(
+    current_node: &str,
+    completed_nodes: &[&str],
+    node_visits: &[(&str, usize)],
+    git_commit_sha: Option<&str>,
+) -> Checkpoint {
+    Checkpoint {
+        timestamp: chrono::Utc
+            .with_ymd_and_hms(2026, 1, 1, 0, 0, 0)
+            .single()
+            .unwrap(),
+        current_node: current_node.to_string(),
+        completed_nodes: completed_nodes
+            .iter()
+            .map(|node| (*node).to_string())
+            .collect(),
+        node_retries: HashMap::new(),
+        context_values: HashMap::new(),
+        node_outcomes: HashMap::new(),
+        next_node_id: None,
+        git_commit_sha: git_commit_sha.map(ToOwned::to_owned),
+        loop_failure_signatures: HashMap::new(),
+        restart_failure_signatures: HashMap::new(),
+        node_visits: node_visits
+            .iter()
+            .map(|(node, visit)| ((*node).to_string(), *visit))
+            .collect(),
+    }
+}
+
+async fn seed_durable_run(storage_dir: &Path, repo_dir: &Path, run_id: &str) {
+    let store_path = storage_dir.join("store");
+    std::fs::create_dir_all(&store_path).unwrap();
+    let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_path).unwrap());
+    let store = SlateStore::new(object_store, "", Duration::from_millis(5));
+    let created_at = chrono::Utc
+        .with_ymd_and_hms(2026, 1, 1, 0, 0, 0)
+        .single()
+        .unwrap();
+    let run_store = store.create_run(run_id, created_at, None).await.unwrap();
+
+    let run_record = RunRecord {
+        run_id: run_id.to_string(),
+        created_at,
+        settings: FabroSettings::default(),
+        graph: Graph::default(),
+        workflow_slug: None,
+        working_directory: repo_dir.to_path_buf(),
+        host_repo_path: Some(repo_dir.to_string_lossy().into_owned()),
+        base_branch: None,
+        labels: HashMap::new(),
+    };
+    run_store.put_run(&run_record).await.unwrap();
+    run_store
+        .put_start(&StartRecord {
+            run_id: run_id.to_string(),
+            start_time: created_at,
+            run_branch: Some(format!("fabro/run/{run_id}")),
+            base_sha: None,
+        })
+        .await
+        .unwrap();
+
+    let start = NodeVisitRef {
+        node_id: "start",
+        visit: 1,
+    };
+    run_store
+        .put_node_prompt(&start, "start prompt")
+        .await
+        .unwrap();
+    let build = NodeVisitRef {
+        node_id: "build",
+        visit: 1,
+    };
+    run_store
+        .put_node_prompt(&build, "build prompt")
+        .await
+        .unwrap();
+
+    run_store
+        .append_checkpoint(&checkpoint_record(
+            "start",
+            &["start"],
+            &[("start", 1)],
+            None,
+        ))
+        .await
+        .unwrap();
+    run_store
+        .append_checkpoint(&checkpoint_record(
+            "build",
+            &["start", "build"],
+            &[("start", 1), ("build", 1)],
+            None,
+        ))
+        .await
+        .unwrap();
+}
+
+fn metadata_checkpoints(repo_dir: &Path, run_id: &str) -> Vec<Checkpoint> {
+    let repo = Repository::discover(repo_dir).unwrap();
+    let store = GitStore::new(repo);
+    let sig = Signature::now("Fabro", "noreply@fabro.sh").unwrap();
+    let branch = format!("fabro/meta/{run_id}");
+    let bs = BranchStore::new(&store, &branch, &sig);
+
+    bs.log(100)
+        .unwrap()
+        .iter()
+        .rev()
+        .filter(|commit| commit.message.starts_with("checkpoint"))
+        .map(|commit| {
+            serde_json::from_slice::<Checkpoint>(
+                &store
+                    .read_blob_at(commit.oid, "checkpoint.json")
+                    .unwrap()
+                    .unwrap(),
+            )
+            .unwrap()
+        })
+        .collect()
+}
+
+fn latest_metadata_checkpoint(repo_dir: &Path, run_id: &str) -> Checkpoint {
+    let repo = Repository::discover(repo_dir).unwrap();
+    let store = GitStore::new(repo);
+    let tip = store
+        .resolve_ref(&format!("fabro/meta/{run_id}"))
+        .unwrap()
+        .unwrap();
+    serde_json::from_slice(&store.read_blob_at(tip, "checkpoint.json").unwrap().unwrap()).unwrap()
+}
+
 // == LLM: prompt ==============================================================
 
 #[test]
@@ -762,6 +961,121 @@ fn run_help_no_longer_shows_resume_or_run_branch() {
         .stdout(predicate::str::contains("--run-branch").not());
 }
 
+#[test]
+fn rewind_and_fork_recover_missing_metadata_from_store() {
+    let storage_root = tempfile::tempdir().unwrap();
+    let storage_dir = storage_root.path().join("fabro-data");
+    let configured_home = init_cli_home(&storage_dir);
+    let repo_dir = tempfile::tempdir().unwrap();
+    Repository::init(repo_dir.path()).unwrap();
+
+    let source_run_id = "run-recovery-source";
+    let expected_shas = seed_run_branch(repo_dir.path(), source_run_id, &["start", "build"]);
+    Runtime::new().unwrap().block_on(seed_durable_run(
+        &storage_dir,
+        repo_dir.path(),
+        source_run_id,
+    ));
+
+    assert!(
+        list_metadata_run_ids(repo_dir.path()).is_empty(),
+        "metadata branch should start missing"
+    );
+
+    let rewind_list = arc()
+        .env("HOME", configured_home.path())
+        .env("NO_COLOR", "1")
+        .current_dir(repo_dir.path())
+        .args(["rewind", source_run_id, "--list"])
+        .timeout(Duration::from_secs(15))
+        .assert()
+        .success()
+        .get_output()
+        .stderr
+        .clone();
+    let rewind_list = String::from_utf8(rewind_list).unwrap();
+    assert!(
+        rewind_list.contains("@1"),
+        "expected first checkpoint: {rewind_list}"
+    );
+    assert!(
+        rewind_list.contains("@2"),
+        "expected second checkpoint: {rewind_list}"
+    );
+    assert!(
+        !rewind_list.contains("no run commit"),
+        "rebuilt timeline should persist backfilled SHAs: {rewind_list}"
+    );
+
+    let rebuilt_checkpoints = metadata_checkpoints(repo_dir.path(), source_run_id);
+    assert_eq!(rebuilt_checkpoints.len(), 2);
+    assert_eq!(
+        rebuilt_checkpoints[0].git_commit_sha.as_deref(),
+        Some(expected_shas[0].as_str())
+    );
+    assert_eq!(
+        rebuilt_checkpoints[1].git_commit_sha.as_deref(),
+        Some(expected_shas[1].as_str())
+    );
+
+    let before_child = list_metadata_run_ids(repo_dir.path());
+    arc()
+        .env("HOME", configured_home.path())
+        .env("NO_COLOR", "1")
+        .current_dir(repo_dir.path())
+        .args(["fork", source_run_id, "--no-push"])
+        .timeout(Duration::from_secs(15))
+        .assert()
+        .success();
+    let after_child = list_metadata_run_ids(repo_dir.path());
+    let child_run_ids: Vec<_> = after_child.difference(&before_child).cloned().collect();
+    assert_eq!(child_run_ids.len(), 1, "expected one child run");
+    let child_run_id = &child_run_ids[0];
+
+    let child_checkpoint = latest_metadata_checkpoint(repo_dir.path(), child_run_id);
+    assert_eq!(
+        child_checkpoint.git_commit_sha.as_deref(),
+        Some(expected_shas[1].as_str())
+    );
+
+    let child_rewind = arc()
+        .env("HOME", configured_home.path())
+        .env("NO_COLOR", "1")
+        .current_dir(repo_dir.path())
+        .args(["rewind", child_run_id, "@1", "--no-push"])
+        .timeout(Duration::from_secs(15))
+        .assert()
+        .success()
+        .get_output()
+        .stderr
+        .clone();
+    let child_rewind = String::from_utf8(child_rewind).unwrap();
+    assert!(
+        child_rewind.contains("Rewound run branch"),
+        "expected child rewind to move the run branch: {child_rewind}"
+    );
+    assert!(
+        !child_rewind.contains("has no git_commit_sha"),
+        "child rewind should not lose git_commit_sha: {child_rewind}"
+    );
+
+    let before_grandchild = after_child;
+    arc()
+        .env("HOME", configured_home.path())
+        .env("NO_COLOR", "1")
+        .current_dir(repo_dir.path())
+        .args(["fork", child_run_id, "--no-push"])
+        .timeout(Duration::from_secs(15))
+        .assert()
+        .success();
+    let after_grandchild = list_metadata_run_ids(repo_dir.path());
+    let grandchild_run_ids: Vec<_> = after_grandchild
+        .difference(&before_grandchild)
+        .cloned()
+        .collect();
+    assert_eq!(grandchild_run_ids.len(), 1, "expected one grandchild run");
+}
+
 // == Bug regression: create/start/attach lifecycle ============================
 
 /// Helper: create a minimal run directory that `resolve_run` can find.

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

@@ -4,6 +4,7 @@ use fabro_git_storage::gitobj::Store;
 use git2::{Oid, Signature};
 
 use crate::git::{MetadataStore, RUN_BRANCH_PREFIX, push_run_branches};
+use crate::records::Checkpoint;
 use crate::records::RunRecord;
 use crate::records::StartRecord;
 
@@ -116,21 +117,25 @@ fn fork_from_entry(
                 entry.metadata_commit_oid
             )
         })?;
+    let mut checkpoint: Checkpoint = serde_json::from_slice(&checkpoint_bytes)
+        .context("failed to parse source checkpoint.json")?;
+    checkpoint.git_commit_sha.clone_from(&entry.run_commit_sha);
+    let checkpoint_bytes =
+        serde_json::to_vec_pretty(&checkpoint).context("failed to serialize checkpoint.json")?;
 
-    let mut file_entries: Vec<(&str, &[u8])> = vec![
-        ("run.json", &new_run_record_bytes),
-        ("checkpoint.json", &checkpoint_bytes),
-    ];
+    let mut init_entries: Vec<(&str, &[u8])> = vec![("run.json", &new_run_record_bytes)];
     if let Some(ref start_record) = new_start_record_bytes {
-        file_entries.push(("start.json", start_record));
+        init_entries.push(("start.json", start_record));
     }
     if let Some(ref sandbox) = sandbox_bytes {
-        file_entries.push(("sandbox.json", sandbox));
+        init_entries.push(("sandbox.json", sandbox));
     }
 
-    let commit_msg = format!("fork from {} @{}", source_run_id, entry.ordinal);
     new_bs
-        .write_entries(&file_entries, &commit_msg)
+        .write_entries(&init_entries, "init run")
+        .map_err(|e| anyhow::anyhow!("failed to write init metadata entries: {e}"))?;
+    new_bs
+        .write_entry("checkpoint.json", &checkpoint_bytes, "checkpoint")
         .map_err(|e| anyhow::anyhow!("failed to write metadata entries: {e}"))?;
 
     if push {
@@ -242,7 +247,7 @@ mod tests {
     fn fork_creates_new_run_and_metadata_branches() {
         let (_dir, store) = temp_repo();
         let source_run_id = "run-source";
-        let _run_oids = setup_source_run(&store, source_run_id, &["start", "build", "test"]);
+        let run_oids = setup_source_run(&store, source_run_id, &["start", "build", "test"]);
 
         let new_run_id = fork(
             &store,
@@ -265,6 +270,14 @@ mod tests {
         let run_json = bs.read_entry("run.json").unwrap().unwrap();
         let run_record: RunRecord = serde_json::from_slice(&run_json).unwrap();
         assert_eq!(run_record.run_id, new_run_id);
+
+        let timeline = build_timeline(&store, &new_run_id).unwrap();
+        assert_eq!(timeline.entries.len(), 1);
+        assert_eq!(timeline.entries[0].node_name, "build");
+        assert_eq!(
+            timeline.entries[0].run_commit_sha,
+            Some(run_oids[1].to_string())
+        );
     }
 
     #[test]

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

@@ -9,9 +9,11 @@ use fabro_store::{
     ListRunsQuery, NodeVisitRef, RunStore as DurableRunStore, Store as DurableStore,
 };
 use git2::{Repository, Signature};
+use ulid::Ulid;
 
-use super::rewind::{RunTimeline, build_timeline};
+use super::rewind::{self, RunTimeline, build_timeline};
 use crate::git::MetadataStore;
+use crate::records::Checkpoint;
 
 pub async fn rebuild_metadata_branch(
     git_store: &GitStore,
@@ -29,69 +31,91 @@ pub async fn rebuild_metadata_branch(
         .ok_or_else(|| anyhow::anyhow!("run record not found for {run_id}"))?;
 
     let sig = Signature::now("Fabro", "noreply@fabro.sh")?;
-    let bs = BranchStore::new(git_store, &branch, &sig);
-    bs.ensure_branch()?;
+    let scratch_branch = format!("fabro/meta-rebuild/{run_id}/{}", Ulid::new());
+    let bs = BranchStore::new(git_store, &scratch_branch, &sig);
 
-    let mut init_entries = Vec::new();
-    init_entries.push((
-        "run.json".to_string(),
-        serde_json::to_vec_pretty(&run_record)?,
-    ));
-    if let Some(start) = run_store.get_start().await? {
-        init_entries.push(("start.json".to_string(), serde_json::to_vec_pretty(&start)?));
-    }
-    if let Some(sandbox) = run_store.get_sandbox().await? {
+    let result = async {
+        bs.ensure_branch()?;
+
+        let mut init_entries = Vec::new();
         init_entries.push((
-            "sandbox.json".to_string(),
-            serde_json::to_vec_pretty(&sandbox)?,
+            "run.json".to_string(),
+            serde_json::to_vec_pretty(&run_record)?,
         ));
-    }
-    write_entries(&bs, &init_entries, "init run")?;
+        if let Some(start) = run_store.get_start().await? {
+            init_entries.push(("start.json".to_string(), serde_json::to_vec_pretty(&start)?));
+        }
+        if let Some(sandbox) = run_store.get_sandbox().await? {
+            init_entries.push((
+                "sandbox.json".to_string(),
+                serde_json::to_vec_pretty(&sandbox)?,
+            ));
+        }
+        write_entries(&bs, &init_entries, "init run")?;
 
-    for (_seq, checkpoint) in run_store.list_checkpoints().await? {
-        let mut entries = Vec::new();
-        entries.push((
-            "checkpoint.json".to_string(),
-            serde_json::to_vec_pretty(&checkpoint)?,
-        ));
+        let mut checkpoints = run_store.list_checkpoints().await?;
+        backfill_missing_checkpoint_shas(git_store, run_id, &mut checkpoints);
 
-        for node_id in &checkpoint.completed_nodes {
-            let max_visit = checkpoint.node_visits.get(node_id).copied().unwrap_or(1);
-            for visit in 1..=max_visit {
-                let visit = u32::try_from(visit)
-                    .with_context(|| format!("visit {visit} for node {node_id} exceeds u32"))?;
-                let node = run_store.get_node(&NodeVisitRef { node_id, visit }).await?;
+        for (_seq, checkpoint) in checkpoints {
+            let mut entries = Vec::new();
+            entries.push((
+                "checkpoint.json".to_string(),
+                serde_json::to_vec_pretty(&checkpoint)?,
+            ));
 
-                if let Some(prompt) = node.prompt {
-                    entries.push((
-                        node_file_path(node_id, visit, "prompt.md"),
-                        prompt.into_bytes(),
-                    ));
-                }
-                if let Some(response) = node.response {
-                    entries.push((
-                        node_file_path(node_id, visit, "response.md"),
-                        response.into_bytes(),
-                    ));
-                }
-                if let Some(status) = node.status {
-                    entries.push((
-                        node_file_path(node_id, visit, "status.json"),
-                        serde_json::to_vec_pretty(&status)?,
-                    ));
+            for node_id in &checkpoint.completed_nodes {
+                let max_visit = checkpoint.node_visits.get(node_id).copied().unwrap_or(1);
+                for visit in 1..=max_visit {
+                    let visit = u32::try_from(visit)
+                        .with_context(|| format!("visit {visit} for node {node_id} exceeds u32"))?;
+                    let node = run_store.get_node(&NodeVisitRef { node_id, visit }).await?;
+
+                    if let Some(prompt) = node.prompt {
+                        entries.push((
+                            node_file_path(node_id, visit, "prompt.md"),
+                            prompt.into_bytes(),
+                        ));
+                    }
+                    if let Some(response) = node.response {
+                        entries.push((
+                            node_file_path(node_id, visit, "response.md"),
+                            response.into_bytes(),
+                        ));
+                    }
+                    if let Some(status) = node.status {
+                        entries.push((
+                            node_file_path(node_id, visit, "status.json"),
+                            serde_json::to_vec_pretty(&status)?,
+                        ));
+                    }
                 }
             }
+
+            write_entries(&bs, &entries, "checkpoint")?;
         }
 
-        write_entries(&bs, &entries, "checkpoint")?;
-    }
+        if let Some(retro) = run_store.get_retro().await? {
+            let entries = vec![("retro.json".to_string(), serde_json::to_vec_pretty(&retro)?)];
+            write_entries(&bs, &entries, "finalize run")?;
+        }
 
-    if let Some(retro) = run_store.get_retro().await? {
-        let entries = vec![("retro.json".to_string(), serde_json::to_vec_pretty(&retro)?)];
-        write_entries(&bs, &entries, "finalize run")?;
+        Ok::<(), anyhow::Error>(())
     }
+    .await;
 
-    Ok(())
+    let final_result = match result {
+        Ok(()) => {
+            let scratch_tip = git_store
+                .resolve_ref(&scratch_branch)?
+                .ok_or_else(|| anyhow::anyhow!("scratch metadata branch missing after rebuild"))?;
+            git_store.update_ref(&branch, scratch_tip)?;
+            Ok(())
+        }
+        Err(err) => Err(err),
+    };
+
+    let _ = git_store.delete_ref(&scratch_branch);
+    final_result
 }
 
 pub async fn build_timeline_or_rebuild(
@@ -177,6 +201,38 @@ fn write_entries(
     Ok(())
 }
 
+fn backfill_missing_checkpoint_shas(
+    git_store: &GitStore,
+    run_id: &str,
+    checkpoints: &mut [(u32, Checkpoint)],
+) {
+    if !checkpoints
+        .iter()
+        .any(|(_, checkpoint)| checkpoint.git_commit_sha.is_none())
+    {
+        return;
+    }
+
+    let node_commits = rewind::run_commit_shas_by_node(git_store, run_id);
+    let mut node_indices: HashMap<String, usize> = HashMap::new();
+
+    for (_seq, checkpoint) in checkpoints.iter_mut() {
+        if checkpoint.git_commit_sha.is_some() {
+            continue;
+        }
+
+        if let Some(shas) = node_commits.get(&checkpoint.current_node) {
+            let idx = node_indices
+                .entry(checkpoint.current_node.clone())
+                .or_insert(0);
+            if *idx < shas.len() {
+                checkpoint.git_commit_sha = Some(shas[*idx].clone());
+                *idx += 1;
+            }
+        }
+    }
+}
+
 fn node_file_path(node_id: &str, visit: u32, filename: &str) -> String {
     if visit <= 1 {
         format!("nodes/{node_id}/{filename}")
@@ -339,6 +395,31 @@ mod tests {
         run_store
     }
 
+    fn seed_run_branch(git_store: &GitStore, run_id: &str, nodes: &[&str]) -> Vec<String> {
+        let sig = test_sig();
+        let run_branch = format!("fabro/run/{run_id}");
+        let empty_tree = git_store.write_empty_tree().unwrap();
+        let mut shas = Vec::new();
+        let mut parent = None;
+
+        for node in nodes {
+            let parents = parent.into_iter().collect::<Vec<_>>();
+            let oid = git_store
+                .write_commit(
+                    empty_tree,
+                    &parents,
+                    &format!("fabro({run_id}): {node} (completed)"),
+                    &sig,
+                )
+                .unwrap();
+            git_store.update_ref(&run_branch, oid).unwrap();
+            shas.push(oid.to_string());
+            parent = Some(oid);
+        }
+
+        shas
+    }
+
     #[tokio::test]
     async fn rebuild_metadata_branch_round_trips_timeline() {
         let (_dir, git_store) = temp_repo();
@@ -694,4 +775,122 @@ mod tests {
             .unwrap();
         assert_eq!(from_refs, "abc-123");
     }
+
+    #[tokio::test]
+    async fn rebuild_metadata_branch_persists_backfilled_run_shas_in_checkpoint_blobs() {
+        let (_dir, git_store) = temp_repo();
+        let durable_store = InMemoryStore::default();
+        let run_store = create_run_store(&durable_store, "run-1", None).await;
+
+        run_store
+            .append_checkpoint(&sample_checkpoint(
+                "start",
+                &["start"],
+                &[("start", 1)],
+                None,
+            ))
+            .await
+            .unwrap();
+        run_store
+            .append_checkpoint(&sample_checkpoint(
+                "build",
+                &["start", "build"],
+                &[("start", 1), ("build", 1)],
+                None,
+            ))
+            .await
+            .unwrap();
+
+        let expected_shas = seed_run_branch(&git_store, "run-1", &["start", "build"]);
+
+        rebuild_metadata_branch(&git_store, run_store.as_ref(), "run-1")
+            .await
+            .unwrap();
+
+        let sig = test_sig();
+        let branch = MetadataStore::branch_name("run-1");
+        let bs = BranchStore::new(&git_store, &branch, &sig);
+        let checkpoint_commits: Vec<_> = bs
+            .log(100)
+            .unwrap()
+            .iter()
+            .rev()
+            .filter(|commit| commit.message.starts_with("checkpoint"))
+            .map(|commit| commit.oid)
+            .collect();
+
+        let first: Checkpoint = serde_json::from_slice(
+            &git_store
+                .read_blob_at(checkpoint_commits[0], "checkpoint.json")
+                .unwrap()
+                .unwrap(),
+        )
+        .unwrap();
+        let second: Checkpoint = serde_json::from_slice(
+            &git_store
+                .read_blob_at(checkpoint_commits[1], "checkpoint.json")
+                .unwrap()
+                .unwrap(),
+        )
+        .unwrap();
+
+        assert_eq!(
+            first.git_commit_sha.as_deref(),
+            Some(expected_shas[0].as_str())
+        );
+        assert_eq!(
+            second.git_commit_sha.as_deref(),
+            Some(expected_shas[1].as_str())
+        );
+    }
+
+    #[tokio::test]
+    async fn rebuild_metadata_branch_is_atomic_on_failure() {
+        let (_dir, git_store) = temp_repo();
+        let durable_store = InMemoryStore::default();
+        let run_store = create_run_store(&durable_store, "run-1", None).await;
+
+        let bad_node = "bad\0node";
+        let bad_visit = NodeVisitRef {
+            node_id: bad_node,
+            visit: 1,
+        };
+        run_store
+            .put_node_prompt(&bad_visit, "prompt")
+            .await
+            .unwrap();
+        run_store
+            .append_checkpoint(&sample_checkpoint(
+                bad_node,
+                &[bad_node],
+                &[(bad_node, 1)],
+                None,
+            ))
+            .await
+            .unwrap();
+
+        let err = rebuild_metadata_branch(&git_store, run_store.as_ref(), "run-1")
+            .await
+            .unwrap_err();
+        assert!(err.to_string().contains("nul") || err.to_string().contains("NUL"));
+        assert!(
+            git_store
+                .resolve_ref(&MetadataStore::branch_name("run-1"))
+                .unwrap()
+                .is_none()
+        );
+
+        let scratch_refs: Vec<_> = git_store
+            .repo()
+            .references()
+            .unwrap()
+            .flatten()
+            .filter_map(|reference| reference.name().map(ToOwned::to_owned))
+            .filter(|name| name.starts_with("refs/heads/fabro/meta-rebuild/run-1/"))
+            .collect();
+        assert!(
+            scratch_refs.is_empty(),
+            "leftover scratch refs: {scratch_refs:?}"
+        );
+    }
 }

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

@@ -166,13 +166,31 @@ fn backfill_run_shas(store: &Store, run_id: &str, timeline: &mut [TimelineEntry]
         return;
     }
 
+    let node_commits = run_commit_shas_by_node(store, run_id);
+    let mut node_indices: HashMap<String, usize> = HashMap::new();
+
+    for entry in timeline.iter_mut() {
+        if entry.run_commit_sha.is_some() {
+            continue;
+        }
+        if let Some(shas) = node_commits.get(&entry.node_name) {
+            let idx = node_indices.entry(entry.node_name.clone()).or_insert(0);
+            if *idx < shas.len() {
+                entry.run_commit_sha = Some(shas[*idx].clone());
+                *idx += 1;
+            }
+        }
+    }
+}
+
+pub(crate) fn run_commit_shas_by_node(store: &Store, run_id: &str) -> HashMap<String, Vec<String>> {
     let run_branch = format!("{RUN_BRANCH_PREFIX}{run_id}");
     let Ok(sig) = Signature::now("Fabro", "noreply@fabro.sh") else {
-        return;
+        return HashMap::new();
     };
     let bs = BranchStore::new(store, &run_branch, &sig);
     let Ok(run_commits) = bs.log(10_000) else {
-        return;
+        return HashMap::new();
     };
 
     let prefix = format!("fabro({run_id}): ");
@@ -191,20 +209,8 @@ fn backfill_run_shas(store: &Store, run_id: &str, timeline: &mut [TimelineEntry]
     for shas in node_commits.values_mut() {
         shas.reverse();
     }
-    let mut node_indices: HashMap<String, usize> = HashMap::new();
 
-    for entry in timeline.iter_mut() {
-        if entry.run_commit_sha.is_some() {
-            continue;
-        }
-        if let Some(shas) = node_commits.get(&entry.node_name) {
-            let idx = node_indices.entry(entry.node_name.clone()).or_insert(0);
-            if *idx < shas.len() {
-                entry.run_commit_sha = Some(shas[*idx].clone());
-                *idx += 1;
-            }
-        }
-    }
+    node_commits
 }
 
 fn detect_parallel_interior(graph: &Graph) -> HashMap<String, String> {