diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index cfa00d501..5940246ac 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -116,6 +116,8 @@ enum Command { #[command(subcommand)] command: SkillCommand, }, + /// Rewind a workflow run to an earlier checkpoint + Rewind(fabro_workflows::cli::rewind::RewindArgs), /// Workflow operations Workflow { #[command(subcommand)] @@ -311,6 +313,7 @@ async fn main_inner() -> (String, Result<()>) { PrCommand::Merge(_) => "pr merge", PrCommand::Close(_) => "pr close", }, + Command::Rewind(_) => "rewind", Command::Workflow { command } => match command { WorkflowCommand::List(_) => "workflow list", }, @@ -607,6 +610,10 @@ async fn main_inner() -> (String, Result<()>) { } } } + Command::Rewind(args) => { + let styles = fabro_util::terminal::Styles::detect_stderr(); + fabro_workflows::cli::rewind::rewind_command(&args, &styles)?; + } Command::Workflow { command } => match command { WorkflowCommand::List(args) => { fabro_workflows::cli::workflow::workflow_list_command(&args)?; diff --git a/lib/crates/fabro-git-storage/src/gitobj.rs b/lib/crates/fabro-git-storage/src/gitobj.rs index 991513504..3c7a9ae0f 100644 --- a/lib/crates/fabro-git-storage/src/gitobj.rs +++ b/lib/crates/fabro-git-storage/src/gitobj.rs @@ -182,6 +182,20 @@ impl Store { } } + /// Read a blob from the tree of a specific commit. Returns `None` if the path doesn't exist. + pub fn read_blob_at(&self, commit_oid: Oid, path: &str) -> Result>> { + let commit = self.repo.find_commit(commit_oid)?; + let tree = commit.tree()?; + match tree.get_path(std::path::Path::new(path)) { + Ok(entry) => { + let blob = self.repo.find_blob(entry.id())?; + Ok(Some(blob.content().to_vec())) + } + Err(e) if e.code() == git2::ErrorCode::NotFound => Ok(None), + Err(e) => Err(e.into()), + } + } + /// Delete a branch reference. No-op if branch doesn't exist. pub fn delete_ref(&self, branch: &str) -> Result<()> { let refname = format!("refs/heads/{branch}"); @@ -537,4 +551,36 @@ mod tests { let (_dir, store) = temp_repo(); store.delete_ref("nonexistent").unwrap(); } + + // -- read_blob_at -- + + #[test] + fn read_blob_at_returns_content() { + let (_dir, store) = temp_repo(); + let sig = Signature::now("Test", "test@example.com").unwrap(); + let bs = crate::branchstore::BranchStore::new(&store, "test/data", &sig); + bs.ensure_branch().unwrap(); + bs.write_entry("hello.txt", b"world", "add hello").unwrap(); + + let log = bs.log(1).unwrap(); + let commit_oid = log[0].oid; + + let content = store.read_blob_at(commit_oid, "hello.txt").unwrap(); + assert_eq!(content.unwrap(), b"world"); + } + + #[test] + fn read_blob_at_returns_none_for_missing_path() { + let (_dir, store) = temp_repo(); + let sig = Signature::now("Test", "test@example.com").unwrap(); + let bs = crate::branchstore::BranchStore::new(&store, "test/data", &sig); + bs.ensure_branch().unwrap(); + bs.write_entry("hello.txt", b"world", "add hello").unwrap(); + + let log = bs.log(1).unwrap(); + let commit_oid = log[0].oid; + + let content = store.read_blob_at(commit_oid, "nonexistent.txt").unwrap(); + assert!(content.is_none()); + } } diff --git a/lib/crates/fabro-workflows/src/cli/mod.rs b/lib/crates/fabro-workflows/src/cli/mod.rs index d2d2f1e63..c8451b768 100644 --- a/lib/crates/fabro-workflows/src/cli/mod.rs +++ b/lib/crates/fabro-workflows/src/cli/mod.rs @@ -8,6 +8,7 @@ pub mod pr; pub mod preview; pub mod progress; pub mod project_config; +pub mod rewind; pub mod run; pub mod run_config; pub mod runs; diff --git a/lib/crates/fabro-workflows/src/cli/rewind.rs b/lib/crates/fabro-workflows/src/cli/rewind.rs new file mode 100644 index 000000000..c7927dde0 --- /dev/null +++ b/lib/crates/fabro-workflows/src/cli/rewind.rs @@ -0,0 +1,869 @@ +use std::collections::HashMap; + +use anyhow::{bail, Context, Result}; +use clap::Args; +use fabro_git_storage::branchstore::{BranchStore, CommitInfo}; +use fabro_git_storage::gitobj::Store; +use fabro_util::terminal::Styles; +use git2::{Oid, Repository, Signature}; + +use crate::checkpoint::Checkpoint; +use crate::git::MetadataStore; +use crate::graph::types::Graph; + +/// Rewind a workflow run to an earlier checkpoint. +#[derive(Debug, Args)] +pub struct RewindArgs { + /// Run ID (or unambiguous prefix) + pub run_id: String, + + /// Target checkpoint: node name, node@visit, or @ordinal (omit with --list) + pub target: Option, + + /// Show the checkpoint timeline instead of rewinding + #[arg(long)] + pub list: bool, + + /// Skip force-pushing rewound refs to the remote + #[arg(long)] + pub no_push: bool, +} + +/// Parsed rewind target. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RewindTarget { + /// @N — the Nth checkpoint (1-based) + Ordinal(usize), + /// node_name — most recent visit of the named node + LatestVisit(String), + /// node_name@N — the Nth visit of the named node + SpecificVisit(String, usize), +} + +/// One row in the checkpoint timeline. +#[derive(Debug, Clone)] +pub struct TimelineEntry { + /// 1-based ordinal (checkpoint sequence number) + pub ordinal: usize, + /// The node that was just completed at this checkpoint + pub node_name: String, + /// Visit number for this node (from node_visits) + pub visit: usize, + /// OID of the metadata-branch commit that contains this checkpoint + pub metadata_commit_oid: Oid, + /// SHA of the run-branch commit captured at this checkpoint + pub run_commit_sha: Option, +} + +/// Parse a target string into a `RewindTarget`. +pub fn parse_target(s: &str) -> Result { + if let Some(rest) = s.strip_prefix('@') { + let n: usize = rest + .parse() + .with_context(|| format!("invalid ordinal: @{rest}"))?; + if n == 0 { + bail!("ordinal must be >= 1"); + } + return Ok(RewindTarget::Ordinal(n)); + } + if let Some(at_pos) = s.rfind('@') { + let name = &s[..at_pos]; + let visit_str = &s[at_pos + 1..]; + if !name.is_empty() && !visit_str.is_empty() { + if let Ok(visit) = visit_str.parse::() { + if visit == 0 { + bail!("visit number must be >= 1"); + } + return Ok(RewindTarget::SpecificVisit(name.to_string(), visit)); + } + } + } + Ok(RewindTarget::LatestVisit(s.to_string())) +} + +/// Build the checkpoint timeline by walking the metadata branch oldest-first. +pub fn build_timeline(store: &Store, run_id: &str) -> Result> { + let branch = MetadataStore::branch_name(run_id); + let sig = Signature::now("Fabro", "noreply@fabro.sh")?; + let bs = BranchStore::new(store, &branch, &sig); + + let commits = bs + .log(10_000) + .map_err(|e| anyhow::anyhow!("failed to read metadata branch log: {e}"))?; + + // Reverse to oldest-first + let commits: Vec<&CommitInfo> = commits.iter().rev().collect(); + + let mut timeline = Vec::new(); + let mut ordinal = 0usize; + + for commit in &commits { + if !commit.message.starts_with("checkpoint") { + continue; + } + let blob = store + .read_blob_at(commit.oid, "checkpoint.json") + .map_err(|e| anyhow::anyhow!("failed to read checkpoint blob: {e}"))?; + let Some(bytes) = blob else { continue }; + let cp: Checkpoint = serde_json::from_slice(&bytes) + .with_context(|| format!("failed to parse checkpoint at {}", commit.oid))?; + + ordinal += 1; + let visit = cp.node_visits.get(&cp.current_node).copied().unwrap_or(1); + + timeline.push(TimelineEntry { + ordinal, + node_name: cp.current_node.clone(), + visit, + metadata_commit_oid: commit.oid, + run_commit_sha: cp.git_commit_sha.clone(), + }); + } + + Ok(timeline) +} + +/// Map interior parallel nodes to their fan-out parallel node ID. +pub fn detect_parallel_interior(graph: &Graph) -> HashMap { + let mut interior_map = HashMap::new(); + + for node in graph.nodes.values() { + if node.handler_type() != Some("parallel") { + continue; + } + let parallel_id = &node.id; + // BFS from parallel node to find interior nodes until we hit the fan_in + let mut queue: Vec = graph + .outgoing_edges(parallel_id) + .iter() + .map(|e| e.to.clone()) + .collect(); + let mut visited = std::collections::HashSet::new(); + + while let Some(current) = queue.pop() { + if !visited.insert(current.clone()) { + continue; + } + if let Some(n) = graph.nodes.get(¤t) { + if n.handler_type() == Some("parallel.fan_in") { + continue; // don't traverse past fan_in + } + } + interior_map.insert(current.clone(), parallel_id.clone()); + for edge in graph.outgoing_edges(¤t) { + queue.push(edge.to.clone()); + } + } + } + + interior_map +} + +/// Resolve a target to a timeline entry, with parallel snap-back. +pub fn resolve_target<'a>( + timeline: &'a [TimelineEntry], + target: &RewindTarget, + parallel_map: &HashMap, +) -> Result<&'a TimelineEntry> { + match target { + RewindTarget::Ordinal(n) => timeline + .iter() + .find(|e| e.ordinal == *n) + .ok_or_else(|| anyhow::anyhow!("ordinal @{n} out of range (max @{})", timeline.len())), + + RewindTarget::LatestVisit(name) => { + let effective_name = parallel_map.get(name).unwrap_or(name); + timeline + .iter() + .rev() + .find(|e| e.node_name == *effective_name) + .ok_or_else(|| { + if effective_name != name { + anyhow::anyhow!( + "node '{name}' is inside parallel '{effective_name}'; \ + no checkpoint found for '{effective_name}'" + ) + } else { + anyhow::anyhow!("no checkpoint found for node '{name}'") + } + }) + } + + RewindTarget::SpecificVisit(name, visit) => { + let effective_name = parallel_map.get(name).unwrap_or(name); + timeline + .iter() + .find(|e| e.node_name == *effective_name && e.visit == *visit) + .ok_or_else(|| { + if effective_name != name { + anyhow::anyhow!( + "node '{name}' is inside parallel '{effective_name}'; \ + no visit {visit} found for '{effective_name}'" + ) + } else { + anyhow::anyhow!("no visit {visit} found for node '{name}'") + } + }) + } + } +} + +/// Print the timeline table to stderr. +pub fn print_timeline( + timeline: &[TimelineEntry], + parallel_map: &HashMap, + styles: &Styles, +) { + if timeline.is_empty() { + eprintln!("No checkpoints found."); + return; + } + + eprintln!( + " {} {} {}", + styles.bold_dim.apply_to(format!("{:<6}", "@")), + styles.bold_dim.apply_to(format!("{:<30}", "Node")), + styles.bold_dim.apply_to("Details"), + ); + + for entry in timeline { + let ordinal_str = format!("@{}", entry.ordinal); + let mut details = Vec::new(); + if entry.visit > 1 { + details.push(format!("visit {}, loop", entry.visit)); + } + if parallel_map.contains_key(&entry.node_name) { + details.push("parallel interior".to_string()); + } + if entry.run_commit_sha.is_none() { + details.push("no run commit".to_string()); + } + + let detail_str = if details.is_empty() { + String::new() + } else { + format!("({})", details.join(", ")) + }; + + eprintln!( + " {} {:<30} {}", + styles.cyan.apply_to(format!("{ordinal_str:<6}")), + entry.node_name, + styles.dim.apply_to(detail_str), + ); + } +} + +/// Move both refs backward to the target checkpoint. +pub fn execute_rewind( + store: &Store, + run_id: &str, + entry: &TimelineEntry, + push: bool, +) -> Result<()> { + // Move metadata branch ref + let meta_branch = MetadataStore::branch_name(run_id); + store + .update_ref(&meta_branch, entry.metadata_commit_oid) + .map_err(|e| anyhow::anyhow!("failed to update metadata ref: {e}"))?; + eprintln!( + "Rewound metadata branch to @{} ({})", + entry.ordinal, entry.node_name + ); + + // Move run branch ref + let run_branch = format!("arc/run/{run_id}"); + match &entry.run_commit_sha { + Some(sha) => { + let oid = + Oid::from_str(sha).with_context(|| format!("invalid run commit SHA: {sha}"))?; + store + .update_ref(&run_branch, oid) + .map_err(|e| anyhow::anyhow!("failed to update run branch ref: {e}"))?; + eprintln!("Rewound run branch arc/run/{run_id} to {}", &sha[..8]); + } + None => { + eprintln!( + "Warning: checkpoint @{} has no git_commit_sha; run branch not moved", + entry.ordinal + ); + } + } + + // Optionally push to remote + if push { + let repo_path = store + .repo() + .workdir() + .or_else(|| store.repo().path().parent()) + .unwrap_or(store.repo().path()); + + // Check if run branch has a remote tracking ref + let remote_ref = format!("refs/remotes/origin/{run_branch}"); + let has_remote_tracking = store.repo().find_reference(&remote_ref).is_ok(); + + if has_remote_tracking { + eprintln!("Force-pushing rewound branches to origin..."); + + // Force-push run branch + if entry.run_commit_sha.is_some() { + let refspec = format!("+refs/heads/{run_branch}:refs/heads/{run_branch}"); + crate::git::push_branch(repo_path, "origin", &refspec) + .map_err(|e| anyhow::anyhow!("failed to push run branch: {e}"))?; + } + + // Force-push metadata branch + let meta_refspec = format!("+refs/heads/{meta_branch}:refs/heads/fabro/meta/{run_id}"); + crate::git::push_branch(repo_path, "origin", &meta_refspec) + .map_err(|e| anyhow::anyhow!("failed to push metadata branch: {e}"))?; + + eprintln!("Remote refs updated."); + } + } + + Ok(()) +} + +/// Find a run ID by exact match or unambiguous prefix. +pub fn find_run_id_by_prefix(repo: &Repository, prefix: &str) -> Result { + let refs = repo.references()?; + let pattern = "refs/heads/refs/fabro/"; + let mut matches = Vec::new(); + + for reference in refs.flatten() { + let name = match reference.name() { + Some(n) => n, + None => continue, + }; + if let Some(run_id) = name.strip_prefix(pattern) { + if run_id == prefix { + return Ok(run_id.to_string()); + } + if run_id.starts_with(prefix) { + matches.push(run_id.to_string()); + } + } + } + + match matches.len() { + 0 => bail!("no run found matching '{prefix}'"), + 1 => Ok(matches.into_iter().next().unwrap()), + _ => { + let mut msg = format!("ambiguous run ID prefix '{prefix}', matches:\n"); + for m in &matches { + msg.push_str(&format!(" {m}\n")); + } + bail!("{msg}") + } + } +} + +/// Entry point for `fabro rewind`. +pub fn rewind_command(args: &RewindArgs, styles: &Styles) -> Result<()> { + let repo = Repository::discover(".").context("not in a git repository")?; + let run_id = find_run_id_by_prefix(&repo, &args.run_id)?; + let store = Store::new(repo); + + let timeline = build_timeline(&store, &run_id)?; + + if args.list || args.target.is_none() { + // Read graph for parallel detection + let parallel_map = load_parallel_map(&store, &run_id); + print_timeline(&timeline, ¶llel_map, styles); + return Ok(()); + } + + let target_str = args.target.as_ref().unwrap(); + let target = parse_target(target_str)?; + + let parallel_map = load_parallel_map(&store, &run_id); + let entry = resolve_target(&timeline, &target, ¶llel_map)?; + + execute_rewind(&store, &run_id, entry, !args.no_push)?; + + eprintln!("\nTo resume: fabro run --run-branch arc/run/{}", run_id); + + Ok(()) +} + +/// Load the graph from the metadata branch and build the parallel interior map. +fn load_parallel_map(store: &Store, run_id: &str) -> HashMap { + let branch = MetadataStore::branch_name(run_id); + let sig = match Signature::now("Fabro", "noreply@fabro.sh") { + Ok(s) => s, + Err(_) => return HashMap::new(), + }; + let bs = BranchStore::new(store, &branch, &sig); + let graph_bytes = match bs.read_entry("graph.fabro") { + Ok(Some(bytes)) => bytes, + _ => return HashMap::new(), + }; + let dot_source = String::from_utf8_lossy(&graph_bytes); + let graph = match crate::parser::parse(&dot_source) { + Ok(g) => g, + Err(_) => return HashMap::new(), + }; + detect_parallel_interior(&graph) +} + +#[cfg(test)] +mod tests { + use super::*; + + // -- parse_target tests -- + + #[test] + fn parse_target_ordinal() { + assert_eq!(parse_target("@4").unwrap(), RewindTarget::Ordinal(4)); + } + + #[test] + fn parse_target_ordinal_one() { + assert_eq!(parse_target("@1").unwrap(), RewindTarget::Ordinal(1)); + } + + #[test] + fn parse_target_ordinal_zero_errors() { + assert!(parse_target("@0").is_err()); + } + + #[test] + fn parse_target_latest_visit() { + assert_eq!( + parse_target("step2").unwrap(), + RewindTarget::LatestVisit("step2".to_string()) + ); + } + + #[test] + fn parse_target_specific_visit() { + assert_eq!( + parse_target("step3@2").unwrap(), + RewindTarget::SpecificVisit("step3".to_string(), 2) + ); + } + + #[test] + fn parse_target_specific_visit_zero_errors() { + assert!(parse_target("step3@0").is_err()); + } + + // -- build_timeline tests -- + + fn temp_repo() -> (tempfile::TempDir, Store) { + let dir = tempfile::TempDir::new().unwrap(); + let repo = Repository::init(dir.path()).unwrap(); + (dir, Store::new(repo)) + } + + fn test_sig() -> Signature<'static> { + Signature::now("Test", "test@example.com").unwrap() + } + + fn make_checkpoint_json(current_node: &str, visit: usize, git_sha: Option<&str>) -> Vec { + let mut node_visits = HashMap::new(); + node_visits.insert(current_node.to_string(), visit); + let cp = serde_json::json!({ + "timestamp": "2025-01-01T00:00:00Z", + "current_node": current_node, + "completed_nodes": [current_node], + "node_retries": {}, + "context_values": {}, + "logs": [], + "node_visits": node_visits, + "git_commit_sha": git_sha, + }); + serde_json::to_vec(&cp).unwrap() + } + + #[test] + fn build_timeline_simple() { + let (_dir, store) = temp_repo(); + let sig = test_sig(); + let branch = MetadataStore::branch_name("test-run-1"); + let bs = BranchStore::new(&store, &branch, &sig); + bs.ensure_branch().unwrap(); + + // init commit (should be skipped) + bs.write_entry("manifest.json", b"{}", "init run").unwrap(); + + // 3 checkpoint commits + let cp1 = make_checkpoint_json("start", 1, Some("aaa")); + bs.write_entry("checkpoint.json", &cp1, "checkpoint") + .unwrap(); + + let cp2 = make_checkpoint_json("build", 1, Some("bbb")); + bs.write_entry("checkpoint.json", &cp2, "checkpoint") + .unwrap(); + + let cp3 = make_checkpoint_json("test", 1, Some("ccc")); + bs.write_entry("checkpoint.json", &cp3, "checkpoint") + .unwrap(); + + let timeline = build_timeline(&store, "test-run-1").unwrap(); + assert_eq!(timeline.len(), 3); + assert_eq!(timeline[0].ordinal, 1); + assert_eq!(timeline[0].node_name, "start"); + assert_eq!(timeline[0].visit, 1); + assert_eq!(timeline[1].ordinal, 2); + assert_eq!(timeline[1].node_name, "build"); + assert_eq!(timeline[2].ordinal, 3); + assert_eq!(timeline[2].node_name, "test"); + } + + #[test] + fn build_timeline_skips_non_checkpoint_commits() { + let (_dir, store) = temp_repo(); + let sig = test_sig(); + let branch = MetadataStore::branch_name("test-run-2"); + let bs = BranchStore::new(&store, &branch, &sig); + bs.ensure_branch().unwrap(); + + bs.write_entry("manifest.json", b"{}", "init run").unwrap(); + + let cp1 = make_checkpoint_json("start", 1, None); + bs.write_entry("checkpoint.json", &cp1, "checkpoint") + .unwrap(); + + // finalize commit — should be skipped + bs.write_entry("retro.json", b"{}", "finalize").unwrap(); + + let timeline = build_timeline(&store, "test-run-2").unwrap(); + assert_eq!(timeline.len(), 1); + assert_eq!(timeline[0].node_name, "start"); + } + + // -- resolve_target tests -- + + fn make_timeline() -> Vec { + vec![ + TimelineEntry { + ordinal: 1, + node_name: "start".to_string(), + visit: 1, + metadata_commit_oid: Oid::zero(), + run_commit_sha: Some("aaa".to_string()), + }, + TimelineEntry { + ordinal: 2, + node_name: "build".to_string(), + visit: 1, + metadata_commit_oid: Oid::zero(), + run_commit_sha: Some("bbb".to_string()), + }, + TimelineEntry { + ordinal: 3, + node_name: "build".to_string(), + visit: 2, + metadata_commit_oid: Oid::zero(), + run_commit_sha: Some("ccc".to_string()), + }, + ] + } + + #[test] + fn resolve_ordinal() { + let timeline = make_timeline(); + let entry = resolve_target(&timeline, &RewindTarget::Ordinal(2), &HashMap::new()).unwrap(); + assert_eq!(entry.ordinal, 2); + assert_eq!(entry.node_name, "build"); + } + + #[test] + fn resolve_latest_visit() { + let timeline = make_timeline(); + let entry = resolve_target( + &timeline, + &RewindTarget::LatestVisit("build".to_string()), + &HashMap::new(), + ) + .unwrap(); + assert_eq!(entry.ordinal, 3); + assert_eq!(entry.visit, 2); + } + + #[test] + fn resolve_specific_visit() { + let timeline = make_timeline(); + let entry = resolve_target( + &timeline, + &RewindTarget::SpecificVisit("build".to_string(), 1), + &HashMap::new(), + ) + .unwrap(); + assert_eq!(entry.ordinal, 2); + assert_eq!(entry.visit, 1); + } + + #[test] + fn resolve_ordinal_out_of_range() { + let timeline = make_timeline(); + let result = resolve_target(&timeline, &RewindTarget::Ordinal(99), &HashMap::new()); + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains("out of range")); + } + + #[test] + fn resolve_unknown_node() { + let timeline = make_timeline(); + let result = resolve_target( + &timeline, + &RewindTarget::LatestVisit("nonexistent".to_string()), + &HashMap::new(), + ); + assert!(result.is_err()); + assert!(result + .unwrap_err() + .to_string() + .contains("no checkpoint found")); + } + + // -- detect_parallel_interior tests -- + + #[test] + fn parallel_interior_detection() { + let mut graph = Graph::new("test"); + let mut parallel_node = crate::graph::types::Node::new("parallel1"); + parallel_node.attrs.insert( + "shape".to_string(), + crate::graph::types::AttrValue::String("component".to_string()), + ); + graph.nodes.insert("parallel1".to_string(), parallel_node); + + let mut fan_in = crate::graph::types::Node::new("fan_in1"); + fan_in.attrs.insert( + "shape".to_string(), + crate::graph::types::AttrValue::String("tripleoctagon".to_string()), + ); + graph.nodes.insert("fan_in1".to_string(), fan_in); + + let mut a = crate::graph::types::Node::new("a"); + a.attrs.insert( + "shape".to_string(), + crate::graph::types::AttrValue::String("box".to_string()), + ); + graph.nodes.insert("a".to_string(), a); + + let mut b = crate::graph::types::Node::new("b"); + b.attrs.insert( + "shape".to_string(), + crate::graph::types::AttrValue::String("box".to_string()), + ); + graph.nodes.insert("b".to_string(), b); + + graph.edges.push(crate::graph::types::Edge { + from: "parallel1".to_string(), + to: "a".to_string(), + attrs: HashMap::new(), + }); + graph.edges.push(crate::graph::types::Edge { + from: "parallel1".to_string(), + to: "b".to_string(), + attrs: HashMap::new(), + }); + graph.edges.push(crate::graph::types::Edge { + from: "a".to_string(), + to: "fan_in1".to_string(), + attrs: HashMap::new(), + }); + graph.edges.push(crate::graph::types::Edge { + from: "b".to_string(), + to: "fan_in1".to_string(), + attrs: HashMap::new(), + }); + + let map = detect_parallel_interior(&graph); + assert_eq!(map.get("a"), Some(&"parallel1".to_string())); + assert_eq!(map.get("b"), Some(&"parallel1".to_string())); + assert!(!map.contains_key("parallel1")); + assert!(!map.contains_key("fan_in1")); + } + + #[test] + fn parallel_snap_back() { + let timeline = vec![ + TimelineEntry { + ordinal: 1, + node_name: "parallel1".to_string(), + visit: 1, + metadata_commit_oid: Oid::zero(), + run_commit_sha: Some("aaa".to_string()), + }, + TimelineEntry { + ordinal: 2, + node_name: "a".to_string(), + visit: 1, + metadata_commit_oid: Oid::zero(), + run_commit_sha: Some("bbb".to_string()), + }, + ]; + + let mut parallel_map = HashMap::new(); + parallel_map.insert("a".to_string(), "parallel1".to_string()); + + // Targeting "a" should snap back to "parallel1" + let entry = resolve_target( + &timeline, + &RewindTarget::LatestVisit("a".to_string()), + ¶llel_map, + ) + .unwrap(); + assert_eq!(entry.node_name, "parallel1"); + assert_eq!(entry.ordinal, 1); + } + + // -- execute_rewind tests -- + + #[test] + fn execute_rewind_moves_metadata_ref() { + let (_dir, store) = temp_repo(); + let sig = test_sig(); + let branch = MetadataStore::branch_name("run-1"); + let bs = BranchStore::new(&store, &branch, &sig); + bs.ensure_branch().unwrap(); + + bs.write_entry("manifest.json", b"{}", "init run").unwrap(); + + let cp1 = make_checkpoint_json("start", 1, None); + let oid1 = bs + .write_entry("checkpoint.json", &cp1, "checkpoint") + .unwrap(); + + let cp2 = make_checkpoint_json("build", 1, None); + bs.write_entry("checkpoint.json", &cp2, "checkpoint") + .unwrap(); + + let cp3 = make_checkpoint_json("test", 1, None); + bs.write_entry("checkpoint.json", &cp3, "checkpoint") + .unwrap(); + + let timeline = build_timeline(&store, "run-1").unwrap(); + let entry = &timeline[0]; // @1 = start + + execute_rewind(&store, "run-1", entry, false).unwrap(); + + // Verify metadata ref points to the @1 commit + let resolved = store.resolve_ref(&branch).unwrap().unwrap(); + assert_eq!(resolved, oid1); + } + + #[test] + fn execute_rewind_moves_run_branch_ref() { + let (_dir, store) = temp_repo(); + let sig = test_sig(); + + // Create a run branch with some commits + let run_branch = "arc/run/run-2"; + let empty_tree = store.write_empty_tree().unwrap(); + let run_c1 = store + .write_commit(empty_tree, &[], "run commit 1", &sig) + .unwrap(); + store.update_ref(run_branch, run_c1).unwrap(); + let run_c2 = store + .write_commit(empty_tree, &[run_c1], "run commit 2", &sig) + .unwrap(); + store.update_ref(run_branch, run_c2).unwrap(); + + // Create metadata branch with checkpoints pointing to run commits + let meta_branch = MetadataStore::branch_name("run-2"); + let meta_bs = BranchStore::new(&store, &meta_branch, &sig); + meta_bs.ensure_branch().unwrap(); + meta_bs + .write_entry("manifest.json", b"{}", "init run") + .unwrap(); + + let cp1 = make_checkpoint_json("start", 1, Some(&run_c1.to_string())); + meta_bs + .write_entry("checkpoint.json", &cp1, "checkpoint") + .unwrap(); + + let cp2 = make_checkpoint_json("build", 1, Some(&run_c2.to_string())); + meta_bs + .write_entry("checkpoint.json", &cp2, "checkpoint") + .unwrap(); + + let timeline = build_timeline(&store, "run-2").unwrap(); + let entry = &timeline[0]; // @1 + + execute_rewind(&store, "run-2", entry, false).unwrap(); + + // Verify run branch ref moved to run_c1 + let resolved = store.resolve_ref(run_branch).unwrap().unwrap(); + assert_eq!(resolved, run_c1); + } + + #[test] + fn execute_rewind_warns_on_missing_run_sha() { + let (_dir, store) = temp_repo(); + let sig = test_sig(); + let branch = MetadataStore::branch_name("run-3"); + let bs = BranchStore::new(&store, &branch, &sig); + bs.ensure_branch().unwrap(); + + bs.write_entry("manifest.json", b"{}", "init run").unwrap(); + + let cp1 = make_checkpoint_json("start", 1, None); + let oid1 = bs + .write_entry("checkpoint.json", &cp1, "checkpoint") + .unwrap(); + + let timeline = build_timeline(&store, "run-3").unwrap(); + + // Should not panic even though run_commit_sha is None + execute_rewind(&store, "run-3", &timeline[0], false).unwrap(); + + // Metadata ref should still be moved + let resolved = store.resolve_ref(&branch).unwrap().unwrap(); + assert_eq!(resolved, oid1); + } + + // -- find_run_id_by_prefix tests -- + + #[test] + fn find_run_id_exact_match() { + let (_dir, store) = temp_repo(); + let sig = test_sig(); + let branch = MetadataStore::branch_name("abc-123"); + let bs = BranchStore::new(&store, &branch, &sig); + bs.ensure_branch().unwrap(); + + let result = find_run_id_by_prefix(store.repo(), "abc-123").unwrap(); + assert_eq!(result, "abc-123"); + } + + #[test] + fn find_run_id_prefix_match() { + let (_dir, store) = temp_repo(); + let sig = test_sig(); + let branch = MetadataStore::branch_name("abc-123-long-id"); + let bs = BranchStore::new(&store, &branch, &sig); + bs.ensure_branch().unwrap(); + + let result = find_run_id_by_prefix(store.repo(), "abc-123").unwrap(); + assert_eq!(result, "abc-123-long-id"); + } + + #[test] + fn find_run_id_ambiguous() { + let (_dir, store) = temp_repo(); + let sig = test_sig(); + + let b1 = MetadataStore::branch_name("abc-111"); + BranchStore::new(&store, &b1, &sig).ensure_branch().unwrap(); + + let b2 = MetadataStore::branch_name("abc-222"); + BranchStore::new(&store, &b2, &sig).ensure_branch().unwrap(); + + let result = find_run_id_by_prefix(store.repo(), "abc"); + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains("ambiguous")); + } + + #[test] + fn find_run_id_not_found() { + let (_dir, store) = temp_repo(); + let result = find_run_id_by_prefix(store.repo(), "nonexistent"); + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains("no run found")); + } +}