Add disk-projecting run store decorator

lib/crates/fabro-retro/src/retro_agent.rs

@@ -196,8 +196,7 @@ pub async fn run_retro_agent(
          rather than reading the entire file. When done, call the `submit_retro` tool with your analysis."
     );
 
-    // Write prompt.md
-    let _ = std::fs::write(retro_dir.join("prompt.md"), &prompt);
+    write_retro_prompt(run_store, &retro_dir, &prompt).await?;
 
     let process_result = session
         .process_input(&prompt)
@@ -239,9 +238,9 @@ pub async fn run_retro_agent(
     };
 
     // Write artifacts (on both success and failure)
+    write_retro_response(run_store, &retro_dir, response_text).await?;
     write_retro_artifacts(
         &retro_dir,
-        response_text,
         provider.as_str(),
         model,
         outcome,
@@ -271,18 +270,47 @@ pub fn dry_run_narrative() -> RetroNarrative {
     }
 }
 
-/// Write retro artifact files (response.md, provider_used.json, status.json) into `retro_dir`.
+async fn write_retro_prompt(
+    run_store: Option<&dyn RunStore>,
+    retro_dir: &Path,
+    prompt: &str,
+) -> anyhow::Result<()> {
+    if let Some(store) = run_store {
+        store
+            .put_retro_prompt(prompt)
+            .await
+            .map_err(|e| anyhow::anyhow!("Failed to save retro prompt to store: {e}"))?;
+    } else {
+        std::fs::write(retro_dir.join("prompt.md"), prompt)?;
+    }
+    Ok(())
+}
+
+async fn write_retro_response(
+    run_store: Option<&dyn RunStore>,
+    retro_dir: &Path,
+    response: &str,
+) -> anyhow::Result<()> {
+    if let Some(store) = run_store {
+        store
+            .put_retro_response(response)
+            .await
+            .map_err(|e| anyhow::anyhow!("Failed to save retro response to store: {e}"))?;
+    } else {
+        std::fs::write(retro_dir.join("response.md"), response)?;
+    }
+    Ok(())
+}
+
+/// Write retro artifact files (provider_used.json, status.json) into `retro_dir`.
 /// Called on both success and failure paths so artifacts are always available for debugging.
 fn write_retro_artifacts(
     retro_dir: &Path,
-    response: &str,
     provider: &str,
     model: &str,
     outcome: &str,
     failure_reason: Option<&str>,
 ) {
-    let _ = std::fs::write(retro_dir.join("response.md"), response);
-
     let provider_used = serde_json::json!({
         "mode": "agent",
         "provider": provider,
@@ -546,7 +574,6 @@ mod tests {
         std::fs::write(retro_dir.join("prompt.md"), "Analyze the run data").unwrap();
         write_retro_artifacts(
             &retro_dir,
-            "response text",
             "anthropic",
             "claude-sonnet-4-20250514",
             "success",
@@ -556,29 +583,12 @@ mod tests {
         assert_eq!(content, "Analyze the run data");
     }
 
-    #[test]
-    fn writes_response_md() {
-        let dir = tempfile::tempdir().unwrap();
-        let retro_dir = dir.path().join("retro");
-        std::fs::create_dir_all(&retro_dir).unwrap();
-        write_retro_artifacts(
-            &retro_dir,
-            "The run completed successfully with minor issues.",
-            "anthropic",
-            "claude-sonnet-4-20250514",
-            "success",
-            None,
-        );
-        let content = std::fs::read_to_string(retro_dir.join("response.md")).unwrap();
-        assert_eq!(content, "The run completed successfully with minor issues.");
-    }
-
     #[test]
     fn writes_provider_used_json() {
         let dir = tempfile::tempdir().unwrap();
         let retro_dir = dir.path().join("retro");
         std::fs::create_dir_all(&retro_dir).unwrap();
-        write_retro_artifacts(&retro_dir, "resp", "openai", "gpt-4o", "success", None);
+        write_retro_artifacts(&retro_dir, "openai", "gpt-4o", "success", None);
         let content = std::fs::read_to_string(retro_dir.join("provider_used.json")).unwrap();
         let parsed: serde_json::Value = serde_json::from_str(&content).unwrap();
         assert_eq!(parsed["mode"], "agent");
@@ -593,7 +603,6 @@ mod tests {
         std::fs::create_dir_all(&retro_dir).unwrap();
         write_retro_artifacts(
             &retro_dir,
-            "resp",
             "anthropic",
             "claude-sonnet-4-20250514",
             "success",
@@ -613,7 +622,6 @@ mod tests {
         std::fs::create_dir_all(&retro_dir).unwrap();
         write_retro_artifacts(
             &retro_dir,
-            "resp",
             "anthropic",
             "claude-sonnet-4-20250514",
             "error",

lib/crates/fabro-store/src/disk_projecting.rs

@@ -0,0 +1,777 @@
+use std::fs::{self, OpenOptions};
+use std::io::Write;
+use std::path::{Path, PathBuf};
+use std::pin::Pin;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use futures::Stream;
+use tracing::warn;
+
+use crate::{
+    EventEnvelope, EventPayload, NodeSnapshot, NodeVisitRef, Result, RunSnapshot, RunStore,
+};
+use fabro_types::{
+    Checkpoint, Conclusion, NodeStatusRecord, Retro, RunRecord, RunStatusRecord, SandboxRecord,
+    StartRecord,
+};
+
+pub struct DiskProjectingRunStore {
+    inner: Arc<dyn RunStore>,
+    run_dir: PathBuf,
+}
+
+impl DiskProjectingRunStore {
+    #[must_use]
+    pub fn new(inner: Arc<dyn RunStore>, run_dir: PathBuf) -> Self {
+        Self { inner, run_dir }
+    }
+
+    fn warn_projection(path: &Path, err: &std::io::Error, critical: bool) {
+        if critical {
+            warn!(
+                path = %path.display(),
+                error = %err,
+                "Critical disk projection failed"
+            );
+        } else {
+            warn!(path = %path.display(), error = %err, "Disk projection failed");
+        }
+    }
+
+    fn write_json_critical<T: serde::Serialize>(path: &Path, value: &T) {
+        if let Err(err) = write_json(path, value) {
+            Self::warn_projection(path, &err, true);
+        }
+    }
+
+    fn write_json_best_effort<T: serde::Serialize>(path: &Path, value: &T) {
+        if let Err(err) = write_json(path, value) {
+            Self::warn_projection(path, &err, false);
+        }
+    }
+
+    fn write_text_best_effort(path: &Path, value: &str) {
+        if let Err(err) = write_text(path, value) {
+            Self::warn_projection(path, &err, false);
+        }
+    }
+
+    fn append_jsonl_critical(&self, payload: &EventPayload) {
+        let progress_path = self.run_dir.join("progress.jsonl");
+        if let Err(err) = append_jsonl(&progress_path, payload) {
+            Self::warn_projection(&progress_path, &err, true);
+        }
+
+        let live_path = self.run_dir.join("live.json");
+        if let Err(err) = write_live_json(&live_path, payload) {
+            Self::warn_projection(&live_path, &err, true);
+        }
+    }
+}
+
+/// Map store node visits onto the legacy on-disk layout used by workflow logs.
+///
+/// The store key layout uses `nodes/{id}/visit-{N}/...`, but existing disk readers
+/// expect first visits at `nodes/{id}/...` and later visits at
+/// `nodes/{id}-visit_{N}/...`.
+fn disk_node_dir(run_dir: &Path, node_id: &str, visit: u32) -> PathBuf {
+    if visit <= 1 {
+        run_dir.join("nodes").join(node_id)
+    } else {
+        run_dir
+            .join("nodes")
+            .join(format!("{node_id}-visit_{visit}"))
+    }
+}
+
+fn ensure_parent_dir(path: &Path) -> std::io::Result<()> {
+    if let Some(parent) = path.parent() {
+        fs::create_dir_all(parent)?;
+    }
+    Ok(())
+}
+
+fn write_json<T: serde::Serialize>(path: &Path, value: &T) -> std::io::Result<()> {
+    ensure_parent_dir(path)?;
+    let json = serde_json::to_string_pretty(value).map_err(std::io::Error::other)?;
+    fs::write(path, json)
+}
+
+fn write_text(path: &Path, value: &str) -> std::io::Result<()> {
+    ensure_parent_dir(path)?;
+    fs::write(path, value)
+}
+
+fn append_jsonl(path: &Path, payload: &EventPayload) -> std::io::Result<()> {
+    ensure_parent_dir(path)?;
+    let line = serde_json::to_string(payload.as_value()).map_err(std::io::Error::other)?;
+    let mut file = OpenOptions::new().create(true).append(true).open(path)?;
+    writeln!(file, "{line}")
+}
+
+fn write_live_json(path: &Path, payload: &EventPayload) -> std::io::Result<()> {
+    ensure_parent_dir(path)?;
+    let json = serde_json::to_string_pretty(payload.as_value()).map_err(std::io::Error::other)?;
+    fs::write(path, json)
+}
+
+#[async_trait]
+impl RunStore for DiskProjectingRunStore {
+    async fn put_run(&self, record: &RunRecord) -> Result<()> {
+        self.inner.put_run(record).await?;
+        Self::write_json_best_effort(&self.run_dir.join("run.json"), record);
+        Ok(())
+    }
+
+    async fn get_run(&self) -> Result<Option<RunRecord>> {
+        self.inner.get_run().await
+    }
+
+    async fn put_start(&self, record: &StartRecord) -> Result<()> {
+        self.inner.put_start(record).await?;
+        Self::write_json_best_effort(&self.run_dir.join("start.json"), record);
+        Ok(())
+    }
+
+    async fn get_start(&self) -> Result<Option<StartRecord>> {
+        self.inner.get_start().await
+    }
+
+    async fn put_status(&self, record: &RunStatusRecord) -> Result<()> {
+        Self::write_json_critical(&self.run_dir.join("status.json"), record);
+        self.inner.put_status(record).await
+    }
+
+    async fn get_status(&self) -> Result<Option<RunStatusRecord>> {
+        self.inner.get_status().await
+    }
+
+    async fn put_checkpoint(&self, record: &Checkpoint) -> Result<()> {
+        self.inner.put_checkpoint(record).await?;
+        Self::write_json_best_effort(&self.run_dir.join("checkpoint.json"), record);
+        Ok(())
+    }
+
+    async fn get_checkpoint(&self) -> Result<Option<Checkpoint>> {
+        self.inner.get_checkpoint().await
+    }
+
+    async fn append_checkpoint(&self, record: &Checkpoint) -> Result<u32> {
+        self.inner.append_checkpoint(record).await
+    }
+
+    async fn list_checkpoints(&self) -> Result<Vec<(u32, Checkpoint)>> {
+        self.inner.list_checkpoints().await
+    }
+
+    async fn put_conclusion(&self, record: &Conclusion) -> Result<()> {
+        Self::write_json_critical(&self.run_dir.join("conclusion.json"), record);
+        self.inner.put_conclusion(record).await
+    }
+
+    async fn get_conclusion(&self) -> Result<Option<Conclusion>> {
+        self.inner.get_conclusion().await
+    }
+
+    async fn put_retro(&self, retro: &Retro) -> Result<()> {
+        self.inner.put_retro(retro).await?;
+        Self::write_json_best_effort(&self.run_dir.join("retro.json"), retro);
+        Ok(())
+    }
+
+    async fn get_retro(&self) -> Result<Option<Retro>> {
+        self.inner.get_retro().await
+    }
+
+    async fn put_graph(&self, dot_source: &str) -> Result<()> {
+        self.inner.put_graph(dot_source).await?;
+        Self::write_text_best_effort(&self.run_dir.join("workflow.fabro"), dot_source);
+        Ok(())
+    }
+
+    async fn get_graph(&self) -> Result<Option<String>> {
+        self.inner.get_graph().await
+    }
+
+    async fn put_sandbox(&self, record: &SandboxRecord) -> Result<()> {
+        self.inner.put_sandbox(record).await?;
+        Self::write_json_best_effort(&self.run_dir.join("sandbox.json"), record);
+        Ok(())
+    }
+
+    async fn get_sandbox(&self) -> Result<Option<SandboxRecord>> {
+        self.inner.get_sandbox().await
+    }
+
+    async fn put_node_prompt(&self, node: &NodeVisitRef<'_>, prompt: &str) -> Result<()> {
+        self.inner.put_node_prompt(node, prompt).await?;
+        Self::write_text_best_effort(
+            &disk_node_dir(&self.run_dir, node.node_id, node.visit).join("prompt.md"),
+            prompt,
+        );
+        Ok(())
+    }
+
+    async fn put_node_response(&self, node: &NodeVisitRef<'_>, response: &str) -> Result<()> {
+        self.inner.put_node_response(node, response).await?;
+        Self::write_text_best_effort(
+            &disk_node_dir(&self.run_dir, node.node_id, node.visit).join("response.md"),
+            response,
+        );
+        Ok(())
+    }
+
+    async fn put_node_status(
+        &self,
+        node: &NodeVisitRef<'_>,
+        status: &NodeStatusRecord,
+    ) -> Result<()> {
+        self.inner.put_node_status(node, status).await?;
+        Self::write_json_best_effort(
+            &disk_node_dir(&self.run_dir, node.node_id, node.visit).join("status.json"),
+            status,
+        );
+        Ok(())
+    }
+
+    async fn put_node_stdout(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> {
+        self.inner.put_node_stdout(node, log).await?;
+        Self::write_text_best_effort(
+            &disk_node_dir(&self.run_dir, node.node_id, node.visit).join("stdout.log"),
+            log,
+        );
+        Ok(())
+    }
+
+    async fn put_node_stderr(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> {
+        self.inner.put_node_stderr(node, log).await?;
+        Self::write_text_best_effort(
+            &disk_node_dir(&self.run_dir, node.node_id, node.visit).join("stderr.log"),
+            log,
+        );
+        Ok(())
+    }
+
+    async fn get_node(&self, node: &NodeVisitRef<'_>) -> Result<NodeSnapshot> {
+        self.inner.get_node(node).await
+    }
+
+    async fn list_node_visits(&self, node_id: &str) -> Result<Vec<u32>> {
+        self.inner.list_node_visits(node_id).await
+    }
+
+    async fn append_event(&self, payload: &EventPayload) -> Result<u32> {
+        self.append_jsonl_critical(payload);
+        self.inner.append_event(payload).await
+    }
+
+    async fn list_events(&self) -> Result<Vec<EventEnvelope>> {
+        self.inner.list_events().await
+    }
+
+    async fn list_events_from(&self, seq: u32) -> Result<Vec<EventEnvelope>> {
+        self.inner.list_events_from(seq).await
+    }
+
+    async fn watch_events_from(
+        &self,
+        seq: u32,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<EventEnvelope>> + Send>>> {
+        self.inner.watch_events_from(seq).await
+    }
+
+    async fn put_retro_prompt(&self, text: &str) -> Result<()> {
+        self.inner.put_retro_prompt(text).await?;
+        Self::write_text_best_effort(&self.run_dir.join("retro").join("prompt.md"), text);
+        Ok(())
+    }
+
+    async fn get_retro_prompt(&self) -> Result<Option<String>> {
+        self.inner.get_retro_prompt().await
+    }
+
+    async fn put_retro_response(&self, text: &str) -> Result<()> {
+        self.inner.put_retro_response(text).await?;
+        Self::write_text_best_effort(&self.run_dir.join("retro").join("response.md"), text);
+        Ok(())
+    }
+
+    async fn get_retro_response(&self) -> Result<Option<String>> {
+        self.inner.get_retro_response().await
+    }
+
+    async fn put_artifact_value(&self, artifact_id: &str, value: &serde_json::Value) -> Result<()> {
+        self.inner.put_artifact_value(artifact_id, value).await
+    }
+
+    async fn get_artifact_value(&self, artifact_id: &str) -> Result<Option<serde_json::Value>> {
+        self.inner.get_artifact_value(artifact_id).await
+    }
+
+    async fn list_artifact_values(&self) -> Result<Vec<String>> {
+        self.inner.list_artifact_values().await
+    }
+
+    async fn put_asset(&self, node: &NodeVisitRef<'_>, filename: &str, data: &[u8]) -> Result<()> {
+        self.inner.put_asset(node, filename, data).await
+    }
+
+    async fn get_asset(&self, node: &NodeVisitRef<'_>, filename: &str) -> Result<Option<Bytes>> {
+        self.inner.get_asset(node, filename).await
+    }
+
+    async fn list_assets(&self, node: &NodeVisitRef<'_>) -> Result<Vec<String>> {
+        self.inner.list_assets(node).await
+    }
+
+    async fn list_all_assets(&self) -> Result<Vec<(String, u32, String)>> {
+        self.inner.list_all_assets().await
+    }
+
+    async fn get_snapshot(&self) -> Result<Option<RunSnapshot>> {
+        self.inner.get_snapshot().await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+    use std::path::PathBuf;
+
+    use chrono::{DateTime, Duration as ChronoDuration, Utc};
+    use tempfile::TempDir;
+
+    use super::*;
+    use crate::{InMemoryStore, Store};
+    use fabro_types::{
+        AggregateStats, AttrValue, FabroSettings, Graph, RunStatus, StageStatus, StatusReason,
+    };
+
+    fn dt(rfc3339: &str) -> DateTime<Utc> {
+        DateTime::parse_from_rfc3339(rfc3339)
+            .unwrap()
+            .with_timezone(&Utc)
+    }
+
+    fn sample_run_record(run_id: &str, created_at: DateTime<Utc>) -> RunRecord {
+        let mut graph = Graph::new("night-sky");
+        graph.attrs.insert(
+            "goal".to_string(),
+            AttrValue::String("map the constellations".to_string()),
+        );
+        RunRecord {
+            run_id: run_id.to_string(),
+            created_at,
+            settings: FabroSettings::default(),
+            graph,
+            workflow_slug: Some("night-sky".to_string()),
+            working_directory: PathBuf::from("/tmp/night-sky"),
+            host_repo_path: Some("github.com/fabro-sh/fabro".to_string()),
+            base_branch: Some("main".to_string()),
+            labels: HashMap::from([("team".to_string(), "infra".to_string())]),
+        }
+    }
+
+    fn sample_start_record(run_id: &str, created_at: DateTime<Utc>) -> StartRecord {
+        StartRecord {
+            run_id: run_id.to_string(),
+            start_time: created_at + ChronoDuration::seconds(5),
+            run_branch: Some("fabro/run/demo".to_string()),
+            base_sha: Some("abc123".to_string()),
+        }
+    }
+
+    fn sample_status(status: RunStatus, reason: Option<StatusReason>) -> RunStatusRecord {
+        RunStatusRecord {
+            status,
+            reason,
+            updated_at: dt("2026-03-27T12:05:00Z"),
+        }
+    }
+
+    fn sample_checkpoint() -> Checkpoint {
+        Checkpoint {
+            timestamp: dt("2026-03-27T12:10:00Z"),
+            current_node: "code".to_string(),
+            completed_nodes: vec!["plan".to_string()],
+            node_retries: HashMap::from([("code".to_string(), 1)]),
+            context_values: HashMap::from([(
+                "artifact".to_string(),
+                serde_json::json!({"kind": "summary"}),
+            )]),
+            node_outcomes: HashMap::new(),
+            next_node_id: Some("review".to_string()),
+            git_commit_sha: Some("def456".to_string()),
+            loop_failure_signatures: HashMap::new(),
+            restart_failure_signatures: HashMap::new(),
+            node_visits: HashMap::from([("code".to_string(), 2)]),
+        }
+    }
+
+    fn sample_conclusion() -> Conclusion {
+        Conclusion {
+            timestamp: dt("2026-03-27T12:15:00Z"),
+            status: StageStatus::Success,
+            duration_ms: 3210,
+            failure_reason: None,
+            final_git_commit_sha: Some("feedbeef".to_string()),
+            stages: Vec::new(),
+            total_cost: Some(1.25),
+            total_retries: 2,
+            total_input_tokens: 10,
+            total_output_tokens: 20,
+            total_cache_read_tokens: 30,
+            total_cache_write_tokens: 40,
+            total_reasoning_tokens: 50,
+            has_pricing: true,
+        }
+    }
+
+    fn sample_retro(run_id: &str) -> Retro {
+        Retro {
+            run_id: run_id.to_string(),
+            workflow_name: "night-sky".to_string(),
+            goal: "map the constellations".to_string(),
+            timestamp: dt("2026-03-27T12:20:00Z"),
+            smoothness: None,
+            stages: Vec::new(),
+            stats: AggregateStats {
+                total_duration_ms: 3210,
+                total_cost: Some(1.25),
+                total_retries: 2,
+                files_touched: vec!["src/lib.rs".to_string()],
+                stages_completed: 3,
+                stages_failed: 0,
+            },
+            intent: Some("ship the fix".to_string()),
+            outcome: Some("done".to_string()),
+            learnings: None,
+            friction_points: None,
+            open_items: None,
+        }
+    }
+
+    fn sample_sandbox() -> SandboxRecord {
+        SandboxRecord {
+            provider: "local".to_string(),
+            working_directory: "/tmp/night-sky".to_string(),
+            identifier: Some("sandbox-1".to_string()),
+            host_working_directory: Some("/tmp/night-sky".to_string()),
+            container_mount_point: None,
+            data_host: None,
+        }
+    }
+
+    fn sample_node_status() -> NodeStatusRecord {
+        NodeStatusRecord {
+            status: StageStatus::PartialSuccess,
+            notes: Some("captured output".to_string()),
+            failure_reason: Some("minor lint".to_string()),
+            timestamp: dt("2026-03-27T12:12:00Z"),
+        }
+    }
+
+    fn event_payload(run_id: &str, ts: &str, event: &str) -> EventPayload {
+        EventPayload::new(
+            serde_json::json!({
+                "ts": ts,
+                "run_id": run_id,
+                "event": event,
+            }),
+            run_id,
+        )
+        .unwrap()
+    }
+
+    async fn make_store(
+        run_dir: &Path,
+        created_at: DateTime<Utc>,
+    ) -> (Arc<dyn RunStore>, DiskProjectingRunStore) {
+        let inner = InMemoryStore::default()
+            .create_run(
+                "run-1",
+                created_at,
+                Some(run_dir.to_string_lossy().as_ref()),
+            )
+            .await
+            .unwrap();
+        let projected = DiskProjectingRunStore::new(Arc::clone(&inner), run_dir.to_path_buf());
+        (inner, projected)
+    }
+
+    #[tokio::test]
+    async fn put_methods_project_expected_files() {
+        let temp = TempDir::new().unwrap();
+        let created_at = dt("2026-03-27T12:00:00Z");
+        let (_inner, store) = make_store(temp.path(), created_at).await;
+
+        let run = sample_run_record("run-1", created_at);
+        let start = sample_start_record("run-1", created_at);
+        let status = sample_status(RunStatus::Running, Some(StatusReason::SandboxInitializing));
+        let checkpoint = sample_checkpoint();
+        let conclusion = sample_conclusion();
+        let retro = sample_retro("run-1");
+        let sandbox = sample_sandbox();
+        let node_status = sample_node_status();
+
+        store.put_run(&run).await.unwrap();
+        store.put_start(&start).await.unwrap();
+        store.put_status(&status).await.unwrap();
+        store.put_checkpoint(&checkpoint).await.unwrap();
+        store.put_conclusion(&conclusion).await.unwrap();
+        store.put_retro(&retro).await.unwrap();
+        store.put_graph("digraph night_sky {}").await.unwrap();
+        store.put_sandbox(&sandbox).await.unwrap();
+
+        let visit_one = NodeVisitRef {
+            node_id: "code",
+            visit: 1,
+        };
+        store
+            .put_node_response(&visit_one, "Applied the fix")
+            .await
+            .unwrap();
+        store
+            .put_node_status(&visit_one, &node_status)
+            .await
+            .unwrap();
+        store.put_node_stdout(&visit_one, "stdout").await.unwrap();
+        store.put_node_stderr(&visit_one, "stderr").await.unwrap();
+
+        let visit_two = NodeVisitRef {
+            node_id: "code",
+            visit: 2,
+        };
+        store
+            .put_node_prompt(&visit_two, "Plan the fix")
+            .await
+            .unwrap();
+        store.put_retro_prompt("How did it go?").await.unwrap();
+        store.put_retro_response("Smooth enough").await.unwrap();
+
+        assert_eq!(
+            serde_json::to_value(
+                serde_json::from_str::<RunRecord>(
+                    &fs::read_to_string(temp.path().join("run.json")).unwrap()
+                )
+                .unwrap()
+            )
+            .unwrap(),
+            serde_json::to_value(run).unwrap()
+        );
+        assert_eq!(
+            serde_json::to_value(
+                serde_json::from_str::<StartRecord>(
+                    &fs::read_to_string(temp.path().join("start.json")).unwrap()
+                )
+                .unwrap()
+            )
+            .unwrap(),
+            serde_json::to_value(start).unwrap()
+        );
+        assert_eq!(
+            serde_json::to_value(
+                serde_json::from_str::<RunStatusRecord>(
+                    &fs::read_to_string(temp.path().join("status.json")).unwrap()
+                )
+                .unwrap()
+            )
+            .unwrap(),
+            serde_json::to_value(status).unwrap()
+        );
+        assert_eq!(
+            serde_json::to_value(
+                serde_json::from_str::<Checkpoint>(
+                    &fs::read_to_string(temp.path().join("checkpoint.json")).unwrap()
+                )
+                .unwrap()
+            )
+            .unwrap(),
+            serde_json::to_value(checkpoint).unwrap()
+        );
+        assert_eq!(
+            serde_json::to_value(
+                serde_json::from_str::<Conclusion>(
+                    &fs::read_to_string(temp.path().join("conclusion.json")).unwrap()
+                )
+                .unwrap()
+            )
+            .unwrap(),
+            serde_json::to_value(conclusion).unwrap()
+        );
+        assert_eq!(
+            serde_json::to_value(
+                serde_json::from_str::<Retro>(
+                    &fs::read_to_string(temp.path().join("retro.json")).unwrap()
+                )
+                .unwrap()
+            )
+            .unwrap(),
+            serde_json::to_value(retro).unwrap()
+        );
+        assert_eq!(
+            fs::read_to_string(temp.path().join("workflow.fabro")).unwrap(),
+            "digraph night_sky {}"
+        );
+        assert_eq!(
+            serde_json::to_value(
+                serde_json::from_str::<SandboxRecord>(
+                    &fs::read_to_string(temp.path().join("sandbox.json")).unwrap()
+                )
+                .unwrap()
+            )
+            .unwrap(),
+            serde_json::to_value(sandbox).unwrap()
+        );
+        assert_eq!(
+            fs::read_to_string(temp.path().join("nodes/code/response.md")).unwrap(),
+            "Applied the fix"
+        );
+        assert_eq!(
+            serde_json::to_value(
+                serde_json::from_str::<NodeStatusRecord>(
+                    &fs::read_to_string(temp.path().join("nodes/code/status.json")).unwrap()
+                )
+                .unwrap()
+            )
+            .unwrap(),
+            serde_json::to_value(node_status).unwrap()
+        );
+        assert_eq!(
+            fs::read_to_string(temp.path().join("nodes/code/stdout.log")).unwrap(),
+            "stdout"
+        );
+        assert_eq!(
+            fs::read_to_string(temp.path().join("nodes/code/stderr.log")).unwrap(),
+            "stderr"
+        );
+        assert_eq!(
+            fs::read_to_string(temp.path().join("nodes/code-visit_2/prompt.md")).unwrap(),
+            "Plan the fix"
+        );
+        assert_eq!(
+            fs::read_to_string(temp.path().join("retro/prompt.md")).unwrap(),
+            "How did it go?"
+        );
+        assert_eq!(
+            fs::read_to_string(temp.path().join("retro/response.md")).unwrap(),
+            "Smooth enough"
+        );
+    }
+
+    #[tokio::test]
+    async fn critical_projections_write_files_in_isolation() {
+        let temp = TempDir::new().unwrap();
+        let created_at = dt("2026-03-27T12:00:00Z");
+        let (_inner, store) = make_store(temp.path(), created_at).await;
+
+        let status = sample_status(RunStatus::Failed, Some(StatusReason::WorkflowError));
+        let conclusion = sample_conclusion();
+
+        store.put_status(&status).await.unwrap();
+        store.put_conclusion(&conclusion).await.unwrap();
+
+        assert!(temp.path().join("status.json").exists());
+        assert!(temp.path().join("conclusion.json").exists());
+    }
+
+    #[tokio::test]
+    async fn append_event_projects_progress_and_live_files() {
+        let temp = TempDir::new().unwrap();
+        let created_at = dt("2026-03-27T12:00:00Z");
+        let (_inner, store) = make_store(temp.path(), created_at).await;
+
+        let first = event_payload("run-1", "2026-03-27T12:00:00Z", "Started");
+        let second = event_payload("run-1", "2026-03-27T12:00:01Z", "Completed");
+
+        store.append_event(&first).await.unwrap();
+        store.append_event(&second).await.unwrap();
+
+        let progress = fs::read_to_string(temp.path().join("progress.jsonl")).unwrap();
+        let lines: Vec<&str> = progress.lines().collect();
+        assert_eq!(lines.len(), 2);
+        let first_value: serde_json::Value = serde_json::from_str(lines[0]).unwrap();
+        let second_value: serde_json::Value = serde_json::from_str(lines[1]).unwrap();
+        assert_eq!(first_value, first.as_value().clone());
+        assert_eq!(second_value, second.as_value().clone());
+
+        let live: serde_json::Value =
+            serde_json::from_str(&fs::read_to_string(temp.path().join("live.json")).unwrap())
+                .unwrap();
+        assert_eq!(live, second.as_value().clone());
+
+        let events = store.list_events().await.unwrap();
+        assert_eq!(events.len(), 2);
+        assert_eq!(events[0].payload, first);
+        assert_eq!(events[1].payload, second);
+    }
+
+    #[tokio::test]
+    async fn get_methods_read_from_inner_store_not_disk() {
+        let temp = TempDir::new().unwrap();
+        let created_at = dt("2026-03-27T12:00:00Z");
+        let (inner, store) = make_store(temp.path(), created_at).await;
+
+        let status = sample_status(RunStatus::Running, Some(StatusReason::SandboxInitializing));
+        inner.put_status(&status).await.unwrap();
+        fs::write(
+            temp.path().join("status.json"),
+            serde_json::to_string_pretty(&sample_status(
+                RunStatus::Succeeded,
+                Some(StatusReason::Completed),
+            ))
+            .unwrap(),
+        )
+        .unwrap();
+
+        assert_eq!(
+            serde_json::to_value(store.get_status().await.unwrap().unwrap()).unwrap(),
+            serde_json::to_value(status).unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn disk_failures_do_not_block_store_writes() {
+        let temp = TempDir::new().unwrap();
+        let created_at = dt("2026-03-27T12:00:00Z");
+        let (inner, store) = make_store(temp.path(), created_at).await;
+
+        let mut permissions = fs::metadata(temp.path()).unwrap().permissions();
+        permissions.set_readonly(true);
+        fs::set_permissions(temp.path(), permissions).unwrap();
+
+        let status = sample_status(RunStatus::Running, Some(StatusReason::SandboxInitializing));
+        let node = NodeVisitRef {
+            node_id: "code",
+            visit: 1,
+        };
+
+        store.put_status(&status).await.unwrap();
+        store.put_node_prompt(&node, "Plan the fix").await.unwrap();
+
+        let stored_status = inner.get_status().await.unwrap();
+        let stored_node = inner.get_node(&node).await.unwrap();
+
+        assert_eq!(
+            serde_json::to_value(stored_status.unwrap()).unwrap(),
+            serde_json::to_value(status).unwrap()
+        );
+        assert_eq!(stored_node.prompt.as_deref(), Some("Plan the fix"));
+    }
+
+    #[test]
+    fn disk_node_dir_matches_legacy_layout() {
+        let run_dir = Path::new("/tmp/fabro-run");
+
+        assert_eq!(
+            disk_node_dir(run_dir, "build", 1),
+            run_dir.join("nodes/build")
+        );
+        assert_eq!(
+            disk_node_dir(run_dir, "build", 2),
+            run_dir.join("nodes/build-visit_2")
+        );
+    }
+}

lib/crates/fabro-store/src/lib.rs

@@ -6,6 +6,7 @@ use bytes::Bytes;
 use chrono::{DateTime, Utc};
 use futures::Stream;
 
+mod disk_projecting;
 mod error;
 mod keys;
 mod memory;
@@ -13,6 +14,7 @@ mod runtime;
 mod slate;
 mod types;
 
+pub use disk_projecting::DiskProjectingRunStore;
 pub use error::{Result, StoreError};
 pub use memory::InMemoryStore;
 pub use runtime::RuntimeState;

lib/crates/fabro-workflows/src/handler/agent.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use fabro_agent::Sandbox;
+use fabro_store::NodeVisitRef;
 
 use crate::context::keys;
 use crate::context::{Context, WorkflowContext};
@@ -255,7 +256,18 @@ impl Handler for AgentHandler {
         let visit = visit_from_context(context);
         let stage_dir = node_dir(run_dir, &node.id, visit);
         fs::create_dir_all(&stage_dir).await?;
-        fs::write(stage_dir.join("prompt.md"), &prompt).await?;
+        let node_ref = NodeVisitRef {
+            node_id: &node.id,
+            visit: u32::try_from(visit).unwrap_or(u32::MAX),
+        };
+        if let Some(ref store) = services.run_store {
+            store
+                .put_node_prompt(&node_ref, &prompt)
+                .await
+                .map_err(|err| FabroError::handler(err.to_string()))?;
+        } else {
+            fs::write(stage_dir.join("prompt.md"), &prompt).await?;
+        }
 
         // 3. Call LLM backend (agent loop)
         let thread_id = context.thread_id();
@@ -314,7 +326,14 @@ impl Handler for AgentHandler {
             };
 
         // 4. Write response to logs
-        fs::write(stage_dir.join("response.md"), &response_text).await?;
+        if let Some(ref store) = services.run_store {
+            store
+                .put_node_response(&node_ref, &response_text)
+                .await
+                .map_err(|err| FabroError::handler(err.to_string()))?;
+        } else {
+            fs::write(stage_dir.join("response.md"), &response_text).await?;
+        }
 
         // 7. Build and write status
         let mut outcome = Outcome::success();

lib/crates/fabro-workflows/src/handler/command.rs

@@ -1,6 +1,7 @@
 use std::path::Path;
 
 use async_trait::async_trait;
+use fabro_store::NodeVisitRef;
 
 use crate::context::Context;
 use crate::context::keys;
@@ -90,6 +91,10 @@ impl Handler for CommandHandler {
         let visit = visit_from_context(context);
         let stage_dir = node_dir(run_dir, &node.id, visit);
         fs::create_dir_all(&stage_dir).await?;
+        let node_ref = NodeVisitRef {
+            node_id: &node.id,
+            visit: u32::try_from(visit).unwrap_or(u32::MAX),
+        };
 
         let invocation = serde_json::json!({
             "command": script,
@@ -123,8 +128,19 @@ impl Handler for CommandHandler {
             .await
             .map_err(|e| FabroError::handler(format!("Failed to spawn script: {e}")))?;
 
-        fs::write(stage_dir.join("stdout.log"), &result.stdout).await?;
-        fs::write(stage_dir.join("stderr.log"), &result.stderr).await?;
+        if let Some(ref store) = services.run_store {
+            store
+                .put_node_stdout(&node_ref, &result.stdout)
+                .await
+                .map_err(|err| FabroError::handler(err.to_string()))?;
+            store
+                .put_node_stderr(&node_ref, &result.stderr)
+                .await
+                .map_err(|err| FabroError::handler(err.to_string()))?;
+        } else {
+            fs::write(stage_dir.join("stdout.log"), &result.stdout).await?;
+            fs::write(stage_dir.join("stderr.log"), &result.stderr).await?;
+        }
 
         let timing = serde_json::json!({
             "duration_ms": result.duration_ms,

lib/crates/fabro-workflows/src/handler/fan_in.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use fabro_agent::Sandbox;
+use fabro_store::{NodeVisitRef, RunStore};
 
 use crate::context::Context;
 use crate::context::keys;
@@ -89,6 +90,7 @@ impl Handler for FanInHandler {
                 &node.id,
                 &services.emitter,
                 &services.sandbox,
+                services.run_store.clone(),
             )
             .await?
         } else {
@@ -224,6 +226,7 @@ async fn llm_evaluate(
     node_id: &str,
     emitter: &Arc<EventEmitter>,
     sandbox: &Arc<dyn Sandbox>,
+    run_store: Option<Arc<dyn RunStore>>,
 ) -> Result<Candidate, FabroError> {
     let results_text =
         serde_json::to_string_pretty(results).unwrap_or_else(|_| results.to_string());
@@ -237,7 +240,18 @@ async fn llm_evaluate(
     let visit = visit_from_context(context);
     let stage_dir = node_dir(run_dir, node_id, visit);
     fs::create_dir_all(&stage_dir).await?;
-    fs::write(stage_dir.join("prompt.md"), &full_prompt).await?;
+    let node_ref = NodeVisitRef {
+        node_id,
+        visit: u32::try_from(visit).unwrap_or(u32::MAX),
+    };
+    if let Some(ref store) = run_store {
+        store
+            .put_node_prompt(&node_ref, &full_prompt)
+            .await
+            .map_err(|err| FabroError::handler(err.to_string()))?;
+    } else {
+        fs::write(stage_dir.join("prompt.md"), &full_prompt).await?;
+    }
 
     // Build a synthetic node for the backend call
     let eval_node = Node::new("fan_in_eval");
@@ -267,7 +281,14 @@ async fn llm_evaluate(
                 .unwrap_or_else(|| "unknown".to_string());
             let response_text =
                 serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string());
-            fs::write(stage_dir.join("response.md"), &response_text).await?;
+            if let Some(ref store) = run_store {
+                store
+                    .put_node_response(&node_ref, &response_text)
+                    .await
+                    .map_err(|err| FabroError::handler(err.to_string()))?;
+            } else {
+                fs::write(stage_dir.join("response.md"), &response_text).await?;
+            }
             Ok(Candidate {
                 id: best_id,
                 status: outcome.status.to_string(),
@@ -276,7 +297,14 @@ async fn llm_evaluate(
         }
         Ok(CodergenResult::Text { text, .. }) => {
             // Write response to logs
-            fs::write(stage_dir.join("response.md"), &text).await?;
+            if let Some(ref store) = run_store {
+                store
+                    .put_node_response(&node_ref, &text)
+                    .await
+                    .map_err(|err| FabroError::handler(err.to_string()))?;
+            } else {
+                fs::write(stage_dir.join("response.md"), &text).await?;
+            }
 
             // The LLM responded with text; try to find a matching candidate ID
             let text = text.trim().to_string();

lib/crates/fabro-workflows/src/handler/mod.rs

@@ -18,6 +18,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use fabro_agent::Sandbox;
+use fabro_store::RunStore;
 
 use crate::context::Context;
 use crate::error::FabroError;
@@ -33,6 +34,7 @@ pub struct EngineServices {
     pub registry: Arc<HandlerRegistry>,
     pub emitter: Arc<EventEmitter>,
     pub sandbox: Arc<dyn Sandbox>,
+    pub run_store: Option<Arc<dyn RunStore>>,
     /// Git state for the current run. Set via `set_git_state` at the start of
     /// `run_via_core` and read by parallel/fan-in handlers.
     pub(crate) git_state: std::sync::RwLock<Option<Arc<GitState>>>,
@@ -73,6 +75,7 @@ impl EngineServices {
             sandbox: Arc::new(fabro_agent::LocalSandbox::new(
                 std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
             )),
+            run_store: None,
             git_state: std::sync::RwLock::new(None),
             hook_runner: None,
             env: HashMap::new(),

lib/crates/fabro-workflows/src/handler/parallel.rs

@@ -262,6 +262,7 @@ impl Handler for ParallelHandler {
             let registry = Arc::clone(&services.registry);
             let emitter = Arc::clone(&services.emitter);
             let hook_runner = services.hook_runner.clone();
+            let run_store = services.run_store.clone();
             let env = services.env.clone();
             let dry_run = services.dry_run;
             let graph = graph.clone();
@@ -309,6 +310,7 @@ impl Handler for ParallelHandler {
                     registry: Arc::clone(&registry),
                     emitter: Arc::clone(&emitter),
                     sandbox: Arc::clone(&setup.sandbox),
+                    run_store: run_store.clone(),
                     git_state: std::sync::RwLock::new(None),
                     hook_runner: hook_runner.clone(),
                     env: env.clone(),

lib/crates/fabro-workflows/src/handler/prompt.rs

@@ -3,6 +3,7 @@ use std::path::Path;
 use async_trait::async_trait;
 
 use fabro_model::Provider;
+use fabro_store::NodeVisitRef;
 
 use crate::context::keys;
 use crate::context::{Context, WorkflowContext};
@@ -91,7 +92,18 @@ impl Handler for PromptHandler {
         let visit = visit_from_context(context);
         let stage_dir = node_dir(run_dir, &node.id, visit);
         fs::create_dir_all(&stage_dir).await?;
-        fs::write(stage_dir.join("prompt.md"), &prompt).await?;
+        let node_ref = NodeVisitRef {
+            node_id: &node.id,
+            visit: u32::try_from(visit).unwrap_or(u32::MAX),
+        };
+        if let Some(ref store) = services.run_store {
+            store
+                .put_node_prompt(&node_ref, &prompt)
+                .await
+                .map_err(|err| FabroError::handler(err.to_string()))?;
+        } else {
+            fs::write(stage_dir.join("prompt.md"), &prompt).await?;
+        }
 
         // 3. Call LLM backend (one_shot)
         let (response_text, stage_usage, backend_files_touched) =
@@ -128,7 +140,14 @@ impl Handler for PromptHandler {
             };
 
         // 4. Write response to logs
-        fs::write(stage_dir.join("response.md"), &response_text).await?;
+        if let Some(ref store) = services.run_store {
+            store
+                .put_node_response(&node_ref, &response_text)
+                .await
+                .map_err(|err| FabroError::handler(err.to_string()))?;
+        } else {
+            fs::write(stage_dir.join("response.md"), &response_text).await?;
+        }
 
         // 5. Build and write status
         let mut outcome = Outcome::success();

lib/crates/fabro-workflows/src/lifecycle/disk.rs

@@ -16,10 +16,9 @@ use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
 use crate::graph::WorkflowGraph;
 use crate::graph::WorkflowNode;
 use crate::outcome::{OutcomeExt, StageUsage};
-use crate::records::{Checkpoint, CheckpointExt};
-use crate::run_dir::{write_node_status, write_start_record};
+use crate::records::{Checkpoint, StartRecord};
 use crate::run_options::RunOptions;
-use crate::run_status::{RunStatus, write_run_status};
+use crate::run_status::RunStatus;
 use fabro_graphviz::graph::types::Graph as GvGraph;
 
 type WfRunState = RunState<Option<StageUsage>>;
@@ -40,10 +39,13 @@ pub(crate) struct DiskLifecycle {
 #[async_trait]
 impl RunLifecycle<WorkflowGraph> for DiskLifecycle {
     async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &WfRunState) -> CoreResult<()> {
-        // Write start.json
-        let start_record = write_start_record(&self.run_dir, &self.run_options);
-        // Write run status as Running
-        write_run_status(&self.run_dir, RunStatus::Running, None);
+        let git_state = self.run_options.git.as_ref();
+        let start_record = StartRecord {
+            run_id: self.run_id.clone(),
+            start_time: chrono::Utc::now(),
+            run_branch: git_state.and_then(|g| g.run_branch.clone()),
+            base_sha: git_state.and_then(|g| g.base_sha.clone()),
+        };
         if let Err(err) = self.run_store.put_start(&start_record).await {
             self.emitter.emit(&WorkflowRunEvent::RunNotice {
                 level: RunNoticeLevel::Warn,
@@ -73,7 +75,6 @@ impl RunLifecycle<WorkflowGraph> for DiskLifecycle {
     ) -> CoreResult<()> {
         let gv = node.inner();
         let visit = state.node_visits.get(gv.id.as_str()).copied().unwrap_or(1);
-        write_node_status(&self.run_dir, &gv.id, visit, &result.outcome);
         let node_status = NodeStatusRecord {
             status: result.outcome.status.clone(),
             notes: result.outcome.notes.clone(),
@@ -130,15 +131,6 @@ impl RunLifecycle<WorkflowGraph> for DiskLifecycle {
             loop_failure_signatures: loop_sigs,
             restart_failure_signatures: restart_sigs,
         };
-
-        let checkpoint_path = self.run_dir.join("checkpoint.json");
-        if let Err(e) = checkpoint.save(&checkpoint_path) {
-            self.emitter.emit(&WorkflowRunEvent::RunNotice {
-                level: RunNoticeLevel::Warn,
-                code: "checkpoint_disk_save_failed".to_string(),
-                message: format!("[node: {}] checkpoint save failed: {e}", node.id()),
-            });
-        }
         if let Err(err) = self.run_store.put_checkpoint(&checkpoint).await {
             self.emitter.emit(&WorkflowRunEvent::RunNotice {
                 level: RunNoticeLevel::Warn,

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

@@ -17,7 +17,6 @@ use crate::git::scan_node_files;
 use crate::graph::WorkflowGraph;
 use crate::graph::WorkflowNode;
 use crate::outcome::{Outcome, StageStatus, StageUsage};
-use crate::records::{Checkpoint, CheckpointExt};
 use crate::run_dir::node_dir;
 use crate::run_options::RunOptions;
 use crate::sandbox_git::{git_checkpoint, git_diff, git_push_host};
@@ -198,20 +197,6 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
                     push_results: Vec::new(),
                 };
 
-                // Re-save checkpoint.json with SHA
-                let checkpoint_path = self.run_dir.join("checkpoint.json");
-                if let Ok(mut cp) = Checkpoint::load(&checkpoint_path) {
-                    cp.git_commit_sha = Some(sha.clone());
-                    if let Err(e) = cp.save(&checkpoint_path) {
-                        self.emitter.emit(&WorkflowRunEvent::RunNotice {
-                            level: RunNoticeLevel::Warn,
-                            code: "checkpoint_resave_failed".to_string(),
-                            message: format!(
-                                "[node: {node_id}] checkpoint re-save with SHA failed: {e}"
-                            ),
-                        });
-                    }
-                }
                 match self.run_store.get_checkpoint().await {
                     Ok(Some(mut checkpoint)) => {
                         checkpoint.git_commit_sha = Some(sha.clone());

lib/crates/fabro-workflows/src/operations/start.rs

@@ -11,14 +11,14 @@ use fabro_config::{project as project_config, run as run_config, sandbox as sand
 use fabro_interview::{AutoApproveInterviewer, Interviewer};
 use fabro_model::{Catalog, FallbackTarget, Provider};
 use fabro_sandbox::{SandboxProvider, SandboxSpec, detect_clone_params};
-use fabro_store::RunStore;
+use fabro_store::{DiskProjectingRunStore, RunStore};
 use serde::Serialize;
 
 use crate::context::Context;
 use crate::error::FabroError;
 use crate::event::{
-    EventEmitter, ProgressLogger, RunNoticeLevel, StoreProgressLogger, WorkflowRunEvent,
-    append_progress_event, build_redacted_event_payload,
+    EventEmitter, RunNoticeLevel, StoreProgressLogger, WorkflowRunEvent, append_progress_event,
+    build_redacted_event_payload,
 };
 use crate::git::GitAuthor;
 use crate::handler::HandlerRegistry;
@@ -26,7 +26,7 @@ use crate::outcome::{Outcome, StageStatus};
 use crate::pipeline::{
     self, DevcontainerSpec, FinalizeOptions, Finalized, InitOptions, LlmSpec, Persisted,
     PullRequestOptions, RetroOptions, SandboxEnvSpec, build_conclusion_from_store,
-    classify_engine_result, persist_terminal_outcome,
+    classify_engine_result,
 };
 use crate::records::{Checkpoint, Conclusion, ConclusionExt, RunRecord, RunRecordExt};
 use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
@@ -115,8 +115,13 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result<Started, F
 pub(super) async fn execute_persisted_run(
     run_dir: &Path,
     checkpoint: Option<Checkpoint>,
-    services: StartServices,
+    mut services: StartServices,
 ) -> Result<Started, FabroError> {
+    let inner_store = Arc::clone(&services.run_store);
+    services.run_store = Arc::new(DiskProjectingRunStore::new(
+        inner_store,
+        run_dir.to_path_buf(),
+    ));
     let run_store = Arc::clone(&services.run_store);
     if let Err(err) = run_store
         .put_status(&run_status::RunStatusRecord::new(
@@ -208,7 +213,6 @@ async fn persist_terminal_engine_failure(
         None,
     )
     .await;
-    persist_terminal_outcome(run_dir, &conclusion, run_status, status_reason);
     if let Err(err) = run_store.put_conclusion(&conclusion).await {
         tracing::warn!(error = %err, "Failed to save terminal engine failure conclusion to store");
     }
@@ -477,8 +481,6 @@ impl RunSession {
             });
         }
 
-        ProgressLogger::new(persisted.run_dir(), record.run_id.clone())
-            .register(self.emitter.as_ref());
         let store_progress_logger =
             StoreProgressLogger::new(Arc::clone(&self.run_store), record.run_id.clone());
         store_progress_logger.register(self.emitter.as_ref());

lib/crates/fabro-workflows/src/pipeline/execute.rs

@@ -73,6 +73,7 @@ pub async fn execute(init: Initialized) -> Executed {
         registry,
         emitter: Arc::clone(&emitter),
         sandbox: Arc::clone(&sandbox),
+        run_store: Some(Arc::clone(&run_store)),
         git_state: std::sync::RwLock::new(git_state),
         hook_runner: hook_runner.clone(),
         env,

lib/crates/fabro-workflows/src/pipeline/finalize.rs

@@ -406,7 +406,6 @@ pub async fn finalize(
         );
     }
 
-    persist_terminal_outcome(&options.run_dir, &conclusion, run_status, status_reason);
     if let Err(err) = options.run_store.put_conclusion(&conclusion).await {
         tracing::warn!(error = %err, "Failed to save conclusion to store");
     }
@@ -468,7 +467,7 @@ mod tests {
         let temp = tempfile::tempdir().unwrap();
         let run_dir = temp.path().join("run");
         std::fs::create_dir_all(&run_dir).unwrap();
-        let run_store = InMemoryStore::default()
+        let inner_store = InMemoryStore::default()
             .create_run(
                 "run-test",
                 Utc::now(),
@@ -476,6 +475,9 @@ mod tests {
             )
             .await
             .unwrap();
+        let run_store: Arc<dyn fabro_store::RunStore> = Arc::new(
+            fabro_store::DiskProjectingRunStore::new(inner_store, run_dir.clone()),
+        );
         let retroed = Retroed {
             graph: Graph::new("test"),
             outcome: Ok(Outcome::success()),
@@ -495,7 +497,7 @@ mod tests {
             &FinalizeOptions {
                 run_dir: run_dir.clone(),
                 run_id: "run-test".to_string(),
-                run_store,
+                run_store: Arc::clone(&run_store),
                 workflow_name: "test".to_string(),
                 hook_runner: None,
                 preserve_sandbox: true,

lib/crates/fabro-workflows/src/pipeline/initialize.rs

@@ -7,8 +7,7 @@ use fabro_agent::Sandbox;
 use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner};
 use fabro_llm::client::Client;
 use fabro_sandbox::{
-    ReadBeforeWriteSandbox, SandboxEventCallback, SandboxRecordExt, WorkdirStrategy,
-    WorktreeConfig, WorktreeSandbox,
+    ReadBeforeWriteSandbox, SandboxEventCallback, WorkdirStrategy, WorktreeConfig, WorktreeSandbox,
 };
 use shlex::try_quote;
 
@@ -496,9 +495,6 @@ pub async fn initialize(
         working_directory: sandbox.working_directory().to_string(),
     });
     let sandbox_record = options.sandbox.to_sandbox_record(&*sandbox);
-    if let Err(e) = sandbox_record.save(&run_dir.join("sandbox.json")) {
-        tracing::warn!(error = %e, "Failed to save sandbox record");
-    }
     if let Err(err) = options.run_store.put_sandbox(&sandbox_record).await {
         tracing::warn!(error = %err, "Failed to save sandbox record to store");
     }
@@ -739,10 +735,18 @@ mod tests {
                 run_id: "run-test".to_string(),
                 run_store: {
                     let store: &dyn fabro_store::Store = &InMemoryStore::default();
-                    store
-                        .create_run("test-run", chrono::Utc::now(), None)
+                    let inner = store
+                        .create_run(
+                            "test-run",
+                            chrono::Utc::now(),
+                            Some(run_dir.to_string_lossy().as_ref()),
+                        )
                         .await
-                        .unwrap()
+                        .unwrap();
+                    Arc::new(fabro_store::DiskProjectingRunStore::new(
+                        inner,
+                        run_dir.clone(),
+                    ))
                 },
                 dry_run: false,
                 emitter,
@@ -809,10 +813,18 @@ mod tests {
                 run_id: "run-test".to_string(),
                 run_store: {
                     let store: &dyn fabro_store::Store = &InMemoryStore::default();
-                    store
-                        .create_run("test-run", chrono::Utc::now(), None)
+                    let inner = store
+                        .create_run(
+                            "test-run",
+                            chrono::Utc::now(),
+                            Some(run_dir.to_string_lossy().as_ref()),
+                        )
                         .await
-                        .unwrap()
+                        .unwrap();
+                    Arc::new(fabro_store::DiskProjectingRunStore::new(
+                        inner,
+                        run_dir.clone(),
+                    ))
                 },
                 dry_run: false,
                 emitter,

lib/crates/fabro-workflows/src/pipeline/retro.rs

@@ -1,7 +1,6 @@
 use std::sync::Arc;
 
 use fabro_agent::SessionEvent;
-use fabro_retro::RetroExt;
 use fabro_retro::retro::{Retro, derive_retro, extract_stage_durations};
 use fabro_retro::retro_agent::{dry_run_narrative, run_retro_agent};
 
@@ -50,9 +49,6 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
         &stage_durations,
     );
 
-    if let Err(e) = retro.save(&options.run_dir) {
-        tracing::warn!(error = %e, "Failed to save initial retro");
-    }
     if let Err(err) = options.run_store.put_retro(&retro).await {
         tracing::warn!(error = %err, "Failed to save initial retro to store");
     }
@@ -116,9 +112,6 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
     match narrative_result {
         Ok(narrative) => {
             retro.apply_narrative(narrative);
-            if let Err(e) = retro.save(&options.run_dir) {
-                tracing::warn!(error = %e, "Failed to save retro with narrative");
-            }
             if let Err(err) = options.run_store.put_retro(&retro).await {
                 tracing::warn!(error = %err, "Failed to save retro with narrative to store");
             }
@@ -213,7 +206,7 @@ mod tests {
         run_dir: &std::path::Path,
         checkpoint: &Checkpoint,
     ) -> Arc<dyn fabro_store::RunStore> {
-        let run_store = InMemoryStore::default()
+        let inner = InMemoryStore::default()
             .create_run(
                 "run-test",
                 Utc::now(),
@@ -221,6 +214,9 @@ mod tests {
             )
             .await
             .unwrap();
+        let run_store: Arc<dyn fabro_store::RunStore> = Arc::new(
+            fabro_store::DiskProjectingRunStore::new(inner, run_dir.to_path_buf()),
+        );
         run_store.put_checkpoint(checkpoint).await.unwrap();
         run_store
     }

lib/crates/fabro-workflows/src/run_dir.rs

@@ -1,26 +1,6 @@
 use std::path::{Path, PathBuf};
 
-use chrono::Utc;
-use fabro_types::NodeStatusRecord;
-
 use crate::context::Context;
-use crate::outcome::{Outcome, OutcomeExt};
-use crate::records::{StartRecord, StartRecordExt};
-use crate::run_options::RunOptions;
-
-/// Write start.json at the start of a workflow run. Returns the StartRecord.
-pub(crate) fn write_start_record(run_dir: &Path, settings: &RunOptions) -> StartRecord {
-    let git_state = settings.git.as_ref();
-    let record = StartRecord {
-        run_id: settings.run_id.clone(),
-        start_time: Utc::now(),
-        run_branch: git_state.and_then(|g| g.run_branch.clone()),
-        base_sha: git_state.and_then(|g| g.base_sha.clone()),
-    };
-    let _ = std::fs::create_dir_all(run_dir);
-    let _ = record.save(run_dir);
-    record
-}
 
 /// Return the directory for a node's logs.
 ///
@@ -44,28 +24,9 @@ pub(crate) fn visit_from_context(context: &Context) -> usize {
     context.node_visit_count().max(1)
 }
 
-/// Write status.json for a completed node into {`run_dir}/nodes/{node_id}/status.json`.
-pub(crate) fn write_node_status(run_dir: &Path, node_id: &str, visit: usize, outcome: &Outcome) {
-    let node_dir = node_dir(run_dir, node_id, visit);
-    let _ = std::fs::create_dir_all(&node_dir);
-    let status = NodeStatusRecord {
-        status: outcome.status.clone(),
-        notes: outcome.notes.clone(),
-        failure_reason: outcome.failure_reason().map(ToOwned::to_owned),
-        timestamp: Utc::now(),
-    };
-    if let Ok(json) = serde_json::to_string_pretty(&status) {
-        let _ = std::fs::write(node_dir.join("status.json"), json);
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::path::Path;
-
-    use fabro_types::StageStatus;
-    use tempfile::TempDir;
 
     use crate::context::Context;
 
@@ -108,46 +69,4 @@ mod tests {
             root.join("nodes").join("work-visit_5")
         );
     }
-
-    #[test]
-    fn write_node_status_uses_typed_record_with_legacy_shape() {
-        let temp = TempDir::new().unwrap();
-        let outcome = Outcome {
-            status: StageStatus::Fail,
-            notes: Some("needs retry".to_string()),
-            failure: Some(crate::outcome::FailureDetail::new(
-                "boom",
-                crate::outcome::FailureCategory::Deterministic,
-            )),
-            ..Outcome::default()
-        };
-
-        write_node_status(temp.path(), "work", 1, &outcome);
-
-        let data = std::fs::read_to_string(temp.path().join("nodes/work/status.json")).unwrap();
-        let value: serde_json::Value = serde_json::from_str(&data).unwrap();
-        assert_eq!(value.get("status"), Some(&serde_json::json!("fail")));
-        assert_eq!(value.get("notes"), Some(&serde_json::json!("needs retry")));
-        assert_eq!(
-            value.get("failure_reason"),
-            Some(&serde_json::json!("boom"))
-        );
-        assert!(value.get("timestamp").and_then(|v| v.as_str()).is_some());
-    }
-
-    #[test]
-    fn write_node_status_preserves_null_optional_fields() {
-        let temp = TempDir::new().unwrap();
-        let outcome = Outcome {
-            status: StageStatus::Success,
-            ..Outcome::default()
-        };
-
-        write_node_status(temp.path(), "work", 1, &outcome);
-
-        let data = std::fs::read_to_string(temp.path().join("nodes/work/status.json")).unwrap();
-        let value: serde_json::Value = serde_json::from_str(&data).unwrap();
-        assert_eq!(value.get("notes"), Some(&serde_json::Value::Null));
-        assert_eq!(value.get("failure_reason"), Some(&serde_json::Value::Null));
-    }
 }

lib/crates/fabro-workflows/src/test_support.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 use chrono::Utc;
 use fabro_agent::Sandbox;
 use fabro_graphviz::graph::Graph as GvGraph;
-use fabro_store::{InMemoryStore, Store};
+use fabro_store::{DiskProjectingRunStore, InMemoryStore, Store};
 
 use crate::error::Result;
 use crate::event::EventEmitter;
@@ -30,7 +30,7 @@ async fn initialized(
     options: InitializedOptions,
 ) -> Initialized {
     std::fs::create_dir_all(&run_options.run_dir).expect("failed to create run dir");
-    let run_store = InMemoryStore::default()
+    let inner_store = InMemoryStore::default()
         .create_run(
             &run_options.run_id,
             Utc::now(),
@@ -38,6 +38,10 @@ async fn initialized(
         )
         .await
         .expect("failed to create in-memory run store");
+    let run_store = Arc::new(DiskProjectingRunStore::new(
+        inner_store,
+        run_options.run_dir.clone(),
+    ));
     Initialized {
         graph: graph.clone(),
         source: String::new(),