scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-05 08:10:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Simplify MetadataStore init_run API and resume graph loading

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-24 19:30:18 -04:00
parent 2aa6476437
commit 16bfb84b40
3 changed files with 54 additions and 82 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
lib/crates/fabro-cli/src/commands/resume.rs
View file

@ -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::<SandboxProvider>().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!(

10
lib/crates/fabro-workflows/src/core_adapter/lifecycle/git.rs
View file

@ -59,17 +59,17 @@ impl RunLifecycle<WorkflowGraph> 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,

81
lib/crates/fabro-workflows/src/git.rs
View file

@ -366,7 +366,7 @@ pub fn scan_node_files(run_dir: &Path) -> Vec<(String, Vec<u8>)> {
/// 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")