Persist command and diff artifacts in the run store

lib/crates/fabro-cli/src/commands/run/diff.rs

@@ -59,6 +59,22 @@ async fn resolve_diff(
     args: &DiffArgs,
 ) -> Result<String> {
     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(|| {

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 -----
+    ");
+}

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<dyn RunStore>),
+            ..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;

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<String> {
 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<dyn RunStore>),
+            ..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();

lib/crates/fabro-workflow/src/lifecycle/git.rs

@@ -292,6 +292,22 @@ impl RunLifecycle<WorkflowGraph> 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<WorkflowGraph> 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(_) => {}