From d593f5973de971f92dac7cd42e757557dd021c12 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 29 Mar 2026 09:10:50 -0400 Subject: [PATCH] Rebuild metadata branches from the run store --- lib/crates/fabro-cli/src/commands/run/fork.rs | 15 +- lib/crates/fabro-cli/src/commands/run/mod.rs | 4 +- .../fabro-cli/src/commands/run/rewind.rs | 16 +- .../fabro-workflows/src/operations/hydrate.rs | 6 +- .../fabro-workflows/src/operations/mod.rs | 4 + .../src/operations/rebuild_meta.rs | 697 ++++++++++++++++++ 6 files changed, 729 insertions(+), 13 deletions(-) create mode 100644 lib/crates/fabro-workflows/src/operations/rebuild_meta.rs diff --git a/lib/crates/fabro-cli/src/commands/run/fork.rs b/lib/crates/fabro-cli/src/commands/run/fork.rs index 894fde574..d654eebc8 100644 --- a/lib/crates/fabro-cli/src/commands/run/fork.rs +++ b/lib/crates/fabro-cli/src/commands/run/fork.rs @@ -1,20 +1,27 @@ use anyhow::Context; use anyhow::Result; +use fabro_config::FabroSettingsExt; use fabro_git_storage::gitobj::Store; use fabro_util::terminal::Styles; use fabro_workflows::operations::{ - ForkRunInput, RewindTarget, build_timeline, find_run_id_by_prefix, fork, + ForkRunInput, RewindTarget, build_timeline_or_rebuild, find_run_id_by_prefix_or_store, fork, }; use git2::Repository; use crate::args::ForkArgs; +use crate::cli_config::load_cli_settings; +use crate::store::{build_store, open_run_reader}; -pub(crate) fn run(args: &ForkArgs, styles: &Styles) -> Result<()> { +pub(crate) async fn run(args: &ForkArgs, 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 cli_settings = load_cli_settings(None)?; + let durable_store = build_store(&cli_settings.storage_dir())?; + let run_id = + find_run_id_by_prefix_or_store(&repo, durable_store.as_ref(), &args.run_id).await?; let store = Store::new(repo); + let run_store = open_run_reader(&cli_settings.storage_dir(), &run_id).await?; - let timeline = build_timeline(&store, &run_id)?; + let timeline = build_timeline_or_rebuild(&store, run_store.as_deref(), &run_id).await?; if args.list { super::rewind::print_timeline(&timeline, styles); diff --git a/lib/crates/fabro-cli/src/commands/run/mod.rs b/lib/crates/fabro-cli/src/commands/run/mod.rs index 956d71d60..c80e5a268 100644 --- a/lib/crates/fabro-cli/src/commands/run/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/mod.rs @@ -87,11 +87,11 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( } RunCommands::Rewind(args) => { let styles = Styles::detect_stderr(); - rewind::run(&args, &styles) + rewind::run(&args, &styles).await } RunCommands::Fork(args) => { let styles = Styles::detect_stderr(); - fork::run(&args, &styles) + fork::run(&args, &styles).await } RunCommands::Wait(args) => { let styles = Styles::detect_stderr(); diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index 2fb86d1a8..327a6befb 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs @@ -2,22 +2,30 @@ use anyhow::Context; use anyhow::Result; use cli_table::format::{Border, Separator}; use cli_table::{Cell, CellStruct, Color, Style, Table, print_stderr}; +use fabro_config::FabroSettingsExt; use fabro_git_storage::gitobj::Store; use fabro_util::terminal::Styles; use fabro_workflows::operations::{ - RewindInput, RewindTarget, RunTimeline, build_timeline, find_run_id_by_prefix, rewind, + RewindInput, RewindTarget, RunTimeline, build_timeline_or_rebuild, + find_run_id_by_prefix_or_store, rewind, }; use git2::Repository; use crate::args::RewindArgs; +use crate::cli_config::load_cli_settings; use crate::shared::color_if; +use crate::store::{build_store, open_run_reader}; -pub(crate) fn run(args: &RewindArgs, styles: &Styles) -> Result<()> { +pub(crate) async fn run(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 cli_settings = load_cli_settings(None)?; + let durable_store = build_store(&cli_settings.storage_dir())?; + let run_id = + find_run_id_by_prefix_or_store(&repo, durable_store.as_ref(), &args.run_id).await?; let store = Store::new(repo); + let run_store = open_run_reader(&cli_settings.storage_dir(), &run_id).await?; - let timeline = build_timeline(&store, &run_id)?; + let timeline = build_timeline_or_rebuild(&store, run_store.as_deref(), &run_id).await?; if args.list || args.target.is_none() { print_timeline(&timeline, styles); diff --git a/lib/crates/fabro-workflows/src/operations/hydrate.rs b/lib/crates/fabro-workflows/src/operations/hydrate.rs index d08674a3f..e6f268582 100644 --- a/lib/crates/fabro-workflows/src/operations/hydrate.rs +++ b/lib/crates/fabro-workflows/src/operations/hydrate.rs @@ -56,7 +56,7 @@ pub async fn open_or_hydrate_run( } Ok(None) => {} Err(err) => { - tracing::warn!(error = %err, "Skipping malformed checkpoint.json during hydration") + tracing::warn!(error = %err, "Skipping malformed checkpoint.json during hydration"); } } match load_conclusion(run_dir) { @@ -68,7 +68,7 @@ pub async fn open_or_hydrate_run( } Ok(None) => {} Err(err) => { - tracing::warn!(error = %err, "Skipping malformed conclusion.json during hydration") + tracing::warn!(error = %err, "Skipping malformed conclusion.json during hydration"); } } match load_retro(run_dir) { @@ -84,7 +84,7 @@ pub async fn open_or_hydrate_run( } Ok(None) => {} Err(err) => { - tracing::warn!(error = %err, "Skipping malformed sandbox.json during hydration") + tracing::warn!(error = %err, "Skipping malformed sandbox.json during hydration"); } } diff --git a/lib/crates/fabro-workflows/src/operations/mod.rs b/lib/crates/fabro-workflows/src/operations/mod.rs index 621cb79ef..abaa993d2 100644 --- a/lib/crates/fabro-workflows/src/operations/mod.rs +++ b/lib/crates/fabro-workflows/src/operations/mod.rs @@ -1,6 +1,7 @@ mod create; mod fork; mod hydrate; +mod rebuild_meta; mod resume; mod rewind; mod source; @@ -13,6 +14,9 @@ pub use crate::pipeline::{DevcontainerSpec, LlmSpec, SandboxEnvSpec}; pub use create::{CreateRunInput, CreatedRun, create}; pub use fork::{ForkRunInput, fork}; pub use hydrate::open_or_hydrate_run; +pub use rebuild_meta::{ + build_timeline_or_rebuild, find_run_id_by_prefix_or_store, rebuild_metadata_branch, +}; pub use resume::resume; pub use rewind::{ RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline, find_run_id_by_prefix, diff --git a/lib/crates/fabro-workflows/src/operations/rebuild_meta.rs b/lib/crates/fabro-workflows/src/operations/rebuild_meta.rs new file mode 100644 index 000000000..991ae5b6c --- /dev/null +++ b/lib/crates/fabro-workflows/src/operations/rebuild_meta.rs @@ -0,0 +1,697 @@ +use std::collections::HashMap; +use std::fmt::Write; +use std::path::PathBuf; + +use anyhow::{Context, Result, bail}; +use fabro_git_storage::branchstore::BranchStore; +use fabro_git_storage::gitobj::Store as GitStore; +use fabro_store::{ + ListRunsQuery, NodeVisitRef, RunStore as DurableRunStore, Store as DurableStore, +}; +use git2::{Repository, Signature}; + +use super::rewind::{RunTimeline, build_timeline}; +use crate::git::MetadataStore; + +pub async fn rebuild_metadata_branch( + git_store: &GitStore, + run_store: &dyn DurableRunStore, + run_id: &str, +) -> Result<()> { + let branch = MetadataStore::branch_name(run_id); + if git_store.resolve_ref(&branch)?.is_some() { + bail!("metadata branch already exists for run {run_id}"); + } + + let run_record = run_store + .get_run() + .await? + .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 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? { + 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)?, + )); + + 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")?; + } + + 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(()) +} + +pub async fn build_timeline_or_rebuild( + git_store: &GitStore, + run_store: Option<&dyn DurableRunStore>, + run_id: &str, +) -> Result { + let branch = MetadataStore::branch_name(run_id); + if git_store.resolve_ref(&branch)?.is_some() { + return build_timeline(git_store, run_id); + } + + if let Some(run_store) = run_store { + rebuild_metadata_branch(git_store, run_store, run_id).await?; + return build_timeline(git_store, run_id); + } + + Ok(RunTimeline { + entries: Vec::new(), + parallel_map: HashMap::new(), + }) +} + +pub async fn find_run_id_by_prefix_or_store( + repo: &Repository, + fabro_store: &dyn DurableStore, + prefix: &str, +) -> Result { + if let Some(run_id) = find_run_id_by_prefix_in_refs(repo, prefix)? { + return Ok(run_id); + } + + let current_repo_root = canonical_repo_root(repo)?; + let mut matches = Vec::new(); + for summary in fabro_store.list_runs(&ListRunsQuery::default()).await? { + if summary.run_id == prefix { + if summary.host_repo_path.is_none() { + return Ok(summary.run_id); + } + + let Some(host_repo_path) = summary.host_repo_path.as_deref() else { + continue; + }; + let Ok(host_repo) = Repository::discover(host_repo_path) else { + continue; + }; + let Ok(host_repo_root) = canonical_repo_root(&host_repo) else { + continue; + }; + if host_repo_root == current_repo_root { + return Ok(summary.run_id); + } + continue; + } + + let Some(host_repo_path) = summary.host_repo_path.as_deref() else { + continue; + }; + let Ok(host_repo) = Repository::discover(host_repo_path) else { + continue; + }; + let Ok(host_repo_root) = canonical_repo_root(&host_repo) else { + continue; + }; + if host_repo_root == current_repo_root && summary.run_id.starts_with(prefix) { + matches.push(summary.run_id); + } + } + + resolve_prefix_matches(prefix, matches) +} + +fn write_entries( + branch_store: &BranchStore<'_>, + entries: &[(String, Vec)], + message: &str, +) -> Result<()> { + let refs: Vec<(&str, &[u8])> = entries + .iter() + .map(|(path, bytes)| (path.as_str(), bytes.as_slice())) + .collect(); + branch_store.write_entries(&refs, message)?; + Ok(()) +} + +fn node_file_path(node_id: &str, visit: u32, filename: &str) -> String { + if visit <= 1 { + format!("nodes/{node_id}/{filename}") + } else { + format!("nodes/{node_id}-visit_{visit}/{filename}") + } +} + +fn find_run_id_by_prefix_in_refs(repo: &Repository, prefix: &str) -> Result> { + let refs = repo.references()?; + let pattern = "refs/heads/fabro/meta/"; + let mut matches = Vec::new(); + + for reference in refs.flatten() { + let Some(name) = reference.name() else { + continue; + }; + let Some(run_id) = name.strip_prefix(pattern) else { + continue; + }; + + if run_id == prefix { + return Ok(Some(run_id.to_string())); + } + if run_id.starts_with(prefix) { + matches.push(run_id.to_string()); + } + } + + if matches.is_empty() { + return Ok(None); + } + + resolve_prefix_matches(prefix, matches).map(Some) +} + +fn canonical_repo_root(repo: &Repository) -> Result { + let root = repo + .workdir() + .or_else(|| repo.path().parent()) + .unwrap_or(repo.path()); + std::fs::canonicalize(root) + .with_context(|| format!("failed to canonicalize repo root {}", root.display())) +} + +fn resolve_prefix_matches(prefix: &str, matches: Vec) -> Result { + 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 run_id in &matches { + let _ = writeln!(msg, " {run_id}"); + } + bail!("{msg}") + } + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::path::PathBuf; + use std::sync::Arc; + + use chrono::{TimeZone, Utc}; + use fabro_config::FabroSettings; + use fabro_graphviz::graph::Graph; + use fabro_store::{InMemoryStore, Store as _}; + use fabro_types::{NodeStatusRecord, RunRecord, SandboxRecord, StageStatus, StartRecord}; + + use super::*; + use crate::operations::test_support::{make_checkpoint_json, temp_repo, test_sig}; + use crate::records::Checkpoint; + + fn created_at() -> chrono::DateTime { + Utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).unwrap() + } + + fn sample_run_record(run_id: &str, host_repo_path: Option<&str>) -> RunRecord { + RunRecord { + run_id: run_id.to_string(), + created_at: created_at(), + settings: FabroSettings::default(), + graph: Graph::new("test"), + workflow_slug: None, + working_directory: PathBuf::from("/tmp/project"), + host_repo_path: host_repo_path.map(ToOwned::to_owned), + base_branch: None, + labels: HashMap::new(), + } + } + + fn sample_start_record(run_id: &str) -> StartRecord { + StartRecord { + run_id: run_id.to_string(), + start_time: created_at(), + run_branch: Some(format!("fabro/run/{run_id}")), + base_sha: Some("base-sha".to_string()), + } + } + + fn sample_sandbox_record() -> SandboxRecord { + SandboxRecord { + provider: "local".to_string(), + working_directory: "/tmp/project".to_string(), + identifier: None, + host_working_directory: None, + container_mount_point: None, + data_host: None, + } + } + + fn sample_checkpoint( + current_node: &str, + completed_nodes: &[&str], + node_visits: &[(&str, usize)], + git_commit_sha: Option<&str>, + ) -> Checkpoint { + Checkpoint { + timestamp: created_at(), + 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(), + } + } + + fn sample_node_status() -> NodeStatusRecord { + NodeStatusRecord { + status: StageStatus::Success, + notes: Some("done".to_string()), + failure_reason: None, + timestamp: created_at(), + } + } + + async fn create_run_store( + store: &InMemoryStore, + run_id: &str, + host_repo_path: Option<&str>, + ) -> Arc { + let run_store = store.create_run(run_id, created_at(), None).await.unwrap(); + run_store + .put_run(&sample_run_record(run_id, host_repo_path)) + .await + .unwrap(); + run_store + } + + #[tokio::test] + async fn rebuild_metadata_branch_round_trips_timeline() { + 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 + .put_start(&sample_start_record("run-1")) + .await + .unwrap(); + run_store + .put_sandbox(&sample_sandbox_record()) + .await + .unwrap(); + + run_store + .append_checkpoint(&sample_checkpoint( + "start", + &["start"], + &[("start", 1)], + Some("aaa"), + )) + .await + .unwrap(); + run_store + .append_checkpoint(&sample_checkpoint( + "build", + &["start", "build"], + &[("start", 1), ("build", 1)], + Some("bbb"), + )) + .await + .unwrap(); + run_store + .append_checkpoint(&sample_checkpoint( + "build", + &["start", "build"], + &[("start", 1), ("build", 2)], + Some("ccc"), + )) + .await + .unwrap(); + + rebuild_metadata_branch(&git_store, run_store.as_ref(), "run-1") + .await + .unwrap(); + + let timeline = build_timeline(&git_store, "run-1").unwrap(); + assert_eq!(timeline.entries.len(), 3); + assert_eq!(timeline.entries[0].node_name, "start"); + assert_eq!(timeline.entries[0].visit, 1); + assert_eq!(timeline.entries[0].run_commit_sha.as_deref(), Some("aaa")); + assert_eq!(timeline.entries[1].node_name, "build"); + assert_eq!(timeline.entries[1].visit, 1); + assert_eq!(timeline.entries[1].run_commit_sha.as_deref(), Some("bbb")); + assert_eq!(timeline.entries[2].node_name, "build"); + assert_eq!(timeline.entries[2].visit, 2); + assert_eq!(timeline.entries[2].run_commit_sha.as_deref(), Some("ccc")); + } + + #[tokio::test] + async fn rebuild_metadata_branch_preserves_historical_node_visits() { + let (_dir, git_store) = temp_repo(); + let durable_store = InMemoryStore::default(); + let run_store = create_run_store(&durable_store, "run-1", None).await; + + let build_v1 = NodeVisitRef { + node_id: "build", + visit: 1, + }; + run_store + .put_node_prompt(&build_v1, "visit one") + .await + .unwrap(); + run_store + .put_node_status(&build_v1, &sample_node_status()) + .await + .unwrap(); + + let build_v2 = NodeVisitRef { + node_id: "build", + visit: 2, + }; + run_store + .put_node_prompt(&build_v2, "visit two") + .await + .unwrap(); + + run_store + .append_checkpoint(&sample_checkpoint( + "build", + &["build"], + &[("build", 1)], + Some("aaa"), + )) + .await + .unwrap(); + run_store + .append_checkpoint(&sample_checkpoint( + "build", + &["build"], + &[("build", 2)], + Some("bbb"), + )) + .await + .unwrap(); + + 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(); + + assert_eq!(checkpoint_commits.len(), 2); + assert_eq!( + git_store + .read_blob_at(checkpoint_commits[0], "nodes/build/prompt.md") + .unwrap() + .as_deref(), + Some("visit one".as_bytes()) + ); + assert!( + git_store + .read_blob_at(checkpoint_commits[0], "nodes/build-visit_2/prompt.md") + .unwrap() + .is_none() + ); + assert_eq!( + git_store + .read_blob_at(checkpoint_commits[1], "nodes/build/prompt.md") + .unwrap() + .as_deref(), + Some("visit one".as_bytes()) + ); + assert_eq!( + git_store + .read_blob_at(checkpoint_commits[1], "nodes/build-visit_2/prompt.md") + .unwrap() + .as_deref(), + Some("visit two".as_bytes()) + ); + } + + #[tokio::test] + async fn rebuild_metadata_branch_refuses_to_overwrite_existing_branch() { + let (_dir, git_store) = temp_repo(); + let durable_store = InMemoryStore::default(); + let run_store = create_run_store(&durable_store, "run-1", None).await; + + let sig = test_sig(); + let branch = MetadataStore::branch_name("run-1"); + let bs = BranchStore::new(&git_store, &branch, &sig); + bs.ensure_branch().unwrap(); + bs.write_entry("run.json", b"{}", "init run").unwrap(); + + let err = rebuild_metadata_branch(&git_store, run_store.as_ref(), "run-1") + .await + .unwrap_err(); + assert!(err.to_string().contains("metadata branch already exists")); + } + + #[tokio::test] + async fn build_timeline_or_rebuild_rebuilds_missing_branch() { + 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)], + Some("aaa"), + )) + .await + .unwrap(); + + let timeline = build_timeline_or_rebuild(&git_store, Some(run_store.as_ref()), "run-1") + .await + .unwrap(); + + assert_eq!(timeline.entries.len(), 1); + assert_eq!(timeline.entries[0].node_name, "start"); + } + + #[tokio::test] + async fn build_timeline_or_rebuild_preserves_existing_branch() { + 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)], + Some("aaa"), + )) + .await + .unwrap(); + run_store + .append_checkpoint(&sample_checkpoint( + "build", + &["start", "build"], + &[("start", 1), ("build", 1)], + Some("bbb"), + )) + .await + .unwrap(); + run_store + .append_checkpoint(&sample_checkpoint( + "test", + &["start", "build", "test"], + &[("start", 1), ("build", 1), ("test", 1)], + Some("ccc"), + )) + .await + .unwrap(); + + let sig = test_sig(); + let branch = MetadataStore::branch_name("run-1"); + let bs = BranchStore::new(&git_store, &branch, &sig); + bs.ensure_branch().unwrap(); + bs.write_entry("run.json", b"{}", "init run").unwrap(); + bs.write_entry( + "checkpoint.json", + &make_checkpoint_json("start", 1, Some("aaa")), + "checkpoint", + ) + .unwrap(); + bs.write_entry( + "checkpoint.json", + &make_checkpoint_json("build", 1, Some("bbb")), + "checkpoint", + ) + .unwrap(); + + let timeline = build_timeline_or_rebuild(&git_store, Some(run_store.as_ref()), "run-1") + .await + .unwrap(); + + assert_eq!(timeline.entries.len(), 2); + assert_eq!(timeline.entries[0].node_name, "start"); + assert_eq!(timeline.entries[1].node_name, "build"); + } + + #[tokio::test] + async fn build_timeline_or_rebuild_returns_empty_without_store() { + let (_dir, git_store) = temp_repo(); + let timeline = build_timeline_or_rebuild(&git_store, None, "run-1") + .await + .unwrap(); + assert!(timeline.entries.is_empty()); + assert!(timeline.parallel_map.is_empty()); + } + + #[tokio::test] + async fn rebuild_metadata_branch_errors_when_run_record_is_missing() { + let (_dir, git_store) = temp_repo(); + let durable_store = InMemoryStore::default(); + let run_store = durable_store + .create_run("run-1", created_at(), None) + .await + .unwrap(); + + let err = rebuild_metadata_branch(&git_store, run_store.as_ref(), "run-1") + .await + .unwrap_err(); + assert!(err.to_string().contains("run record not found")); + } + + #[tokio::test] + async fn find_run_id_by_prefix_or_store_falls_back_to_store() { + let (dir, git_store) = temp_repo(); + let durable_store = InMemoryStore::default(); + let repo_path = dir.path().to_string_lossy().to_string(); + let _run_store = create_run_store(&durable_store, "abc-123-long", Some(&repo_path)).await; + + let run_id = find_run_id_by_prefix_or_store(git_store.repo(), &durable_store, "abc-123") + .await + .unwrap(); + + assert_eq!(run_id, "abc-123-long"); + } + + #[tokio::test] + async fn find_run_id_by_prefix_or_store_excludes_other_repos() { + let (_dir, git_store) = temp_repo(); + let (other_dir, _other_git_store) = temp_repo(); + let durable_store = InMemoryStore::default(); + let other_repo_path = other_dir.path().to_string_lossy().to_string(); + let _run_store = + create_run_store(&durable_store, "abc-123-long", Some(&other_repo_path)).await; + + let err = find_run_id_by_prefix_or_store(git_store.repo(), &durable_store, "abc-123") + .await + .unwrap_err(); + + assert!(err.to_string().contains("no run found matching")); + } + + #[tokio::test] + async fn find_run_id_by_prefix_or_store_requires_exact_match_without_repo_path() { + let (_dir, git_store) = temp_repo(); + let durable_store = InMemoryStore::default(); + let _run_store = create_run_store(&durable_store, "abc-123-long", None).await; + + let prefix_err = + find_run_id_by_prefix_or_store(git_store.repo(), &durable_store, "abc-123") + .await + .unwrap_err(); + assert!(prefix_err.to_string().contains("no run found matching")); + + let exact = + find_run_id_by_prefix_or_store(git_store.repo(), &durable_store, "abc-123-long") + .await + .unwrap(); + assert_eq!(exact, "abc-123-long"); + } + + #[tokio::test] + async fn exact_match_wins_over_prefix_ambiguity() { + let (_dir, git_store) = temp_repo(); + let durable_store = InMemoryStore::default(); + let repo_path = git_store.repo_dir().to_string_lossy().to_string(); + let _short = create_run_store(&durable_store, "abc-123", Some(&repo_path)).await; + let _long = create_run_store(&durable_store, "abc-123-long", Some(&repo_path)).await; + + let from_store = + find_run_id_by_prefix_or_store(git_store.repo(), &durable_store, "abc-123") + .await + .unwrap(); + assert_eq!(from_store, "abc-123"); + + let sig = test_sig(); + let short_branch = + BranchStore::new(&git_store, MetadataStore::branch_name("abc-123"), &sig); + short_branch.ensure_branch().unwrap(); + + let long_branch = + BranchStore::new(&git_store, MetadataStore::branch_name("abc-123-long"), &sig); + long_branch.ensure_branch().unwrap(); + + let from_refs = find_run_id_by_prefix_or_store(git_store.repo(), &durable_store, "abc-123") + .await + .unwrap(); + assert_eq!(from_refs, "abc-123"); + } +}