From cfdb6248bbd376de1c7b3381e3b78931de11990e Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Wed, 1 Apr 2026 22:21:45 -0400 Subject: [PATCH] Persist command and diff artifacts in the run store --- lib/crates/fabro-cli/src/commands/run/diff.rs | 16 +++++ lib/crates/fabro-cli/tests/it/cmd/diff.rs | 44 ++++++++++++++ .../fabro-workflow/src/handler/command.rs | 52 ++++++++++++++++ .../fabro-workflow/src/handler/parallel.rs | 59 +++++++++++++++++++ .../fabro-workflow/src/lifecycle/git.rs | 26 ++++++++ 5 files changed, 197 insertions(+) diff --git a/lib/crates/fabro-cli/src/commands/run/diff.rs b/lib/crates/fabro-cli/src/commands/run/diff.rs index 8ce3e98ab..e9461744f 100644 --- a/lib/crates/fabro-cli/src/commands/run/diff.rs +++ b/lib/crates/fabro-cli/src/commands/run/diff.rs @@ -59,6 +59,22 @@ async fn resolve_diff( args: &DiffArgs, ) -> Result { if let Some(ref node_id) = args.node { + if let Some(run_store) = run_store { + if let Ok(visits) = run_store.list_node_visits(node_id).await { + if let Some(visit) = visits.into_iter().max() { + if let Ok(node) = run_store + .get_node(&fabro_store::NodeVisitRef { node_id, visit }) + .await + { + if let Some(patch) = node.diff { + debug!(node_id, visit, "Reading per-node diff from store"); + return Ok(patch); + } + } + } + } + } + debug!(node_id, "Reading per-node diff"); let node_patch = run_dir.join("nodes").join(node_id).join("diff.patch"); return std::fs::read_to_string(&node_patch).with_context(|| { diff --git a/lib/crates/fabro-cli/tests/it/cmd/diff.rs b/lib/crates/fabro-cli/tests/it/cmd/diff.rs index fc9840006..74567e26d 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/diff.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/diff.rs @@ -166,3 +166,47 @@ fn diff_node_outputs_specific_patch() { ----- stderr ----- "); } + +#[test] +fn diff_node_reads_store_patch_without_disk_file() { + let context = test_context!(); + let setup = setup_git_backed_changed_run(&context); + let run_id: RunId = setup.run.run_id.parse().unwrap(); + let patch = + std::fs::read_to_string(setup.run.run_dir.join("nodes/step_one/diff.patch")).unwrap(); + std::fs::remove_file(setup.run.run_dir.join("nodes/step_one/diff.patch")).unwrap(); + + with_runtime(|runtime| { + runtime.block_on(async { + let store = build_store(&context.storage_dir); + let run_store = store.open_run(&run_id).await.unwrap().unwrap(); + run_store + .put_node_diff( + &fabro_store::NodeVisitRef { + node_id: "step_one", + visit: 1, + }, + &patch, + ) + .await + .unwrap(); + }); + }); + + let mut cmd = context.command(); + cmd.args(["diff", &setup.run.run_id, "--node", "step_one"]); + + fabro_snapshot!(git_filters(&context), cmd, @" + success: true + exit_code: 0 + ----- stdout ----- + diff --git a/story.txt b/story.txt + index [SHA]..[SHA] 100644 + --- a/story.txt + +++ b/story.txt + @@ -1 +1,2 @@ + line 1 + +line 2 + ----- stderr ----- + "); +} diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs index 928c49d8c..0cae5fe6c 100644 --- a/lib/crates/fabro-workflow/src/handler/command.rs +++ b/lib/crates/fabro-workflow/src/handler/command.rs @@ -102,6 +102,12 @@ impl Handler for CommandHandler { "language": language, "timeout_ms": timeout_ms(node), }); + if let Some(ref store) = services.run_store { + store + .put_node_script_invocation(&node_ref, &invocation) + .await + .map_err(|err| FabroError::handler(err.to_string()))?; + } fs::write( stage_dir.join("script_invocation.json"), serde_json::to_string_pretty(&invocation).unwrap(), @@ -155,6 +161,12 @@ impl Handler for CommandHandler { "exit_code": if result.timed_out { serde_json::Value::Null } else { serde_json::json!(result.exit_code) }, "timed_out": result.timed_out, }); + if let Some(ref store) = services.run_store { + store + .put_node_script_timing(&node_ref, &timing) + .await + .map_err(|err| FabroError::handler(err.to_string()))?; + } fs::write( stage_dir.join("script_timing.json"), serde_json::to_string_pretty(&timing).unwrap(), @@ -217,6 +229,8 @@ mod tests { use super::*; use crate::outcome::StageStatus; use fabro_graphviz::graph::AttrValue; + use fabro_store::{InMemoryStore, NodeVisitRef, RunStore, Store}; + use std::sync::Arc; use std::time::Duration; fn make_services() -> EngineServices { @@ -561,6 +575,44 @@ mod tests { assert_eq!(json["timed_out"], true); } + #[tokio::test] + async fn stores_script_invocation_and_timing_in_run_store() { + let handler = CommandHandler; + let mut node = Node::new("script_node"); + node.attrs.insert( + "script".to_string(), + AttrValue::String("echo hello".to_string()), + ); + let context = Context::new(); + let graph = Graph::new("test"); + let run_dir = tempfile::tempdir().unwrap(); + let store = Arc::new(InMemoryStore::default()); + let run_store = store + .create_run(&fabro_types::fixtures::RUN_1, chrono::Utc::now(), None) + .await + .unwrap(); + let services = EngineServices { + run_store: Some(Arc::clone(&run_store) as Arc), + ..EngineServices::test_default() + }; + + handler + .execute(&node, &context, &graph, run_dir.path(), &services) + .await + .unwrap(); + + let snapshot = run_store + .get_node(&NodeVisitRef { + node_id: "script_node", + visit: 1, + }) + .await + .unwrap(); + + assert_eq!(snapshot.script_invocation.unwrap()["command"], "echo hello"); + assert_eq!(snapshot.script_timing.unwrap()["exit_code"], 0); + } + #[tokio::test] async fn script_handler_python_echo() { let handler = CommandHandler; diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 2d4102ca9..176df2bc8 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -4,6 +4,7 @@ use std::time::Instant; use async_trait::async_trait; use fabro_agent::{Sandbox, WorktreeConfig, WorktreeSandbox}; +use fabro_store::NodeVisitRef; use fabro_types::RunId; use tokio::sync::Semaphore; @@ -480,6 +481,16 @@ impl Handler for ParallelHandler { let visit = visit_from_context(context); let node_dir = node_dir(run_dir, &node.id, visit); let _ = fs::create_dir_all(&node_dir).await; + if let Some(ref store) = services.run_store { + let node_ref = NodeVisitRef { + node_id: &node.id, + visit: u32::try_from(visit).unwrap_or(u32::MAX), + }; + store + .put_node_parallel_results(&node_ref, &serde_json::json!(results_json)) + .await + .map_err(|err| FabroError::handler(err.to_string()))?; + } if let Ok(json) = serde_json::to_string_pretty(&results_json) { let _ = fs::write(node_dir.join("parallel_results.json"), json).await; } @@ -583,7 +594,9 @@ fn find_join_node(results: &[BranchResult], graph: &Graph) -> Option { mod tests { use super::*; use fabro_graphviz::graph::{AttrValue, Edge}; + use fabro_store::{InMemoryStore, RunStore, Store}; use fabro_types::fixtures; + use std::sync::Arc; fn make_services() -> EngineServices { EngineServices::test_default() @@ -665,6 +678,52 @@ mod tests { assert_eq!(parsed.as_array().unwrap().len(), 2); } + #[tokio::test] + async fn parallel_handler_stores_results_in_run_store() { + let store = Arc::new(InMemoryStore::default()); + let run_store = store + .create_run(&fixtures::RUN_1, chrono::Utc::now(), None) + .await + .unwrap(); + let services = EngineServices { + run_store: Some(Arc::clone(&run_store) as Arc), + ..EngineServices::test_default() + }; + let mut node = Node::new("par"); + node.attrs.insert( + "shape".to_string(), + AttrValue::String("component".to_string()), + ); + let context = test_context(); + let mut graph = Graph::new("test"); + graph.nodes.insert("par".to_string(), node.clone()); + graph + .nodes + .insert("branch_a".to_string(), Node::new("branch_a")); + graph + .nodes + .insert("branch_b".to_string(), Node::new("branch_b")); + graph.edges.push(Edge::new("par", "branch_a")); + graph.edges.push(Edge::new("par", "branch_b")); + + let tmp = tempfile::tempdir().unwrap(); + ParallelHandler + .execute(&node, &context, &graph, tmp.path(), &services) + .await + .unwrap(); + + let snapshot = run_store + .get_node(&NodeVisitRef { + node_id: "par", + visit: 1, + }) + .await + .unwrap(); + let results = snapshot.parallel_results.unwrap(); + assert!(results.is_array()); + assert_eq!(results.as_array().unwrap().len(), 2); + } + #[tokio::test] async fn parallel_handler_first_success_policy() { let services = make_services(); diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index 5938b6cc2..ca31f267f 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -292,6 +292,22 @@ impl RunLifecycle for GitLifecycle { match git_diff(&*self.sandbox, &prev).await { Ok(patch) if !patch.is_empty() => { + let node_ref = fabro_store::NodeVisitRef { + node_id, + visit: u32::try_from(visit).unwrap_or(u32::MAX), + }; + if let Err(err) = self.run_store.put_node_diff(&node_ref, &patch).await { + self.emitter.emit(&WorkflowRunEvent::RunNotice { + level: RunNoticeLevel::Warn, + code: "git_diff_store_failed".to_string(), + message: format!( + "[node: {node_id}] failed to persist diff in run store: {err}" + ), + }); + return Err(CoreError::Other(format!( + "failed to persist node diff for '{node_id}': {err}" + ))); + } let _ = std::fs::write(&diff_dest, &patch); git_result.diff = Some(patch); } @@ -338,6 +354,16 @@ impl RunLifecycle for GitLifecycle { let diff_dest = self.run_dir.join("final.patch"); match git_diff(&*self.sandbox, &base_sha).await { Ok(patch) if !patch.is_empty() => { + if let Err(err) = self.run_store.put_final_patch(&patch).await { + self.emitter.emit(&WorkflowRunEvent::RunNotice { + level: RunNoticeLevel::Warn, + code: "final_diff_store_failed".to_string(), + message: format!( + "failed to persist final diff in run store: {err}" + ), + }); + return; + } let _ = std::fs::write(&diff_dest, patch); } Ok(_) => {}