From 16bfb84b404a037ef5d8d4fa395a59ea32f5c8cf Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 24 Mar 2026 19:30:18 -0400 Subject: [PATCH] Simplify MetadataStore init_run API and resume graph loading Collapse init_run/init_run_with_records/init_run_inner into a single init_run(run_id, files) that takes all files as a flat slice. Resume from metadata branch now uses RunRecord's embedded graph directly when available, falling back to graph.fabro DOT parsing for old runs. Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/crates/fabro-cli/src/commands/resume.rs | 45 ++++++++--- .../src/core_adapter/lifecycle/git.rs | 10 +-- lib/crates/fabro-workflows/src/git.rs | 81 ++++--------------- 3 files changed, 54 insertions(+), 82 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/resume.rs b/lib/crates/fabro-cli/src/commands/resume.rs index 1421c50ff..8bf300f6c 100644 --- a/lib/crates/fabro-cli/src/commands/resume.rs +++ b/lib/crates/fabro-cli/src/commands/resume.rs @@ -553,11 +553,6 @@ async fn prepare_from_branch( .ok_or_else(|| { anyhow::anyhow!("no checkpoint found on metadata branch for run {run_id}") })?; - let source = fabro_workflows::git::MetadataStore::read_graph_dot(&resume_repo_path, &run_id)? - .ok_or_else(|| { - anyhow::anyhow!("no graph.fabro found on metadata branch for run {run_id}") - })?; - let repo_info = fabro_sandbox::daytona::detect_repo_info(&resume_repo_path).ok(); let origin_url = repo_info.as_ref().map(|(url, _)| url.clone()); let detected_base_branch = record @@ -585,7 +580,39 @@ async fn prepare_from_branch( prepared.workflow_slug, ) } + } else if let Some(ref rec) = record { + // Use the fully transformed graph from the RunRecord + let graph = rec.graph.clone(); + let source = String::new(); // no DOT source needed — graph is from RunRecord + let run_cfg = Some(rec.config.clone()); + let sandbox_provider = if args.dry_run { + SandboxProvider::Local + } else { + let sp = rec + .config + .sandbox + .as_ref() + .and_then(|s| s.provider.as_deref()) + .and_then(|s| s.parse::().ok()) + .unwrap_or_default(); + args.sandbox.map(Into::into).unwrap_or(sp) + }; + ( + graph, + source, + run_cfg, + sandbox_provider, + rec.workflow_slug.clone(), + ) } else { + // Fallback: read DOT source from metadata branch + let source = + fabro_workflows::git::MetadataStore::read_graph_dot(&resume_repo_path, &run_id)? + .ok_or_else(|| { + anyhow::anyhow!( + "no run.json or graph.fabro found on metadata branch for run {run_id}" + ) + })?; let (graph, diagnostics) = fabro_workflows::workflow::WorkflowBuilder::new().prepare(&source)?; print_diagnostics(&diagnostics, styles); @@ -598,13 +625,7 @@ async fn prepare_from_branch( } else { resolve_sandbox_provider(args.sandbox.map(Into::into), None, run_defaults)? }; - ( - graph, - source.clone(), - None, - sandbox_provider, - record.as_ref().and_then(|r| r.workflow_slug.clone()), - ) + (graph, source.clone(), None, sandbox_provider, None) }; eprintln!( diff --git a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs index b1bcbf61c..cc3425e60 100644 --- a/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs +++ b/lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs @@ -59,17 +59,17 @@ impl RunLifecycle for GitLifecycle { let run_json = std::fs::read(self.run_dir.join("run.json")).ok(); let start_json = std::fs::read(self.run_dir.join("start.json")).ok(); let sandbox_json = std::fs::read(self.run_dir.join("sandbox.json")).ok(); - let mut extra_files: Vec<(&str, &[u8])> = Vec::new(); + let mut files: Vec<(&str, &[u8])> = Vec::new(); if let Some(ref data) = run_json { - extra_files.push(("run.json", data)); + files.push(("run.json", data)); } if let Some(ref data) = start_json { - extra_files.push(("start.json", data)); + files.push(("start.json", data)); } if let Some(ref data) = sandbox_json { - extra_files.push(("sandbox.json", data)); + files.push(("sandbox.json", data)); } - if let Err(e) = store.init_run(&self.run_id, &[], &[], &extra_files) { + if let Err(e) = store.init_run(&self.run_id, &files) { tracing::warn!( run_id = %self.run_id, error = %e, diff --git a/lib/crates/fabro-workflows/src/git.rs b/lib/crates/fabro-workflows/src/git.rs index 550a985f6..5e464875f 100644 --- a/lib/crates/fabro-workflows/src/git.rs +++ b/lib/crates/fabro-workflows/src/git.rs @@ -366,7 +366,7 @@ pub fn scan_node_files(run_dir: &Path) -> Vec<(String, Vec)> { /// Git-native metadata storage for pipeline runs. /// -/// Stores checkpoint data, manifests, and graph DOT on an orphan branch +/// Stores checkpoint data, run records, and metadata on an orphan branch /// (`fabro/meta/{run_id}`) so that runs can be resumed from git alone. pub struct MetadataStore { repo_path: std::path::PathBuf, @@ -402,63 +402,18 @@ impl MetadataStore { Ok((store, sig)) } - /// Initialize a run's metadata branch with manifest, graph DOT, and optional extra files. - pub fn init_run( - &self, - run_id: &str, - manifest_json: &[u8], - graph_dot: &[u8], - extra_files: &[(&str, &[u8])], - ) -> Result<()> { - self.init_run_inner(run_id, manifest_json, graph_dot, None, None, extra_files) - } - - /// Initialize a run's metadata branch with manifest, graph DOT, run record, start record, - /// and optional extra files. - pub fn init_run_with_records( - &self, - run_id: &str, - manifest_json: &[u8], - graph_dot: &[u8], - run_record_json: &[u8], - start_record_json: &[u8], - extra_files: &[(&str, &[u8])], - ) -> Result<()> { - self.init_run_inner( - run_id, - manifest_json, - graph_dot, - Some(run_record_json), - Some(start_record_json), - extra_files, - ) - } - - fn init_run_inner( - &self, - run_id: &str, - manifest_json: &[u8], - graph_dot: &[u8], - run_record_json: Option<&[u8]>, - start_record_json: Option<&[u8]>, - extra_files: &[(&str, &[u8])], - ) -> Result<()> { + /// Initialize a run's metadata branch with the given files. + /// + /// Callers pass all files (run.json, start.json, sandbox.json, etc.) + /// via the `files` slice. + pub fn init_run(&self, run_id: &str, files: &[(&str, &[u8])]) -> Result<()> { let (store, sig) = self.open_store()?; let branch = Self::branch_name(run_id); let bs = BranchStore::new(&store, &branch, &sig); bs.ensure_branch() .map_err(|e| git_error(format!("ensure_branch failed: {e}")))?; - let mut entries: Vec<(&str, &[u8])> = - vec![("manifest.json", manifest_json), ("graph.fabro", graph_dot)]; - if let Some(rr) = run_record_json { - entries.push(("run.json", rr)); - } - if let Some(sr) = start_record_json { - entries.push(("start.json", sr)); - } - entries.extend_from_slice(extra_files); let msg = self.commit_message("init run"); - bs.write_entries(&entries, &msg) + bs.write_entries(files, &msg) .map_err(|e| git_error(format!("write_entries failed: {e}")))?; Ok(()) } @@ -674,7 +629,7 @@ mod tests { let run_record = br#"{"run_id":"RUN1","created_at":"2025-01-01T00:00:00Z","config":{},"graph":{"name":"test","nodes":{},"edges":[],"attrs":{}},"working_directory":"/tmp"}"#; let dot = b"digraph { start -> end }"; store - .init_run("RUN1", &[], dot, &[("run.json", run_record)]) + .init_run("RUN1", &[("run.json", run_record), ("graph.fabro", dot)]) .unwrap(); let read_record = MetadataStore::read_run_record(dir.path(), "RUN1") @@ -695,7 +650,7 @@ mod tests { init_repo(dir.path()); let store = MetadataStore::new(dir.path(), &GitAuthor::default()); - store.init_run("RUN2", b"{}", b"digraph {}", &[]).unwrap(); + store.init_run("RUN2", &[]).unwrap(); let ctx = crate::context::Context::new(); ctx.set("goal", serde_json::json!("test")); @@ -731,7 +686,7 @@ mod tests { init_repo(dir.path()); let store = MetadataStore::new(dir.path(), &GitAuthor::default()); - store.init_run("RUN3", b"{}", b"digraph {}", &[]).unwrap(); + store.init_run("RUN3", &[]).unwrap(); let ctx = crate::context::Context::new(); let cp1 = crate::checkpoint::Checkpoint::from_context( @@ -784,7 +739,7 @@ mod tests { init_repo(dir.path()); let store = MetadataStore::new(dir.path(), &GitAuthor::default()); - store.init_run("RUN4", b"{}", b"digraph {}", &[]).unwrap(); + store.init_run("RUN4", &[]).unwrap(); let artifact_data = br#"{"large_output":"some data"}"#; let cp_json = b"{}"; // minimal checkpoint for the test @@ -859,7 +814,8 @@ mod tests { init_repo(dir.path()); let store = MetadataStore::new(dir.path(), &GitAuthor::default()); - store.init_run("RUN5", b"{}", b"digraph {}", &[]).unwrap(); + let run_record = br#"{"run_id":"RUN5","created_at":"2025-01-01T00:00:00Z","config":{},"graph":{"name":"test","nodes":{},"edges":[],"attrs":{}},"working_directory":"/tmp"}"#; + store.init_run("RUN5", &[("run.json", run_record)]).unwrap(); store .write_files( @@ -875,10 +831,10 @@ mod tests { assert_eq!(data, b"{\"status\":\"ok\"}"); // Original files still present - let dot = MetadataStore::read_graph_dot(dir.path(), "RUN5") + let record = MetadataStore::read_run_record(dir.path(), "RUN5") .unwrap() .unwrap(); - assert_eq!(dot, "digraph {}"); + assert_eq!(record.run_id, "RUN5"); } #[test] @@ -888,12 +844,7 @@ mod tests { let store = MetadataStore::new(dir.path(), &GitAuthor::default()); store - .init_run( - "RUN6", - b"{}", - b"digraph {}", - &[("sandbox.json", b"{\"type\":\"local\"}")], - ) + .init_run("RUN6", &[("sandbox.json", b"{\"type\":\"local\"}")]) .unwrap(); let data = MetadataStore::read_file(dir.path(), "RUN6", "sandbox.json")