diff --git a/lib/crates/fabro-retro/src/retro_agent.rs b/lib/crates/fabro-retro/src/retro_agent.rs index f4ffce5aa..1d1b62f71 100644 --- a/lib/crates/fabro-retro/src/retro_agent.rs +++ b/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", diff --git a/lib/crates/fabro-store/src/disk_projecting.rs b/lib/crates/fabro-store/src/disk_projecting.rs new file mode 100644 index 000000000..932d7dc79 --- /dev/null +++ b/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, + run_dir: PathBuf, +} + +impl DiskProjectingRunStore { + #[must_use] + pub fn new(inner: Arc, 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(path: &Path, value: &T) { + if let Err(err) = write_json(path, value) { + Self::warn_projection(path, &err, true); + } + } + + fn write_json_best_effort(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(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> { + 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> { + 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> { + 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> { + self.inner.get_checkpoint().await + } + + async fn append_checkpoint(&self, record: &Checkpoint) -> Result { + self.inner.append_checkpoint(record).await + } + + async fn list_checkpoints(&self) -> Result> { + 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> { + 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> { + 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> { + 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> { + 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 { + self.inner.get_node(node).await + } + + async fn list_node_visits(&self, node_id: &str) -> Result> { + self.inner.list_node_visits(node_id).await + } + + async fn append_event(&self, payload: &EventPayload) -> Result { + self.append_jsonl_critical(payload); + self.inner.append_event(payload).await + } + + async fn list_events(&self) -> Result> { + self.inner.list_events().await + } + + async fn list_events_from(&self, seq: u32) -> Result> { + self.inner.list_events_from(seq).await + } + + async fn watch_events_from( + &self, + seq: u32, + ) -> Result> + 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> { + 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> { + 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> { + self.inner.get_artifact_value(artifact_id).await + } + + async fn list_artifact_values(&self) -> Result> { + 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> { + self.inner.get_asset(node, filename).await + } + + async fn list_assets(&self, node: &NodeVisitRef<'_>) -> Result> { + self.inner.list_assets(node).await + } + + async fn list_all_assets(&self) -> Result> { + self.inner.list_all_assets().await + } + + async fn get_snapshot(&self) -> Result> { + 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 { + DateTime::parse_from_rfc3339(rfc3339) + .unwrap() + .with_timezone(&Utc) + } + + fn sample_run_record(run_id: &str, created_at: DateTime) -> 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) -> 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) -> 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, + ) -> (Arc, 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::( + &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::( + &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::( + &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::( + &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::( + &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::( + &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::( + &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::( + &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") + ); + } +} diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index 90dccb261..3064f3918 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/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; diff --git a/lib/crates/fabro-workflows/src/handler/agent.rs b/lib/crates/fabro-workflows/src/handler/agent.rs index 25f01c618..646e9af37 100644 --- a/lib/crates/fabro-workflows/src/handler/agent.rs +++ b/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(); diff --git a/lib/crates/fabro-workflows/src/handler/command.rs b/lib/crates/fabro-workflows/src/handler/command.rs index 971bf8e9a..55ff15eaa 100644 --- a/lib/crates/fabro-workflows/src/handler/command.rs +++ b/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, diff --git a/lib/crates/fabro-workflows/src/handler/fan_in.rs b/lib/crates/fabro-workflows/src/handler/fan_in.rs index 56e742c5c..8423f6308 100644 --- a/lib/crates/fabro-workflows/src/handler/fan_in.rs +++ b/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, sandbox: &Arc, + run_store: Option>, ) -> Result { 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(); diff --git a/lib/crates/fabro-workflows/src/handler/mod.rs b/lib/crates/fabro-workflows/src/handler/mod.rs index 3e0995973..7e8df9140 100644 --- a/lib/crates/fabro-workflows/src/handler/mod.rs +++ b/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, pub emitter: Arc, pub sandbox: Arc, + pub run_store: Option>, /// 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>>, @@ -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(), diff --git a/lib/crates/fabro-workflows/src/handler/parallel.rs b/lib/crates/fabro-workflows/src/handler/parallel.rs index 644e8389d..07b824385 100644 --- a/lib/crates/fabro-workflows/src/handler/parallel.rs +++ b/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(®istry), 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(), diff --git a/lib/crates/fabro-workflows/src/handler/prompt.rs b/lib/crates/fabro-workflows/src/handler/prompt.rs index c761a8024..a24bb3ce8 100644 --- a/lib/crates/fabro-workflows/src/handler/prompt.rs +++ b/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(); diff --git a/lib/crates/fabro-workflows/src/lifecycle/disk.rs b/lib/crates/fabro-workflows/src/lifecycle/disk.rs index c0dab2365..0b8f50d99 100644 --- a/lib/crates/fabro-workflows/src/lifecycle/disk.rs +++ b/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>; @@ -40,10 +39,13 @@ pub(crate) struct DiskLifecycle { #[async_trait] impl RunLifecycle 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 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 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, diff --git a/lib/crates/fabro-workflows/src/lifecycle/git.rs b/lib/crates/fabro-workflows/src/lifecycle/git.rs index 61964db7c..f59723bc3 100644 --- a/lib/crates/fabro-workflows/src/lifecycle/git.rs +++ b/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 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()); diff --git a/lib/crates/fabro-workflows/src/operations/start.rs b/lib/crates/fabro-workflows/src/operations/start.rs index 02be4f1dc..0cb804645 100644 --- a/lib/crates/fabro-workflows/src/operations/start.rs +++ b/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, - services: StartServices, + mut services: StartServices, ) -> Result { + 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()); diff --git a/lib/crates/fabro-workflows/src/pipeline/execute.rs b/lib/crates/fabro-workflows/src/pipeline/execute.rs index 0da5fc54f..4c35874ac 100644 --- a/lib/crates/fabro-workflows/src/pipeline/execute.rs +++ b/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, diff --git a/lib/crates/fabro-workflows/src/pipeline/finalize.rs b/lib/crates/fabro-workflows/src/pipeline/finalize.rs index b79239ed3..d6edb30b3 100644 --- a/lib/crates/fabro-workflows/src/pipeline/finalize.rs +++ b/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 = 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, diff --git a/lib/crates/fabro-workflows/src/pipeline/initialize.rs b/lib/crates/fabro-workflows/src/pipeline/initialize.rs index 9e8b54075..a1e5fb06b 100644 --- a/lib/crates/fabro-workflows/src/pipeline/initialize.rs +++ b/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, diff --git a/lib/crates/fabro-workflows/src/pipeline/retro.rs b/lib/crates/fabro-workflows/src/pipeline/retro.rs index 1d95bc0df..482ff7bed 100644 --- a/lib/crates/fabro-workflows/src/pipeline/retro.rs +++ b/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 { &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 { 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 { - 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 = Arc::new( + fabro_store::DiskProjectingRunStore::new(inner, run_dir.to_path_buf()), + ); run_store.put_checkpoint(checkpoint).await.unwrap(); run_store } diff --git a/lib/crates/fabro-workflows/src/run_dir.rs b/lib/crates/fabro-workflows/src/run_dir.rs index f31d87eca..a1028e1c6 100644 --- a/lib/crates/fabro-workflows/src/run_dir.rs +++ b/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)); - } } diff --git a/lib/crates/fabro-workflows/src/test_support.rs b/lib/crates/fabro-workflows/src/test_support.rs index d860f3122..9806f9352 100644 --- a/lib/crates/fabro-workflows/src/test_support.rs +++ b/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(),