From 93eab718927ae336b09855dcd235bc71ed2cff0e Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Wed, 1 Apr 2026 22:23:22 -0700 Subject: [PATCH] Remove legacy run file projection layer --- .../fabro-cli/src/commands/run/attach.rs | 21 +- lib/crates/fabro-cli/src/commands/run/diff.rs | 8 +- .../fabro-cli/src/commands/run/output.rs | 6 +- .../fabro-cli/src/commands/run/rewind.rs | 15 +- lib/crates/fabro-cli/src/commands/runs/rm.rs | 10 +- lib/crates/fabro-cli/tests/it/cmd/attach.rs | 4 +- lib/crates/fabro-cli/tests/it/cmd/create.rs | 55 +- lib/crates/fabro-cli/tests/it/cmd/detached.rs | 23 +- lib/crates/fabro-cli/tests/it/cmd/diff.rs | 29 +- lib/crates/fabro-cli/tests/it/cmd/inspect.rs | 2 +- lib/crates/fabro-cli/tests/it/cmd/logs.rs | 2 +- lib/crates/fabro-cli/tests/it/cmd/resume.rs | 2 +- lib/crates/fabro-cli/tests/it/cmd/rewind.rs | 2 +- lib/crates/fabro-cli/tests/it/cmd/rm.rs | 2 +- lib/crates/fabro-cli/tests/it/cmd/run.rs | 576 +---------- lib/crates/fabro-cli/tests/it/cmd/start.rs | 2 +- lib/crates/fabro-cli/tests/it/cmd/support.rs | 96 +- lib/crates/fabro-retro/src/lib.rs | 2 - lib/crates/fabro-retro/src/retro.rs | 54 +- lib/crates/fabro-retro/src/retro_agent.rs | 137 +-- lib/crates/fabro-server/src/server.rs | 51 +- lib/crates/fabro-store/src/disk_projecting.rs | 955 ------------------ lib/crates/fabro-store/src/lib.rs | 2 - .../fabro-workflow/src/handler/agent.rs | 7 - .../fabro-workflow/src/handler/prompt.rs | 7 - .../fabro-workflow/src/lifecycle/git.rs | 6 - .../fabro-workflow/src/operations/create.rs | 16 +- .../fabro-workflow/src/operations/resume.rs | 7 +- .../fabro-workflow/src/operations/start.rs | 52 +- .../src/pipeline/execute/tests.rs | 151 +-- .../fabro-workflow/src/pipeline/finalize.rs | 106 +- .../fabro-workflow/src/pipeline/initialize.rs | 12 +- lib/crates/fabro-workflow/src/pipeline/mod.rs | 3 +- .../fabro-workflow/src/pipeline/persist.rs | 223 ++-- .../src/pipeline/pull_request.rs | 25 +- .../fabro-workflow/src/pipeline/retro.rs | 24 +- .../fabro-workflow/src/pipeline/types.rs | 5 - lib/crates/fabro-workflow/src/run_lookup.rs | 28 +- lib/crates/fabro-workflow/src/test_support.rs | 7 +- 39 files changed, 488 insertions(+), 2247 deletions(-) delete mode 100644 lib/crates/fabro-store/src/disk_projecting.rs diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index db80e428b..35121d849 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -6,7 +6,6 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant}; use anyhow::Result; -use fabro_config::FabroSettingsExt; use fabro_types::RunId; use futures::StreamExt; @@ -14,7 +13,7 @@ use fabro_interview::{AnswerValue, ConsoleInterviewer}; use fabro_store::{EventEnvelope, RunStore, RuntimeState}; use fabro_util::terminal::Styles; use fabro_workflow::outcome::StageStatus; -use fabro_workflow::records::{Conclusion, ConclusionExt, RunRecord, RunRecordExt}; +use fabro_workflow::records::{Conclusion, ConclusionExt}; use fabro_workflow::run_status::{RunStatus, RunStatusRecord, RunStatusRecordExt}; use serde_json::{Map, Value}; use tokio::signal::ctrl_c; @@ -44,18 +43,10 @@ pub(crate) async fn attach_run( engine_child: Option, json_output: bool, ) -> Result { - let run_record = RunRecord::load(run_dir).ok(); let inferred_storage_dir = infer_storage_dir(run_dir); - let fallback_storage_dir = run_record - .as_ref() - .map(|record| record.settings.storage_dir()) - .or(inferred_storage_dir); let inferred_run_id = infer_run_id(run_dir); - let storage_dir = storage_dir.map(Path::to_path_buf).or(fallback_storage_dir); - let run_id = run_id - .copied() - .or_else(|| run_record.as_ref().map(|record| record.run_id)) - .or(inferred_run_id); + let storage_dir = storage_dir.map(Path::to_path_buf).or(inferred_storage_dir); + let run_id = run_id.copied().or(inferred_run_id); if let (Some(storage_dir), Some(run_id)) = (storage_dir.as_deref(), run_id.as_ref()) { match store::open_run_reader(storage_dir, run_id).await { @@ -104,13 +95,9 @@ pub(crate) async fn attach_run( } } - let verbose = run_record - .as_ref() - .map(|record| record.settings.verbose_enabled()) - .unwrap_or(false); attach_run_files( run_dir, - verbose, + false, kill_on_detach, styles, engine_child, diff --git a/lib/crates/fabro-cli/src/commands/run/diff.rs b/lib/crates/fabro-cli/src/commands/run/diff.rs index aecaeab08..e0b7cbbc5 100644 --- a/lib/crates/fabro-cli/src/commands/run/diff.rs +++ b/lib/crates/fabro-cli/src/commands/run/diff.rs @@ -54,7 +54,7 @@ pub(crate) async fn run(args: DiffArgs, globals: &GlobalArgs) -> Result<()> { } async fn resolve_diff( - run_dir: &Path, + _run_dir: &Path, run_store: &dyn fabro_store::RunStore, args: &DiffArgs, ) -> Result { @@ -73,11 +73,7 @@ async fn resolve_diff( } } - debug!(node_id, "Reading per-node diff"); - let node_patch = run_dir.join("nodes").join(node_id).join("diff.patch"); - return std::fs::read_to_string(&node_patch).with_context(|| { - format!("No diff found for node '{node_id}' — check the node ID and try again") - }); + bail!("No diff found for node '{node_id}' — check the node ID and try again"); } let start = run_store diff --git a/lib/crates/fabro-cli/src/commands/run/output.rs b/lib/crates/fabro-cli/src/commands/run/output.rs index 8f616b36a..cc9e5aa1b 100644 --- a/lib/crates/fabro-cli/src/commands/run/output.rs +++ b/lib/crates/fabro-cli/src/commands/run/output.rs @@ -10,7 +10,7 @@ use fabro_util::text::strip_goal_decoration; use fabro_workflow::asset_snapshot::collect_asset_paths; use fabro_workflow::outcome::{StageStatus, format_cost}; use fabro_workflow::pipeline::{Persisted, Validated}; -use fabro_workflow::records::{Checkpoint, CheckpointExt, Conclusion}; +use fabro_workflow::records::Conclusion; use indicatif::HumanDuration; use crate::shared::{format_tokens_human, print_diagnostics, relative_path, tilde_path}; @@ -204,12 +204,12 @@ pub(crate) fn print_run_conclusion( pub(crate) async fn print_final_output( run_store: Option<&dyn fabro_store::RunStore>, - run_dir: &Path, + _run_dir: &Path, styles: &Styles, ) { let checkpoint = match run_store { Some(run_store) => run_store.get_checkpoint().await.ok().flatten(), - None => Checkpoint::load(&run_dir.join("checkpoint.json")).ok(), + None => None, }; let Some(checkpoint) = checkpoint else { return; diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index c42ac6387..045c00f6d 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs @@ -10,7 +10,6 @@ use fabro_workflow::operations::{ RewindInput, RewindTarget, RunTimeline, build_timeline_or_rebuild, find_run_id_by_prefix_or_store, rewind, }; -use fabro_workflow::records::CheckpointExt; use fabro_workflow::records::{RunRecord, RunRecordExt, StartRecord, StartRecordExt}; use fabro_workflow::run_lookup::{resolve_run_combined, runs_base}; use fabro_workflow::run_status::RunStatus; @@ -132,16 +131,8 @@ async fn reset_rewound_run_state( .context("failed to restore run record after rewind: missing run metadata")?; let checkpoint = MetadataStore::read_checkpoint(git_store.repo_dir(), &run_id.to_string())? .context("rewound metadata branch is missing checkpoint.json")?; - checkpoint.save(&run_dir.join("checkpoint.json"))?; - for name in [ - "conclusion.json", - "pull_request.json", - "detached_failure.json", - "progress.jsonl", - "retro.json", - "final.patch", - ] { + for name in ["detached_failure.json"] { let _ = std::fs::remove_file(run_dir.join(name)); } @@ -164,9 +155,7 @@ async fn reset_rewound_run_state( .await .map_err(|err| anyhow::anyhow!("failed to restore start record after rewind: {err}"))?; } - if let Some(dot_source) = - store_graph.or_else(|| std::fs::read_to_string(run_dir.join("workflow.fabro")).ok()) - { + if let Some(dot_source) = store_graph { run_store .put_graph(&dot_source) .await diff --git a/lib/crates/fabro-cli/src/commands/runs/rm.rs b/lib/crates/fabro-cli/src/commands/runs/rm.rs index cd9e910ad..f8216c403 100644 --- a/lib/crates/fabro-cli/src/commands/runs/rm.rs +++ b/lib/crates/fabro-cli/src/commands/runs/rm.rs @@ -2,14 +2,13 @@ use std::path::Path; use anyhow::{Context, Result, bail}; use fabro_config::FabroSettingsExt; -use fabro_sandbox::SandboxRecordExt; use fabro_store::Store; use tracing::warn; use fabro_sandbox::reconnect::reconnect as reconnect_sandbox; use fabro_workflow::run_lookup::RunInfo; use fabro_workflow::run_lookup::{resolve_run_combined, runs_base}; -use fabro_workflow::run_status::{RunStatus, RunStatusRecord, write_run_status}; +use fabro_workflow::run_status::{RunStatus, RunStatusRecord}; use crate::args::{GlobalArgs, RunsRemoveArgs}; use crate::shared::print_json_pretty; @@ -116,7 +115,6 @@ pub(crate) async fn remove_run_with_cleanup(store: &dyn Store, run: &RunInfo) -> } async fn remove_run_dir_with_cleanup(store: &dyn Store, run: &RunInfo) -> Result<()> { - write_run_status(&run.path, RunStatus::Removing, None); let run_store = match store.open_run_reader(&run.run_id).await { Ok(run_store) => run_store, Err(err) => { @@ -168,7 +166,7 @@ async fn delete_run_store_state(store: &dyn Store, run: &RunInfo) -> Result<()> } async fn load_sandbox_record( - run_dir: &Path, + _run_dir: &Path, run_store: Option<&dyn fabro_store::RunStore>, ) -> Option { if let Some(run_store) = run_store { @@ -180,7 +178,5 @@ async fn load_sandbox_record( } } } - - let sandbox_path = run_dir.join("sandbox.json"); - fabro_sandbox::SandboxRecord::load(&sandbox_path).ok() + None } diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index 750652c86..9985b069a 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -125,8 +125,8 @@ fn attach_replays_from_store_without_run_json_or_progress_jsonl() { .success(); let run = resolve_run(&context, run_id); - std::fs::remove_file(run.run_dir.join("run.json")).unwrap(); - std::fs::remove_file(run.run_dir.join("progress.jsonl")).unwrap(); + let _ = std::fs::remove_file(run.run_dir.join("run.json")); + let _ = std::fs::remove_file(run.run_dir.join("progress.jsonl")); let mut cmd = context.command(); cmd.args(["attach", run_id]); diff --git a/lib/crates/fabro-cli/tests/it/cmd/create.rs b/lib/crates/fabro-cli/tests/it/cmd/create.rs index 8f6835b44..4b0d79c31 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/create.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/create.rs @@ -3,9 +3,9 @@ use serde_json::json; use fabro_test::{fabro_snapshot, test_context}; -use crate::support::{fabro_json_snapshot, read_json}; +use crate::support::fabro_json_snapshot; -use super::support::{fixture, output_stdout, resolve_run}; +use super::support::{fixture, output_stdout, resolve_run, run_snapshot}; #[test] fn help() { @@ -77,14 +77,13 @@ digraph BarBaz { .success(); let run_dir = context.find_run_dir(run_id); - let run_record = read_json(run_dir.join("run.json")); - let cached_graph = std::fs::read_to_string(run_dir.join("workflow.fabro")).unwrap(); + let snapshot = run_snapshot(&run_dir); fabro_json_snapshot!( context, serde_json::json!({ - "workflow_slug": run_record["workflow_slug"], - "graph_name": run_record["graph"]["name"], - "cached_graph_lines": cached_graph.lines().collect::>(), + "workflow_slug": snapshot.run.workflow_slug, + "graph_name": snapshot.run.graph.name, + "cached_graph_lines": snapshot.graph.expect("graph should exist").lines().collect::>(), }), @r#" { @@ -133,14 +132,13 @@ digraph FooWorkflow { .success(); let run_dir = context.find_run_dir(run_id); - let run_record = read_json(run_dir.join("run.json")); - let cached_graph = std::fs::read_to_string(run_dir.join("workflow.fabro")).unwrap(); + let snapshot = run_snapshot(&run_dir); fabro_json_snapshot!( context, serde_json::json!({ - "workflow_slug": run_record["workflow_slug"], - "graph_name": run_record["graph"]["name"], - "cached_graph_lines": cached_graph.lines().collect::>(), + "workflow_slug": snapshot.run.workflow_slug, + "graph_name": snapshot.run.graph.name, + "cached_graph_lines": snapshot.graph.expect("graph should exist").lines().collect::>(), }), @r#" { @@ -159,7 +157,7 @@ digraph FooWorkflow { } #[test] -fn create_persists_requested_overrides_into_run_json() { +fn create_persists_requested_overrides_into_store() { let context = test_context!(); let workflow = fixture("simple.fabro"); let mut cmd = context.command(); @@ -200,26 +198,26 @@ fn create_persists_requested_overrides_into_run_json() { .expect("create should print a run ID") .to_string(); let run = resolve_run(&context, &run_id); - let run_json = read_json(run.run_dir.join("run.json")); + let snapshot = run_snapshot(&run.run_dir); let labels = json!({ - "env": run_json.pointer("/labels/env"), - "team": run_json.pointer("/labels/team"), + "env": snapshot.run.labels.get("env"), + "team": snapshot.run.labels.get("team"), }); let compact = json!({ - "workflow_slug": run_json["workflow_slug"], + "workflow_slug": snapshot.run.workflow_slug, "settings": { - "goal": run_json.pointer("/settings/goal"), - "dry_run": run_json.pointer("/settings/dry_run"), - "auto_approve": run_json.pointer("/settings/auto_approve"), - "no_retro": run_json.pointer("/settings/no_retro"), - "verbose": run_json.pointer("/settings/verbose"), + "goal": snapshot.run.settings.goal, + "dry_run": snapshot.run.settings.dry_run, + "auto_approve": snapshot.run.settings.auto_approve, + "no_retro": snapshot.run.settings.no_retro, + "verbose": snapshot.run.settings.verbose, "llm": { - "model": run_json.pointer("/settings/llm/model"), - "provider": run_json.pointer("/settings/llm/provider"), + "model": snapshot.run.settings.llm.as_ref().and_then(|llm| llm.model.clone()), + "provider": snapshot.run.settings.llm.as_ref().and_then(|llm| llm.provider.clone()), }, "sandbox": { - "provider": run_json.pointer("/settings/sandbox/provider"), - "preserve": run_json.pointer("/settings/sandbox/preserve"), + "provider": snapshot.run.settings.sandbox.as_ref().and_then(|sandbox| sandbox.provider.clone()), + "preserve": snapshot.run.settings.sandbox.as_ref().and_then(|sandbox| sandbox.preserve), }, }, "labels": labels, @@ -274,11 +272,10 @@ fn create_json_implies_auto_approve() { .as_str() .expect("create JSON should include run_id"); let run = resolve_run(&context, run_id); - let run_json = read_json(run.run_dir.join("run.json")); assert_eq!( - run_json.pointer("/settings/auto_approve"), - Some(&json!(true)) + run_snapshot(&run.run_dir).run.settings.auto_approve, + Some(true) ); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/detached.rs b/lib/crates/fabro-cli/tests/it/cmd/detached.rs index b6d7f11fb..751fcf403 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/detached.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/detached.rs @@ -1,6 +1,7 @@ use fabro_test::{fabro_snapshot, test_context}; -use crate::support::{fabro_json_snapshot, read_json}; +use super::support::run_snapshot; +use crate::support::fabro_json_snapshot; #[test] fn help() { @@ -86,7 +87,12 @@ digraph CachedGraph { .assert() .success(); - let conclusion = read_json(run_dir.join("conclusion.json")); + let conclusion = serde_json::to_value( + run_snapshot(&run_dir) + .conclusion + .expect("conclusion should exist"), + ) + .unwrap(); fabro_json_snapshot!( context, serde_json::json!({ @@ -140,11 +146,11 @@ digraph GitHubApp { .success(); let run_dir = context.find_run_dir(run_id); - let run_record = read_json(run_dir.join("run.json")); + let snapshot = run_snapshot(&run_dir); fabro_json_snapshot!( context, serde_json::json!({ - "app_id": run_record["settings"]["git"]["app_id"], + "app_id": snapshot.run.settings.git.and_then(|git| git.app_id), }), @r#" { @@ -207,8 +213,6 @@ digraph DetachedStoreOnly { .success(); let run_dir = context.find_run_dir(run_id); - std::fs::remove_file(run_dir.join("run.json")).unwrap(); - context .command() .args([ @@ -224,7 +228,12 @@ digraph DetachedStoreOnly { .assert() .success(); - let conclusion = read_json(run_dir.join("conclusion.json")); + let conclusion = serde_json::to_value( + run_snapshot(&run_dir) + .conclusion + .expect("conclusion should exist"), + ) + .unwrap(); fabro_json_snapshot!( context, serde_json::json!({ diff --git a/lib/crates/fabro-cli/tests/it/cmd/diff.rs b/lib/crates/fabro-cli/tests/it/cmd/diff.rs index 74567e26d..096cf5bfa 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/diff.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/diff.rs @@ -115,8 +115,14 @@ fn diff_completed_run_reads_store_final_patch_without_disk_file() { let context = test_context!(); let setup = setup_git_backed_changed_run(&context); let run_id: RunId = setup.run.run_id.parse().unwrap(); - let patch = std::fs::read_to_string(setup.run.run_dir.join("final.patch")).unwrap(); - std::fs::remove_file(setup.run.run_dir.join("final.patch")).unwrap(); + let patch = with_runtime(|runtime| { + runtime.block_on(async { + let store = build_store(&context.storage_dir); + let run_store = store.open_run(&run_id).await.unwrap().unwrap(); + run_store.get_final_patch().await.unwrap().unwrap() + }) + }); + let _ = std::fs::remove_file(setup.run.run_dir.join("final.patch")); with_runtime(|runtime| { runtime.block_on(async { @@ -172,9 +178,22 @@ fn diff_node_reads_store_patch_without_disk_file() { let context = test_context!(); let setup = setup_git_backed_changed_run(&context); let run_id: RunId = setup.run.run_id.parse().unwrap(); - let patch = - std::fs::read_to_string(setup.run.run_dir.join("nodes/step_one/diff.patch")).unwrap(); - std::fs::remove_file(setup.run.run_dir.join("nodes/step_one/diff.patch")).unwrap(); + let patch = with_runtime(|runtime| { + runtime.block_on(async { + let store = build_store(&context.storage_dir); + let run_store = store.open_run(&run_id).await.unwrap().unwrap(); + run_store + .get_node(&fabro_store::NodeVisitRef { + node_id: "step_one", + visit: 1, + }) + .await + .unwrap() + .diff + .unwrap() + }) + }); + let _ = std::fs::remove_file(setup.run.run_dir.join("nodes/step_one/diff.patch")); with_runtime(|runtime| { runtime.block_on(async { diff --git a/lib/crates/fabro-cli/tests/it/cmd/inspect.rs b/lib/crates/fabro-cli/tests/it/cmd/inspect.rs index 732d71f52..fd1daf2b0 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/inspect.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/inspect.rs @@ -116,7 +116,7 @@ fn inspect_completed_run_reads_store_without_disk_metadata_files() { "checkpoint.json", "sandbox.json", ] { - std::fs::remove_file(run.run_dir.join(name)).unwrap(); + let _ = std::fs::remove_file(run.run_dir.join(name)); } let output = run_success(&context, &["inspect", &run.run_id]); diff --git a/lib/crates/fabro-cli/tests/it/cmd/logs.rs b/lib/crates/fabro-cli/tests/it/cmd/logs.rs index 315c0cef0..bb199a9d7 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/logs.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/logs.rs @@ -92,7 +92,7 @@ fn logs_completed_run_outputs_raw_ndjson() { fn logs_completed_run_reads_store_without_progress_jsonl() { let context = test_context!(); let run = setup_completed_dry_run(&context); - std::fs::remove_file(run.run_dir.join("progress.jsonl")).unwrap(); + let _ = std::fs::remove_file(run.run_dir.join("progress.jsonl")); let mut filters = context.filters(); filters.push(( diff --git a/lib/crates/fabro-cli/tests/it/cmd/resume.rs b/lib/crates/fabro-cli/tests/it/cmd/resume.rs index f75f07a92..4a2cf84c1 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/resume.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/resume.rs @@ -71,7 +71,7 @@ fn resume_rewound_run_succeeds() { &setup.repo_dir, &["rev-parse", &format!("fabro/run/{}", setup.run.run_id)], ); - std::fs::remove_file(setup.run.run_dir.join("run.json")).unwrap(); + let _ = std::fs::remove_file(setup.run.run_dir.join("run.json")); let mut resume_cmd = context.command(); resume_cmd.current_dir(&setup.repo_dir); diff --git a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs index 0b59449e8..0e79c3a3f 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs @@ -85,7 +85,7 @@ fn rewind_target_updates_metadata_and_resume_hint() { let mut cmd = context.command(); cmd.current_dir(&setup.repo_dir); - std::fs::remove_file(setup.run.run_dir.join("run.json")).unwrap(); + let _ = std::fs::remove_file(setup.run.run_dir.join("run.json")); cmd.args(["rewind", &setup.run.run_id, "@1", "--no-push"]); let (snapshot, output) = run_and_format(&mut cmd, &git_filters(&context)); diff --git a/lib/crates/fabro-cli/tests/it/cmd/rm.rs b/lib/crates/fabro-cli/tests/it/cmd/rm.rs index 70030259e..2bb393990 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/rm.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/rm.rs @@ -122,7 +122,7 @@ fn rm_force_deletes_submitted_run() { fn rm_force_deletes_run_without_sandbox_json_when_store_has_sandbox() { let context = test_context!(); let setup = setup_local_sandbox_run(&context); - std::fs::remove_file(setup.run.run_dir.join("sandbox.json")).unwrap(); + let _ = std::fs::remove_file(setup.run.run_dir.join("sandbox.json")); let mut filters = context.filters(); filters.push(( diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 36c840d3a..2c79d65b1 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -1,9 +1,7 @@ use fabro_test::{fabro_snapshot, test_context}; use serde_json::Value; -use crate::support::{ - example_fixture, fabro_json_snapshot, read_json, read_jsonl, run_output_filters, -}; +use crate::support::{example_fixture, fabro_json_snapshot, run_output_filters}; #[test] fn help() { @@ -77,7 +75,7 @@ fn dry_run_simple() { } #[test] -fn dry_run_writes_jsonl_and_live_json() { +fn dry_run_persists_event_history_in_store() { let context = test_context!(); let run_id = "01ARZ3NDEKTSV4RRFFQ69G5FB8"; @@ -96,521 +94,61 @@ fn dry_run_writes_jsonl_and_live_json() { .assert() .success(); - let run_dir = context.find_run_dir(run_id); - let jsonl_path = run_dir.join("progress.jsonl"); - let progress = read_jsonl(&jsonl_path); + context.find_run_dir(run_id); + let output = context + .command() + .args(["logs", run_id]) + .output() + .expect("logs command should execute"); + assert!( + output.status.success(), + "logs failed:\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + let progress: Vec = String::from_utf8(output.stdout) + .expect("stdout should be UTF-8") + .lines() + .filter(|line| !line.trim().is_empty()) + .map(|line| serde_json::from_str(line).expect("logs output should be JSONL")) + .collect(); assert!( !progress.is_empty(), - "progress.jsonl should have at least one line" + "store-backed event history should have at least one line" + ); + assert_eq!( + progress.first().and_then(|event| event["event"].as_str()), + Some("run.created") + ); + assert_eq!( + progress + .first() + .and_then(|event| event.pointer("/properties/settings/auto_approve")) + .and_then(Value::as_bool), + Some(true) + ); + assert_eq!( + progress.last().and_then(|event| event["event"].as_str()), + Some("sandbox.cleanup.completed") ); - fabro_json_snapshot!(context, &progress, @r#" - [ - { - "id": "[EVENT_ID]", - "ts": "[TIMESTAMP]", - "run_id": "[ULID]", - "event": "run.created", - "properties": { - "workflow_slug": "simple", - "settings": { - "auto_approve": true, - "dry_run": true, - "fabro": { - "root": "fabro/" - }, - "features": { - "retros": false, - "session_sandboxes": false - }, - "goal": "Run tests and report results", - "hooks": [ - { - "blocking": true, - "command": "cargo fmt", - "event": "post_tool_use", - "matcher": "write_file|edit_file|apply_patch", - "name": "cargo-fmt", - "sandbox": null, - "timeout_ms": null - } - ], - "llm": { - "fallbacks": null, - "model": "claude-sonnet-4-6", - "provider": "anthropic" - }, - "mode": "standalone", - "pull_request": { - "auto_merge": false, - "draft": false, - "enabled": true, - "merge_strategy": "squash" - }, - "sandbox": { - "daytona": { - "auto_stop_interval": 30, - "labels": { - "repo": "fabro-sh/fabro" - }, - "network": null, - "skip_clone": false, - "snapshot": { - "cpu": 4, - "disk": 20, - "dockerfile": "FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n", - "memory": 8, - "name": "fabro-v6" - } - }, - "devcontainer": null, - "env": null, - "local": null, - "preserve": null, - "provider": "local" - }, - "storage_dir": "[STORAGE_DIR]", - "version": 1 - }, - "graph": { - "attrs": { - "goal": { - "String": "Run tests and report results" - }, - "rankdir": { - "String": "LR" - } - }, - "edges": [ - { - "attrs": {}, - "from": "start", - "to": "run_tests" - }, - { - "attrs": {}, - "from": "run_tests", - "to": "report" - }, - { - "attrs": {}, - "from": "report", - "to": "exit" - } - ], - "name": "Simple", - "nodes": { - "exit": { - "attrs": { - "label": { - "String": "Exit" - }, - "shape": { - "String": "Msquare" - } - }, - "id": "exit" - }, - "report": { - "attrs": { - "label": { - "String": "Report" - }, - "prompt": { - "String": "Summarize the test results" - } - }, - "id": "report" - }, - "run_tests": { - "attrs": { - "label": { - "String": "Run Tests" - }, - "prompt": { - "String": "Run the test suite and report results" - } - }, - "id": "run_tests" - }, - "start": { - "attrs": { - "label": { - "String": "Start" - }, - "shape": { - "String": "Mdiamond" - } - }, - "id": "start" - } - } - }, - "workflow_source": "digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n", - "labels": {}, - "run_dir": "[DRY_RUN_DIR]", - "working_directory": "[TEMP_DIR]", - "host_repo_path": "[TEMP_DIR]" - } - }, - { - "event": "sandbox.initializing", - "id": "[EVENT_ID]", - "properties": { - "provider": "local" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "sandbox.ready", - "id": "[EVENT_ID]", - "properties": { - "cpu": null, - "duration_ms": "[DURATION_MS]", - "memory": null, - "name": null, - "provider": "local", - "url": null - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "sandbox.initialized", - "id": "[EVENT_ID]", - "properties": { - "provider": "local", - "working_directory": "[TEMP_DIR]" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "run.started", - "id": "[EVENT_ID]", - "properties": { - "goal": "Run tests and report results", - "name": "Simple" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "stage.started", - "id": "[EVENT_ID]", - "node_id": "start", - "node_label": "Start", - "properties": { - "attempt": 1, - "handler_type": "start", - "index": 0, - "max_attempts": 1 - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "stage.completed", - "id": "[EVENT_ID]", - "node_id": "start", - "node_label": "Start", - "properties": { - "attempt": 1, - "context_values": { - "current.preamble": "Goal: Run tests and report results/n", - "current_node": "start", - "graph.goal": "Run tests and report results", - "graph.rankdir": "LR", - "internal.fidelity": "compact", - "internal.node_visit_count": 1, - "internal.run_id": "[ULID]", - "internal.thread_id": null - }, - "duration_ms": "[DURATION_MS]", - "files_touched": [], - "index": 0, - "max_attempts": 1, - "node_visits": { - "start": 1 - }, - "notes": "[Simulated] start", - "preferred_label": null, - "status": "success", - "suggested_next_ids": [], - "usage": null - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "edge.selected", - "id": "[EVENT_ID]", - "properties": { - "condition": null, - "from_node": "start", - "is_jump": false, - "label": null, - "reason": "unconditional", - "stage_status": "success", - "to_node": "run_tests" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "checkpoint.completed", - "id": "[EVENT_ID]", - "node_id": "start", - "node_label": "start", - "properties": { - "status": "success" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "stage.started", - "id": "[EVENT_ID]", - "node_id": "run_tests", - "node_label": "Run Tests", - "properties": { - "attempt": 1, - "handler_type": "agent", - "index": 1, - "max_attempts": 1 - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "stage.completed", - "id": "[EVENT_ID]", - "node_id": "run_tests", - "node_label": "Run Tests", - "properties": { - "attempt": 1, - "context_updates": { - "last_response": "[Simulated] Response for stage: run_tests", - "last_stage": "run_tests", - "response.run_tests": "[Simulated] Response for stage: run_tests" - }, - "context_values": { - "current.preamble": "Goal: Run tests and report results/n", - "current_node": "run_tests", - "failure_class": "", - "failure_signature": "", - "graph.goal": "Run tests and report results", - "graph.rankdir": "LR", - "internal.fidelity": "compact", - "internal.node_visit_count": 1, - "internal.retry_count.start": 0, - "internal.run_id": "[ULID]", - "internal.thread_id": "start", - "outcome": "success", - "thread.start.current_node": "run_tests" - }, - "duration_ms": "[DURATION_MS]", - "files_touched": [], - "index": 1, - "max_attempts": 1, - "node_visits": { - "run_tests": 1, - "start": 1 - }, - "notes": "[Simulated] run_tests", - "preferred_label": null, - "status": "success", - "suggested_next_ids": [], - "usage": null - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "edge.selected", - "id": "[EVENT_ID]", - "properties": { - "condition": null, - "from_node": "run_tests", - "is_jump": false, - "label": null, - "reason": "unconditional", - "stage_status": "success", - "to_node": "report" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "checkpoint.completed", - "id": "[EVENT_ID]", - "node_id": "run_tests", - "node_label": "run_tests", - "properties": { - "status": "success" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "stage.started", - "id": "[EVENT_ID]", - "node_id": "report", - "node_label": "Report", - "properties": { - "attempt": 1, - "handler_type": "agent", - "index": 2, - "max_attempts": 1 - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "stage.completed", - "id": "[EVENT_ID]", - "node_id": "report", - "node_label": "Report", - "properties": { - "attempt": 1, - "context_updates": { - "last_response": "[Simulated] Response for stage: report", - "last_stage": "report", - "response.report": "[Simulated] Response for stage: report" - }, - "context_values": { - "current.preamble": "Goal: Run tests and report results/n/n## Completed stages/n- **run_tests**: success/n", - "current_node": "report", - "failure_class": "", - "failure_signature": "", - "graph.goal": "Run tests and report results", - "graph.rankdir": "LR", - "internal.fidelity": "compact", - "internal.node_visit_count": 1, - "internal.retry_count.run_tests": 0, - "internal.retry_count.start": 0, - "internal.run_id": "[ULID]", - "internal.thread_id": "run_tests", - "last_response": "[Simulated] Response for stage: run_tests", - "last_stage": "run_tests", - "outcome": "success", - "response.run_tests": "[Simulated] Response for stage: run_tests", - "thread.run_tests.current_node": "report", - "thread.start.current_node": "run_tests" - }, - "duration_ms": "[DURATION_MS]", - "files_touched": [], - "index": 2, - "max_attempts": 1, - "node_visits": { - "report": 1, - "run_tests": 1, - "start": 1 - }, - "notes": "[Simulated] report", - "preferred_label": null, - "status": "success", - "suggested_next_ids": [], - "usage": null - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "edge.selected", - "id": "[EVENT_ID]", - "properties": { - "condition": null, - "from_node": "report", - "is_jump": false, - "label": null, - "reason": "unconditional", - "stage_status": "success", - "to_node": "exit" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "checkpoint.completed", - "id": "[EVENT_ID]", - "node_id": "report", - "node_label": "report", - "properties": { - "status": "success" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "stage.started", - "id": "[EVENT_ID]", - "node_id": "exit", - "node_label": "Exit", - "properties": { - "attempt": 1, - "handler_type": "exit", - "index": 3, - "max_attempts": 1 - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "stage.completed", - "id": "[EVENT_ID]", - "node_id": "exit", - "node_label": "Exit", - "properties": { - "attempt": 1, - "duration_ms": "[DURATION_MS]", - "files_touched": [], - "index": 3, - "max_attempts": 1, - "notes": null, - "preferred_label": null, - "status": "success", - "suggested_next_ids": [], - "usage": null - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "run.completed", - "id": "[EVENT_ID]", - "properties": { - "artifact_count": 0, - "duration_ms": "[DURATION_MS]", - "status": "success" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "sandbox.cleanup.started", - "id": "[EVENT_ID]", - "properties": { - "provider": "local" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "sandbox.cleanup.completed", - "id": "[EVENT_ID]", - "properties": { - "duration_ms": "[DURATION_MS]", - "provider": "local" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - } - ] - "#); - let live_path = run_dir.join("live.json"); - let live_content = read_json(&live_path); + let tail_output = context + .command() + .args(["logs", "--tail", "1", run_id]) + .output() + .expect("tail logs command should execute"); + assert!( + tail_output.status.success(), + "tail logs failed:\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&tail_output.stdout), + String::from_utf8_lossy(&tail_output.stderr) + ); + let live_content: Value = String::from_utf8(tail_output.stdout) + .expect("stdout should be UTF-8") + .lines() + .find(|line| !line.trim().is_empty()) + .map(|line| serde_json::from_str(line).expect("tail logs output should be JSON")) + .expect("tail logs should include the latest event"); fabro_json_snapshot!(context, &live_content, @r#" { "event": "sandbox.cleanup.completed", @@ -645,9 +183,7 @@ fn run_id_passthrough_uses_provided_ulid() { .assert() .success(); - let run_dir = context.find_run_dir(run_id); - let run_record = read_json(run_dir.join("run.json")); - assert_eq!(run_record["run_id"].as_str(), Some(run_id)); + context.find_run_dir(run_id); } #[test] @@ -1219,10 +755,8 @@ fn json_run_implies_auto_approve_for_human_gates() { ] "#); - let run = context.single_run_dir(); - let run_json = read_json(run.join("run.json")); assert_eq!( - run_json.pointer("/settings/auto_approve"), + progress[0].pointer("/properties/settings/auto_approve"), Some(&serde_json::json!(true)) ); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/start.rs b/lib/crates/fabro-cli/tests/it/cmd/start.rs index ba6016635..912e232df 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/start.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/start.rs @@ -97,7 +97,7 @@ fn start_by_run_id_starts_created_run_without_run_json_or_status_json() { .success(); let run_dir = context.find_run_dir(run_id); - std::fs::remove_file(run_dir.join("run.json")).unwrap(); + let _ = std::fs::remove_file(run_dir.join("run.json")); context.command().args(["start", run_id]).assert().success(); let output = context diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index 0a6d5ab36..9a7ebb4ed 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -1,9 +1,13 @@ use std::collections::BTreeSet; use std::path::{Path, PathBuf}; use std::process::Output; +use std::sync::Arc; use std::time::{Duration, Instant}; +use fabro_store::{RunSnapshot, RunStore, SlateStore, Store}; use fabro_test::TestContext; +use fabro_types::RunId; +use object_store::local::LocalFileSystem; use serde_json::Value; use shlex::try_quote; @@ -162,10 +166,13 @@ pub(crate) fn setup_detached_dry_run(context: &TestContext) -> RunSetup { .to_string(); let run = resolve_run(context, &run_id); let deadline = Instant::now() + COMMAND_TIMEOUT; - while !run.run_dir.join("progress.jsonl").exists() { + while run_store(&run.run_dir) + .and_then(|store| block_on(store.list_events()).ok()) + .is_none_or(|events| events.is_empty()) + { assert!( Instant::now() < deadline, - "timed out waiting for progress.jsonl for {run_id}" + "timed out waiting for store events for {run_id}" ); std::thread::sleep(Duration::from_millis(50)); } @@ -281,8 +288,10 @@ worktree_mode = "never" let run = run_local_workflow(context, &workspace_dir, "run.toml"); assert!( - run.run_dir.join("sandbox.json").exists(), - "setup_local_sandbox_run should persist sandbox.json" + run_store(&run.run_dir) + .and_then(|store| block_on(store.get_sandbox()).ok()) + .flatten() + .is_some() ); WorkspaceRunSetup { run, workspace_dir } @@ -369,8 +378,10 @@ pub(crate) fn write_gated_workflow(path: &Path, name: &str, goal: &str) -> Workf pub(crate) fn wait_for_status(run_dir: &Path, expected: &[&str]) -> String { let deadline = Instant::now() + COMMAND_TIMEOUT; loop { - if let Some(status) = read_json_if_exists(&run_dir.join("status.json")) - .and_then(|value| value["status"].as_str().map(ToOwned::to_owned)) + if let Some(status) = run_store(run_dir) + .and_then(|store| block_on(store.get_status()).ok()) + .flatten() + .map(|record| record.status.to_string()) { if expected.iter().any(|candidate| *candidate == status) { return status; @@ -401,10 +412,7 @@ pub(crate) fn only_run(context: &TestContext) -> RunSetup { runs_dir.display() ); let run_dir = entries[0].clone(); - let run_id = read_json(&run_dir.join("run.json"))["run_id"] - .as_str() - .expect("run.json should include run_id") - .to_string(); + let run_id = infer_run_id(&run_dir); RunSetup { run_id, run_dir } } @@ -457,6 +465,42 @@ pub(crate) fn find_run_dir(storage_dir: &Path, run_id: &str) -> Option }) } +fn infer_run_id(run_dir: &Path) -> String { + if let Ok(id) = std::fs::read_to_string(run_dir.join("id.txt")) { + return id.trim().to_string(); + } + run_dir + .file_name() + .map(|name| name.to_string_lossy().to_string()) + .and_then(|name| name.rsplit('-').next().map(ToOwned::to_owned)) + .filter(|value| !value.is_empty()) + .expect("run directory name should contain run id suffix") +} + +fn block_on(future: impl std::future::Future) -> T { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(future) +} + +fn run_store(run_dir: &Path) -> Option> { + let runs_dir = run_dir.parent()?; + let storage_dir = runs_dir.parent()?; + let run_id: RunId = infer_run_id(run_dir).parse().ok()?; + let object_store = Arc::new(LocalFileSystem::new_with_prefix(storage_dir.join("store")).ok()?); + let store = Arc::new(SlateStore::new(object_store, "", Duration::from_millis(5))); + block_on(store.open_run_reader(&run_id)).ok().flatten() +} + +pub(crate) fn run_snapshot(run_dir: &Path) -> RunSnapshot { + run_store(run_dir) + .and_then(|store| block_on(store.get_snapshot()).ok()) + .flatten() + .expect("run store snapshot should exist") +} + pub(crate) fn git_stdout(repo_dir: &Path, args: &[&str]) -> String { stdout(&git_success(repo_dir, args)) } @@ -724,7 +768,12 @@ fn setup_git_backed_run(context: &TestContext, workflow: GitWorkflowKind) -> Git } let run = only_run(context); - let start = read_json(&run.run_dir.join("start.json")); + let start = serde_json::to_value( + run_snapshot(&run.run_dir) + .start + .expect("start record should exist"), + ) + .unwrap(); assert_eq!( start["run_branch"].as_str(), Some(format!("fabro/run/{}", run.run_id).as_str()) @@ -733,22 +782,27 @@ fn setup_git_backed_run(context: &TestContext, workflow: GitWorkflowKind) -> Git match workflow { GitWorkflowKind::Changed => { assert!( - run.run_dir.join("final.patch").exists(), - "changed git-backed run should emit final.patch" + run_snapshot(&run.run_dir).final_patch.is_some(), + "changed git-backed run should persist final patch in store" + ); + let snapshot = run_snapshot(&run.run_dir); + assert!( + snapshot + .nodes + .iter() + .any(|node| node.node_id == "step_one" && node.diff.is_some()) ); assert!( - run.run_dir.join("nodes/step_one/diff.patch").exists(), - "changed git-backed run should emit a diff for step_one" - ); - assert!( - run.run_dir.join("nodes/step_two/diff.patch").exists(), - "changed git-backed run should emit a diff for step_two" + snapshot + .nodes + .iter() + .any(|node| node.node_id == "step_two" && node.diff.is_some()) ); } GitWorkflowKind::Noop => { assert!( - !run.run_dir.join("final.patch").exists(), - "no-op git-backed run should not emit final.patch" + run_snapshot(&run.run_dir).final_patch.is_none(), + "no-op git-backed run should not persist final.patch" ); } } diff --git a/lib/crates/fabro-retro/src/lib.rs b/lib/crates/fabro-retro/src/lib.rs index 6100640af..5c630bf09 100644 --- a/lib/crates/fabro-retro/src/lib.rs +++ b/lib/crates/fabro-retro/src/lib.rs @@ -1,4 +1,2 @@ pub mod retro; pub mod retro_agent; - -pub use retro::RetroExt; diff --git a/lib/crates/fabro-retro/src/retro.rs b/lib/crates/fabro-retro/src/retro.rs index 034085aa3..72f6e37e4 100644 --- a/lib/crates/fabro-retro/src/retro.rs +++ b/lib/crates/fabro-retro/src/retro.rs @@ -1,11 +1,9 @@ -use std::collections::HashMap; -use std::path::Path; - use fabro_types::RunId; pub use fabro_types::retro::{ AggregateStats, FrictionKind, FrictionPoint, Learning, LearningCategory, OpenItem, OpenItemKind, Retro, RetroNarrative, SmoothnessRating, StageRetro, }; +use std::collections::HashMap; #[derive(Debug, Clone)] pub struct CompletedStage { @@ -20,56 +18,6 @@ pub struct CompletedStage { pub files_touched: Vec, } -pub trait RetroExt { - fn save(&self, run_dir: &Path) -> anyhow::Result<()>; - fn load(run_dir: &Path) -> anyhow::Result - where - Self: Sized; -} - -impl RetroExt for Retro { - fn save(&self, run_dir: &Path) -> anyhow::Result<()> { - let json = serde_json::to_string_pretty(self) - .map_err(|e| anyhow::anyhow!("retro serialize failed: {e}"))?; - std::fs::write(run_dir.join("retro.json"), json)?; - Ok(()) - } - - fn load(run_dir: &Path) -> anyhow::Result { - let data = std::fs::read_to_string(run_dir.join("retro.json"))?; - serde_json::from_str(&data).map_err(|e| anyhow::anyhow!("retro deserialize failed: {e}")) - } -} - -pub fn extract_stage_durations(run_dir: &Path) -> HashMap { - let mut durations = HashMap::new(); - let jsonl_path = run_dir.join("progress.jsonl"); - let Ok(data) = std::fs::read_to_string(&jsonl_path) else { - return durations; - }; - for line in data.lines() { - let Ok(envelope) = serde_json::from_str::(line) else { - continue; - }; - if envelope.get("event").and_then(|v| v.as_str()) != Some("stage.completed") { - continue; - } - let Some(name) = envelope.get("node_id").and_then(|v| v.as_str()) else { - continue; - }; - let Some(duration_ms) = envelope - .get("properties") - .and_then(serde_json::Value::as_object) - .and_then(|properties| properties.get("duration_ms")) - .and_then(serde_json::Value::as_u64) - else { - continue; - }; - durations.insert(name.to_string(), duration_ms); - } - durations -} - pub fn derive_retro( run_id: RunId, workflow_name: &str, diff --git a/lib/crates/fabro-retro/src/retro_agent.rs b/lib/crates/fabro-retro/src/retro_agent.rs index 5c87af9a2..fdf91b7cb 100644 --- a/lib/crates/fabro-retro/src/retro_agent.rs +++ b/lib/crates/fabro-retro/src/retro_agent.rs @@ -22,8 +22,8 @@ const RETRO_SYSTEM_PROMPT: &str = r"You are a workflow run retrospective analyst You have access to the run's data files: - `progress.jsonl` — the full event stream (stage starts/completions, agent tool calls, errors, retries) - `checkpoint.json` — final execution state with node outcomes -- `run.json` — run record with config, graph, and metadata (if available) -- `start.json` — start record with start time and git info (if available) +- `run.json` — run record with config, graph, and metadata +- `start.json` — start record with start time and git info ## Your task @@ -379,7 +379,7 @@ fn build_profile(provider: Provider, model: &str) -> Box { async fn upload_data_files( sandbox: &Arc, run_store: Option<&dyn RunStore>, - run_dir: &Path, + _run_dir: &Path, target_dir: &str, ) -> anyhow::Result<()> { // Create target directory @@ -388,37 +388,23 @@ async fn upload_data_files( .await .map_err(|e| anyhow::anyhow!("Failed to create retro data dir: {e}"))?; - // progress.jsonl — try store first, fall back to filesystem - let progress_content = if let Some(store) = run_store { - match store.list_events().await { - Ok(envelopes) => { - let lines: Vec = envelopes - .into_iter() - .filter_map(|env| serde_json::to_string(env.payload.as_value()).ok()) - .collect(); - if lines.is_empty() { - None - } else { - Some(lines.join("\n") + "\n") - } - } - Err(e) => { - tracing::debug!(error = %e, "Could not read events from store, falling back to filesystem"); - None - } - } - } else { - None + let Some(store) = run_store else { + anyhow::bail!("retro analysis now requires a run store"); }; - let progress_content = if progress_content.is_some() { - progress_content - } else { - let source = run_dir.join("progress.jsonl"); - if source.exists() { - Some(std::fs::read_to_string(&source)?) - } else { - None + + let progress_content = match store.list_events().await { + Ok(envelopes) => { + let lines: Vec = envelopes + .into_iter() + .filter_map(|env| serde_json::to_string(env.payload.as_value()).ok()) + .collect(); + if lines.is_empty() { + None + } else { + Some(lines.join("\n") + "\n") + } } + Err(e) => return Err(anyhow::anyhow!("Failed to load events from store: {e}")), }; if let Some(content) = progress_content { sandbox @@ -427,80 +413,39 @@ async fn upload_data_files( .map_err(|e| anyhow::anyhow!("Failed to upload progress.jsonl: {e}"))?; } - // checkpoint.json — try store first, fall back to filesystem - let checkpoint_content = if let Some(store) = run_store { - match store.get_checkpoint().await { - Ok(Some(cp)) => serde_json::to_string_pretty(&cp).ok(), - Ok(None) => None, - Err(e) => { - tracing::debug!(error = %e, "Could not read checkpoint from store, falling back to filesystem"); - None - } - } - } else { - None - }; - upload_file_with_fallback( - sandbox, - run_dir, - target_dir, - "checkpoint.json", - checkpoint_content, - ) - .await?; + let checkpoint_content = store + .get_checkpoint() + .await + .map_err(|e| anyhow::anyhow!("Failed to load checkpoint from store: {e}"))? + .map(|cp| serde_json::to_string_pretty(&cp)) + .transpose()?; + upload_file(sandbox, target_dir, "checkpoint.json", checkpoint_content).await?; - // run.json — try store first, fall back to filesystem - let run_content = if let Some(store) = run_store { - match store.get_run().await { - Ok(Some(run)) => serde_json::to_string_pretty(&run).ok(), - Ok(None) => None, - Err(e) => { - tracing::debug!(error = %e, "Could not read run from store, falling back to filesystem"); - None - } - } - } else { - None - }; - upload_file_with_fallback(sandbox, run_dir, target_dir, "run.json", run_content).await?; + let run_content = store + .get_run() + .await + .map_err(|e| anyhow::anyhow!("Failed to load run metadata from store: {e}"))? + .map(|run| serde_json::to_string_pretty(&run)) + .transpose()?; + upload_file(sandbox, target_dir, "run.json", run_content).await?; - // start.json — try store first, fall back to filesystem - let start_content = if let Some(store) = run_store { - match store.get_start().await { - Ok(Some(start)) => serde_json::to_string_pretty(&start).ok(), - Ok(None) => None, - Err(e) => { - tracing::debug!(error = %e, "Could not read start from store, falling back to filesystem"); - None - } - } - } else { - None - }; - upload_file_with_fallback(sandbox, run_dir, target_dir, "start.json", start_content).await?; + let start_content = store + .get_start() + .await + .map_err(|e| anyhow::anyhow!("Failed to load start metadata from store: {e}"))? + .map(|start| serde_json::to_string_pretty(&start)) + .transpose()?; + upload_file(sandbox, target_dir, "start.json", start_content).await?; Ok(()) } -/// Upload a single file to the sandbox. If `store_content` is `Some`, use it directly; -/// otherwise fall back to reading from `run_dir/filename` on the filesystem. -async fn upload_file_with_fallback( +async fn upload_file( sandbox: &Arc, - run_dir: &Path, target_dir: &str, filename: &str, - store_content: Option, + content: Option, ) -> anyhow::Result<()> { - let content = if store_content.is_some() { - store_content - } else { - let source = run_dir.join(filename); - if source.exists() { - Some(std::fs::read_to_string(&source)?) - } else { - None - } - }; if let Some(content) = content { sandbox .write_file(&format!("{target_dir}/{filename}"), &content) diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 1839da772..5d261ea83 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -20,7 +20,6 @@ use fabro_llm::types::{ ContentPart, FinishReason, Message as LlmMessage, Request as LlmRequest, Response as LlmResponse, Role, StreamEvent, ToolChoice, ToolDefinition, Usage, }; -use fabro_retro::retro::{Retro, extract_stage_durations}; use fabro_store::{InMemoryStore, Store}; use fabro_types::RunId; use fabro_util::redact::redact_jsonl_line; @@ -47,7 +46,6 @@ use crate::sessions as sessions_mod; use crate::sessions::{SessionStore, new_session_store}; use crate::web_auth; use fabro_interview::{Answer, Interviewer, QuestionType, WebInterviewer}; -use fabro_retro::RetroExt; use fabro_workflow::context::Context; use fabro_workflow::event::{EventEmitter, RunEventEnvelope}; use fabro_workflow::operations::{self, CreateRunInput, WorkflowInput}; @@ -771,7 +769,7 @@ async fn execute_run(state: Arc, run_id: RunId) { Ok(events) => fabro_workflow::extract_stage_durations_from_events(&events), Err(err) => { tracing::warn!(run_id = %run_id, error = %err, "Failed to load run events from store"); - extract_stage_durations(&run_dir) + Default::default() } }; let mut agg = state @@ -1559,43 +1557,26 @@ async fn get_retro( Ok(id) => id, Err(response) => return response, }; - let run_dir = { + { let runs = state.runs.lock().expect("runs lock poisoned"); - match runs.get(&id) { - Some(managed_run) => managed_run.run_dir.clone(), - None => return ApiError::not_found("Run not found.").into_response(), + if !runs.contains_key(&id) { + return ApiError::not_found("Run not found.").into_response(); } - }; - - let Some(run_dir) = run_dir else { - return (StatusCode::OK, Json(serde_json::json!(null))).into_response(); - }; + } match state.store.open_run_reader(&id).await { Ok(Some(run_store)) => match run_store.get_retro().await { Ok(Some(retro)) => (StatusCode::OK, Json(retro)).into_response(), - Ok(None) => match Retro::load(&run_dir) { - Ok(retro) => (StatusCode::OK, Json(retro)).into_response(), - Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), - }, + Ok(None) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), Err(err) => { tracing::warn!(run_id = %id, error = %err, "Failed to load retro from store"); - match Retro::load(&run_dir) { - Ok(retro) => (StatusCode::OK, Json(retro)).into_response(), - Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), - } + (StatusCode::OK, Json(serde_json::json!(null))).into_response() } }, - Ok(None) => match Retro::load(&run_dir) { - Ok(retro) => (StatusCode::OK, Json(retro)).into_response(), - Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), - }, + Ok(None) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), Err(err) => { tracing::warn!(run_id = %id, error = %err, "Failed to open run store reader"); - match Retro::load(&run_dir) { - Ok(retro) => (StatusCode::OK, Json(retro)).into_response(), - Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), - } + (StatusCode::OK, Json(serde_json::json!(null))).into_response() } } } @@ -1643,7 +1624,6 @@ mod tests { AuthProvider, AuthSettings, GitAuthorSettings, GitProvider, GitSettings, WebSettings, }; use fabro_types::fixtures; - use fabro_workflow::records::{RunRecord, RunRecordExt}; use tower::ServiceExt; const MINIMAL_DOT: &str = r#"digraph Test { @@ -2507,13 +2487,22 @@ mod tests { let body = body_json(response.into_body()).await; let run_id = body["id"].as_str().unwrap().parse::().unwrap(); - let run_dir = { + let _run_dir = { let runs = state.runs.lock().expect("runs lock poisoned"); runs.get(&run_id) .and_then(|run| run.run_dir.clone()) .expect("run_dir should be recorded") }; - let run_record = RunRecord::load(&run_dir).unwrap(); + let run_record = state + .store + .open_run_reader(&run_id) + .await + .unwrap() + .expect("run store should exist") + .get_run() + .await + .unwrap() + .expect("run record should exist"); let mut expected_settings = settings; expected_settings.goal = Some("Test".to_string()); diff --git a/lib/crates/fabro-store/src/disk_projecting.rs b/lib/crates/fabro-store/src/disk_projecting.rs deleted file mode 100644 index 13a73f927..000000000 --- a/lib/crates/fabro-store/src/disk_projecting.rs +++ /dev/null @@ -1,955 +0,0 @@ -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, NodeOutcomeRecord, NodeSnapshot, NodeVisitRef, Result, - RunSnapshot, RunStore, -}; -use fabro_types::{ - Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunRecord, RunStatusRecord, - SandboxRecord, StartRecord, -}; - -#[derive(Debug, Clone)] -pub struct ProjectionError { - pub path: PathBuf, - pub critical: bool, - pub error: String, -} - -pub struct DiskProjectingRunStore { - inner: Arc, - run_dir: PathBuf, - on_projection_error: Option>, -} - -impl DiskProjectingRunStore { - #[must_use] - pub fn new(inner: Arc, run_dir: PathBuf) -> Self { - Self { - inner, - run_dir, - on_projection_error: None, - } - } - - #[must_use] - pub fn on_projection_error( - mut self, - callback: Arc, - ) -> Self { - self.on_projection_error = Some(callback); - self - } - - fn report_projection_error(&self, 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"); - } - - if let Some(ref callback) = self.on_projection_error { - callback(ProjectionError { - path: path.to_path_buf(), - critical, - error: err.to_string(), - }); - } - } - - fn write_json_critical(&self, path: &Path, value: &T) { - if let Err(err) = write_json(path, value) { - self.report_projection_error(path, &err, true); - } - } - - fn write_json_best_effort(&self, path: &Path, value: &T) { - if let Err(err) = write_json(path, value) { - self.report_projection_error(path, &err, false); - } - } - - fn write_text_best_effort(&self, path: &Path, value: &str) { - if let Err(err) = write_text(path, value) { - self.report_projection_error(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.report_projection_error(&progress_path, &err, true); - } - - let live_path = self.run_dir.join("live.json"); - if let Err(err) = write_live_json(&live_path, payload) { - self.report_projection_error(&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_outcome( - &self, - node: &NodeVisitRef<'_>, - outcome: &NodeOutcomeRecord, - ) -> Result<()> { - self.inner.put_node_outcome(node, outcome).await?; - self.write_json_best_effort( - &disk_node_dir(&self.run_dir, node.node_id, node.visit).join("outcome.json"), - outcome, - ); - Ok(()) - } - - async fn put_node_provider_used( - &self, - node: &NodeVisitRef<'_>, - provider_used: &serde_json::Value, - ) -> Result<()> { - self.inner - .put_node_provider_used(node, provider_used) - .await?; - self.write_json_best_effort( - &disk_node_dir(&self.run_dir, node.node_id, node.visit).join("provider_used.json"), - provider_used, - ); - Ok(()) - } - - async fn put_node_diff(&self, node: &NodeVisitRef<'_>, diff: &str) -> Result<()> { - self.inner.put_node_diff(node, diff).await?; - self.write_text_best_effort( - &disk_node_dir(&self.run_dir, node.node_id, node.visit).join("diff.patch"), - diff, - ); - Ok(()) - } - - async fn put_node_script_invocation( - &self, - node: &NodeVisitRef<'_>, - invocation: &serde_json::Value, - ) -> Result<()> { - self.inner - .put_node_script_invocation(node, invocation) - .await?; - self.write_json_best_effort( - &disk_node_dir(&self.run_dir, node.node_id, node.visit).join("script_invocation.json"), - invocation, - ); - Ok(()) - } - - async fn put_node_script_timing( - &self, - node: &NodeVisitRef<'_>, - timing: &serde_json::Value, - ) -> Result<()> { - self.inner.put_node_script_timing(node, timing).await?; - self.write_json_best_effort( - &disk_node_dir(&self.run_dir, node.node_id, node.visit).join("script_timing.json"), - timing, - ); - Ok(()) - } - - async fn put_node_parallel_results( - &self, - node: &NodeVisitRef<'_>, - results: &serde_json::Value, - ) -> Result<()> { - self.inner.put_node_parallel_results(node, results).await?; - self.write_json_best_effort( - &disk_node_dir(&self.run_dir, node.node_id, node.visit).join("parallel_results.json"), - results, - ); - 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 list_node_ids(&self) -> Result> { - self.inner.list_node_ids().await - } - - async fn put_final_patch(&self, patch: &str) -> Result<()> { - self.inner.put_final_patch(patch).await?; - self.write_text_best_effort(&self.run_dir.join("final.patch"), patch); - Ok(()) - } - - async fn get_final_patch(&self) -> Result> { - self.inner.get_final_patch().await - } - - async fn put_pull_request(&self, record: &PullRequestRecord) -> Result<()> { - self.inner.put_pull_request(record).await?; - self.write_json_best_effort(&self.run_dir.join("pull_request.json"), record); - Ok(()) - } - - async fn get_pull_request(&self) -> Result> { - self.inner.get_pull_request().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, RunId, RunStatus, StageStatus, - StatusReason, fixtures, - }; - - fn dt(rfc3339: &str) -> DateTime { - DateTime::parse_from_rfc3339(rfc3339) - .unwrap() - .with_timezone(&Utc) - } - - fn test_run_id(label: &str) -> RunId { - match label { - "run-1" => fixtures::RUN_1, - _ => panic!("unknown test run id: {label}"), - } - } - - 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: test_run_id(run_id), - 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: test_run_id(run_id), - 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: test_run_id(run_id), - 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, - } - } - - 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!({ - "id": format!("evt-{run_id}-{event}"), - "ts": ts, - "run_id": test_run_id(run_id).to_string(), - "event": event, - }), - &test_run_id(run_id), - ) - .unwrap() - } - - async fn make_store( - run_dir: &Path, - created_at: DateTime, - ) -> (Arc, DiskProjectingRunStore) { - let inner = InMemoryStore::default() - .create_run( - &test_run_id("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")); - } - - #[tokio::test] - async fn projection_error_callback_runs_on_disk_failure() { - let temp = TempDir::new().unwrap(); - let created_at = dt("2026-03-27T12:00:00Z"); - let inner = InMemoryStore::default() - .create_run( - &test_run_id("run-1"), - created_at, - Some(temp.path().to_string_lossy().as_ref()), - ) - .await - .unwrap(); - let seen = Arc::new(std::sync::Mutex::new(Vec::::new())); - let seen_clone = Arc::clone(&seen); - let store = DiskProjectingRunStore::new(inner, temp.path().to_path_buf()) - .on_projection_error(Arc::new(move |error| { - seen_clone.lock().unwrap().push(error); - })); - - let mut permissions = fs::metadata(temp.path()).unwrap().permissions(); - permissions.set_readonly(true); - fs::set_permissions(temp.path(), permissions).unwrap(); - - store - .put_status(&sample_status( - RunStatus::Running, - Some(StatusReason::SandboxInitializing), - )) - .await - .unwrap(); - - let seen = seen.lock().unwrap(); - assert_eq!(seen.len(), 1); - assert!(seen[0].critical); - assert_eq!(seen[0].path, temp.path().join("status.json")); - assert!(!seen[0].error.is_empty()); - } - - #[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 2210b79a9..8d48a6a39 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -6,7 +6,6 @@ use bytes::Bytes; use chrono::{DateTime, Utc}; use futures::Stream; -mod disk_projecting; mod error; mod keys; mod memory; @@ -14,7 +13,6 @@ mod runtime; mod slate; mod types; -pub use disk_projecting::{DiskProjectingRunStore, ProjectionError}; pub use error::{Result, StoreError}; pub use memory::InMemoryStore; pub use runtime::RuntimeState; diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index e4921f181..bcb7b396b 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -331,9 +331,6 @@ impl Handler for AgentHandler { match result { Ok(CodergenResult::Full(outcome)) => { sync_provider_used_to_store(&stage_dir, &node_ref, services).await?; - let status_json = serde_json::to_string_pretty(&outcome) - .unwrap_or_else(|_| "{}".to_string()); - fs::write(stage_dir.join("status.json"), &status_json).await?; return Ok(outcome); } Ok(CodergenResult::Text { @@ -418,10 +415,6 @@ impl Handler for AgentHandler { outcome.usage = stage_usage; outcome.files_touched = backend_files_touched; - let status_json = - serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string()); - fs::write(stage_dir.join("status.json"), &status_json).await?; - Ok(outcome) } } diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index 7bd136870..0f1c71b32 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -116,9 +116,6 @@ impl Handler for PromptHandler { match result { Ok(CodergenResult::Full(outcome)) => { sync_provider_used_to_store(&stage_dir, &node_ref, services).await?; - let status_json = serde_json::to_string_pretty(&outcome) - .unwrap_or_else(|_| "{}".to_string()); - fs::write(stage_dir.join("status.json"), &status_json).await?; return Ok(outcome); } Ok(CodergenResult::Text { @@ -191,10 +188,6 @@ impl Handler for PromptHandler { outcome.usage = stage_usage; outcome.files_touched = backend_files_touched; - let status_json = - serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string()); - fs::write(stage_dir.join("status.json"), &status_json).await?; - Ok(outcome) } } diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index 50a909a26..eef58a9dd 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -18,7 +18,6 @@ use crate::git::scan_node_files_from_store; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; use crate::outcome::{Outcome, StageStatus, StageUsage}; -use crate::run_dir::node_dir; use crate::run_options::RunOptions; use crate::sandbox_git::{git_checkpoint, git_diff, git_push_host}; @@ -289,8 +288,6 @@ impl RunLifecycle for GitLifecycle { .and_then(|g| g.base_sha.clone()) }) .unwrap_or_else(|| sha.clone()); - let diff_dest = node_dir(&self.run_dir, node_id, visit).join("diff.patch"); - match git_diff(&*self.sandbox, &prev).await { Ok(patch) if !patch.is_empty() => { let node_ref = fabro_store::NodeVisitRef { @@ -309,7 +306,6 @@ impl RunLifecycle for GitLifecycle { "failed to persist node diff for '{node_id}': {err}" ))); } - let _ = std::fs::write(&diff_dest, &patch); git_result.diff = Some(patch); } Ok(_) => {} @@ -352,7 +348,6 @@ impl RunLifecycle for GitLifecycle { .as_ref() .and_then(|g| g.base_sha.clone()) { - let diff_dest = self.run_dir.join("final.patch"); match git_diff(&*self.sandbox, &base_sha).await { Ok(patch) if !patch.is_empty() => { if let Err(err) = self.run_store.put_final_patch(&patch).await { @@ -365,7 +360,6 @@ impl RunLifecycle for GitLifecycle { }); return; } - let _ = std::fs::write(&diff_dest, patch); } Ok(_) => {} Err(err) => { diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index 81badb0ee..2399f08f1 100644 --- a/lib/crates/fabro-workflow/src/operations/create.rs +++ b/lib/crates/fabro-workflow/src/operations/create.rs @@ -1,15 +1,13 @@ -use std::collections::BTreeMap; -use std::collections::HashMap; -use std::path::{Path, PathBuf}; -use std::sync::Arc; - use chrono::{Local, Utc}; use fabro_config::{FabroSettings, FabroSettingsExt}; use fabro_graphviz::graph::{AttrValue, Graph}; use fabro_model::{Catalog, Provider}; use fabro_sandbox::SandboxProvider; -use fabro_store::{DiskProjectingRunStore, RunStore, Store}; +use fabro_store::Store; use fabro_types::RunId; +use std::collections::BTreeMap; +use std::collections::HashMap; +use std::path::{Path, PathBuf}; use crate::error::FabroError; use crate::pipeline::types::PersistOptions; @@ -141,7 +139,7 @@ async fn persist_created_run( ) -> Result<(), FabroError> { let record = persisted.run_record(); let run_dir_string = persisted.run_dir().to_string_lossy().to_string(); - let inner_run_store = match store + let run_store = match store .create_run(&record.run_id, record.created_at, Some(&run_dir_string)) .await { @@ -152,10 +150,6 @@ async fn persist_created_run( .map_err(|open_err| FabroError::engine(open_err.to_string()))? .ok_or_else(|| FabroError::engine(err.to_string()))?, }; - let run_store: Arc = Arc::new(DiskProjectingRunStore::new( - inner_run_store, - persisted.run_dir().to_path_buf(), - )); run_store.put_run(record).await.map_err(store_error)?; if !workflow_source.is_empty() { diff --git a/lib/crates/fabro-workflow/src/operations/resume.rs b/lib/crates/fabro-workflow/src/operations/resume.rs index 268f04a69..e6b7246b1 100644 --- a/lib/crates/fabro-workflow/src/operations/resume.rs +++ b/lib/crates/fabro-workflow/src/operations/resume.rs @@ -68,12 +68,7 @@ fn cleanup_resume_artifacts(run_dir: &Path) { let _ = std::fs::remove_file(path); } - for name in [ - "conclusion.json", - "pull_request.json", - "detached_failure.json", - "progress.jsonl", - ] { + for name in ["detached_failure.json"] { let _ = std::fs::remove_file(run_dir.join(name)); } } diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index 9c351c615..1d4d75b17 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -11,16 +11,15 @@ 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}; -use fabro_store::{DiskProjectingRunStore, ProjectionError, RunStore}; +use fabro_store::RunStore; use fabro_types::RunId; use serde::Serialize; use crate::context::Context; use crate::error::FabroError; use crate::event::{ - EventEmitter, RunNoticeLevel, StoreProgressLogger, WorkflowRunEvent, append_progress_event, - append_progress_event_with_line, canonicalize_event, event_payload_from_redacted_json, - redacted_event_json, + EventEmitter, RunNoticeLevel, StoreProgressLogger, WorkflowRunEvent, canonicalize_event, + event_payload_from_redacted_json, redacted_event_json, }; use crate::git::MetadataStore; use crate::handler::HandlerRegistry; @@ -116,38 +115,10 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result, - mut services: StartServices, + services: StartServices, ) -> Result { let cancel_token = services.cancel_token.clone(); let run_id = services.run_id; - let inner_store = Arc::clone(&services.run_store); - let projection_run_dir = run_dir.to_path_buf(); - services.run_store = Arc::new( - DiskProjectingRunStore::new(inner_store, run_dir.to_path_buf()).on_projection_error( - Arc::new(move |projection_error: ProjectionError| { - // Write directly to progress.jsonl/live.json so projection failures do not - // recurse back through the decorated store.append_event() path. - let envelope = canonicalize_event( - &run_id, - &WorkflowRunEvent::RunNotice { - level: RunNoticeLevel::Warn, - code: "disk_projection_failed".to_string(), - message: format!( - "{}disk projection failed for {}: {}", - if projection_error.critical { - "critical " - } else { - "" - }, - projection_error.path.display(), - projection_error.error - ), - }, - ); - let _ = append_progress_event(&projection_run_dir, &envelope); - }), - ), - ); let run_store = Arc::clone(&services.run_store); if let Err(err) = run_store .put_status(&run_status::RunStatusRecord::new( @@ -208,7 +179,7 @@ pub(super) async fn execute_persisted_run( bootstrap_guard.defuse(); let mut completion_guard = - DetachedRunCompletionGuard::arm(run_dir, run_id, Arc::clone(&run_store), cancel_token); + DetachedRunCompletionGuard::arm(run_id, Arc::clone(&run_store), cancel_token); let run_start = Instant::now(); let started = Box::pin(session.run(persisted, checkpoint)).await; @@ -228,7 +199,7 @@ pub(super) async fn execute_persisted_run( async fn persist_terminal_engine_failure( run_store: &dyn RunStore, - run_dir: &Path, + _run_dir: &Path, error: &FabroError, duration: Duration, ) { @@ -237,7 +208,6 @@ async fn persist_terminal_engine_failure( classify_engine_result(&engine_result); let conclusion = build_conclusion_from_store( run_store, - run_dir, final_status, failure_reason, u64::try_from(duration.as_millis()).unwrap(), @@ -655,7 +625,6 @@ const POSTRUN_ABORTED_MESSAGE: &str = "Run aborted before post-run finalization const POSTRUN_CANCELLED_MESSAGE: &str = "Run cancelled before post-run finalization completed."; struct DetachedRunCompletionGuard { - run_dir: PathBuf, run_store: Arc, run_id: RunId, cancel_token: Option>, @@ -664,13 +633,11 @@ struct DetachedRunCompletionGuard { impl DetachedRunCompletionGuard { fn arm( - run_dir: &Path, run_id: RunId, run_store: Arc, cancel_token: Option>, ) -> Self { Self { - run_dir: run_dir.to_path_buf(), run_store, run_id, cancel_token, @@ -727,11 +694,6 @@ impl Drop for DetachedRunCompletionGuard { }; if line.is_empty() { None - } else if let Err(err) = - append_progress_event_with_line(&self.run_dir, &envelope, &line) - { - tracing::warn!(error = %err, "Failed to append post-run abort event"); - None } else { Some((self.run_id, line)) } @@ -836,8 +798,6 @@ async fn persist_detached_failure( }; let envelope = canonicalize_event(&run_id, &event); let line = redacted_event_json(&envelope).map_err(|err| FabroError::Io(err.to_string()))?; - append_progress_event_with_line(run_dir, &envelope, &line) - .map_err(|err| FabroError::Io(err.to_string()))?; match event_payload_from_redacted_json(&line, &run_id) { Ok(payload) => { if let Err(err) = run_store.append_event(&payload).await { diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs index 1bff3fd32..f1d88b02d 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs @@ -26,7 +26,7 @@ use crate::handler::{Handler as HandlerTrait, HandlerRegistry}; use crate::outcome::{Outcome, OutcomeExt, StageStatus}; use crate::pipeline::initialize; use crate::pipeline::types::{InitOptions, LlmSpec, Persisted, SandboxEnvSpec}; -use crate::records::{Checkpoint, CheckpointExt, RunRecord, StartRecordExt}; +use crate::records::RunRecord; use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions}; use crate::test_support::run_graph; @@ -165,6 +165,62 @@ async fn test_run_store(_run_dir: &Path, run_id: &RunId) -> Arc Executed { + execute_test_run_with_options(test_run_options(run_dir, run_id), graph, None).await +} + +async fn execute_test_run_with_options( + run_options: RunOptions, + graph: Graph, + registry_override: Option>, +) -> Executed { + let run_id_value = run_options.run_id; + let git_options = run_options.git.clone(); + let initialized = initialize( + persisted_workflow(graph, String::new(), &run_options.run_dir, run_id_value), + InitOptions { + run_id: run_id_value, + run_store: test_run_store(&run_options.run_dir, &run_id_value).await, + dry_run: false, + emitter: test_emitter_arc("test-run"), + sandbox: SandboxSpec::Local { + working_directory: std::env::current_dir().unwrap(), + }, + llm: LlmSpec { + model: "test-model".to_string(), + provider: fabro_llm::Provider::Anthropic, + fallback_chain: Vec::new(), + mcp_servers: Vec::new(), + dry_run: true, + }, + interviewer: Arc::new(AutoApproveInterviewer), + lifecycle: LifecycleOptions { + setup_commands: vec![], + setup_command_timeout_ms: 1_000, + devcontainer_phases: vec![], + }, + run_options, + hooks: HookConfig { hooks: vec![] }, + sandbox_env: SandboxEnvSpec { + devcontainer_env: HashMap::new(), + toml_env: HashMap::new(), + github_permissions: None, + origin_url: None, + }, + devcontainer: None, + git: git_options, + worktree_mode: None, + registry_override, + checkpoint: None, + seed_context: None, + }, + ) + .await + .unwrap(); + + execute(initialized).await +} + #[tokio::test] async fn execute_runs_start_to_exit_and_returns_final_context() { let temp = tempfile::tempdir().unwrap(); @@ -447,16 +503,8 @@ async fn execute_runs_simple_workflow() { #[tokio::test] async fn execute_saves_checkpoint() { let dir = tempfile::tempdir().unwrap(); - run_graph( - make_registry(), - test_emitter_arc("test-run"), - local_env(), - &simple_graph(), - &test_run_options(dir.path(), "test-run"), - ) - .await - .unwrap(); - assert!(dir.path().join("checkpoint.json").exists()); + let executed = execute_test_run(dir.path(), simple_graph(), "test-run").await; + assert!(executed.run_store.get_checkpoint().await.unwrap().is_some()); } #[tokio::test] @@ -499,17 +547,8 @@ async fn execute_error_when_no_start_node() { #[tokio::test] async fn execute_mirrors_graph_goal_to_context() { let dir = tempfile::tempdir().unwrap(); - run_graph( - make_registry(), - test_emitter_arc("test-run"), - local_env(), - &simple_graph(), - &test_run_options(dir.path(), "test-run"), - ) - .await - .unwrap(); - - let cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); + let executed = execute_test_run(dir.path(), simple_graph(), "test-run").await; + let cp = executed.run_store.get_checkpoint().await.unwrap().unwrap(); assert_eq!( cp.context_values.get(context::keys::GRAPH_GOAL), Some(&serde_json::json!("Run tests")) @@ -548,17 +587,8 @@ async fn execute_conditional_routing_uses_unconditional_success_path() { g.edges.push(Edge::new("path_a", "exit")); g.edges.push(Edge::new("path_b", "exit")); - run_graph( - make_registry(), - test_emitter_arc("test-run"), - local_env(), - &g, - &test_run_options(dir.path(), "test-run"), - ) - .await - .unwrap(); - - let cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); + let executed = execute_test_run(dir.path(), g, "test-run").await; + let cp = executed.run_store.get_checkpoint().await.unwrap().unwrap(); assert!(cp.completed_nodes.contains(&"path_b".to_string())); assert!(!cp.completed_nodes.contains(&"path_a".to_string())); } @@ -573,17 +603,8 @@ async fn execute_writes_start_json_and_node_status() { meta_branch: None, }); - run_graph( - make_registry(), - test_emitter_arc("test-run"), - local_env(), - &simple_graph(), - &run_options, - ) - .await - .unwrap(); - - let start = crate::records::StartRecord::load(dir.path()).unwrap(); + let executed = execute_test_run_with_options(run_options, simple_graph(), None).await; + let start = executed.run_store.get_start().await.unwrap().unwrap(); assert_eq!(start.run_id, test_run_id("test-run")); assert_eq!( start.run_branch.as_deref(), @@ -591,12 +612,19 @@ async fn execute_writes_start_json_and_node_status() { ); assert_eq!(start.base_sha.as_deref(), Some("abc123")); - let status_path = dir.path().join("nodes").join("start").join("status.json"); - assert!(status_path.exists()); + let node = executed + .run_store + .get_node(&fabro_store::NodeVisitRef { + node_id: "start", + visit: 1, + }) + .await + .unwrap(); + assert_eq!(node.status.unwrap().status, StageStatus::Success); } #[tokio::test] -async fn timeout_causes_fail_status_json() { +async fn timeout_causes_fail_status_record() { let dir = tempfile::tempdir().unwrap(); let mut g = Graph::new("timeout_test"); @@ -635,20 +663,23 @@ async fn timeout_causes_fail_status_json() { let mut registry = make_registry(); registry.register("slow", Box::new(SlowHandler { sleep_ms: 500 })); - run_graph( - registry, - test_emitter_arc("test-run"), - local_env(), - &g, - &test_run_options(dir.path(), "test-run"), + let executed = execute_test_run_with_options( + test_run_options(dir.path(), "test-run"), + g, + Some(Arc::new(registry)), ) - .await - .unwrap(); - - let status_path = dir.path().join("nodes").join("work").join("status.json"); - let status: serde_json::Value = - serde_json::from_str(&std::fs::read_to_string(&status_path).unwrap()).unwrap(); - assert_eq!(status["status"], "fail"); + .await; + let status = executed + .run_store + .get_node(&fabro_store::NodeVisitRef { + node_id: "work", + visit: 1, + }) + .await + .unwrap() + .status + .unwrap(); + assert_eq!(status.status, StageStatus::Fail); } #[tokio::test] diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index 9e08e5d14..3f0b29be1 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -5,12 +5,11 @@ use crate::error::FabroError; use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; use crate::git::{MetadataStore, scan_node_files_from_store}; use crate::outcome::{Outcome, OutcomeExt, StageStatus}; -use crate::records::{Checkpoint, CheckpointExt, Conclusion, StageSummary}; +use crate::records::{Checkpoint, Conclusion, StageSummary}; use crate::run_options::RunOptions; use crate::run_status::{RunStatus, StatusReason}; use crate::sandbox_git::git_push_host; use fabro_hooks::{HookContext, HookEvent, HookRunner}; -use fabro_retro::retro::extract_stage_durations; use fabro_store::RunStore; use super::types::{Concluded, FinalizeOptions, Retroed}; @@ -63,105 +62,19 @@ pub fn classify_engine_result( } } -pub fn build_conclusion( - run_dir: &Path, - status: StageStatus, - failure_reason: Option, - run_duration_ms: u64, - final_git_commit_sha: Option, -) -> Conclusion { - let checkpoint = Checkpoint::load(&run_dir.join("checkpoint.json")).ok(); - let stage_durations = extract_stage_durations(run_dir); - - let mut total_input_tokens: i64 = 0; - let mut total_output_tokens: i64 = 0; - let mut total_cache_read_tokens: i64 = 0; - let mut total_cache_write_tokens: i64 = 0; - let mut total_reasoning_tokens: i64 = 0; - let mut has_pricing = false; - - let (stages, total_cost, total_retries) = if let Some(ref cp) = checkpoint { - let mut stages = Vec::new(); - let mut cost_sum: Option = None; - let mut retries_sum: u32 = 0; - - for node_id in &cp.completed_nodes { - let outcome = cp.node_outcomes.get(node_id); - let retries = cp - .node_retries - .get(node_id) - .copied() - .unwrap_or(1) - .saturating_sub(1); - retries_sum += retries; - - let cost = outcome.and_then(|o| o.usage.as_ref()).and_then(|u| u.cost); - if let Some(c) = cost { - *cost_sum.get_or_insert(0.0) += c; - has_pricing = true; - } - - if let Some(usage) = outcome.and_then(|o| o.usage.as_ref()) { - total_input_tokens += usage.input_tokens; - total_output_tokens += usage.output_tokens; - total_cache_read_tokens += usage.cache_read_tokens.unwrap_or(0); - total_cache_write_tokens += usage.cache_write_tokens.unwrap_or(0); - total_reasoning_tokens += usage.reasoning_tokens.unwrap_or(0); - } - - stages.push(StageSummary { - stage_id: node_id.clone(), - stage_label: node_id.clone(), - duration_ms: stage_durations.get(node_id).copied().unwrap_or(0), - cost, - retries, - }); - } - (stages, cost_sum, retries_sum) - } else { - (vec![], None, 0) - }; - - Conclusion { - timestamp: chrono::Utc::now(), - status, - duration_ms: run_duration_ms, - failure_reason, - final_git_commit_sha, - stages, - total_cost, - total_retries, - total_input_tokens, - total_output_tokens, - total_cache_read_tokens, - total_cache_write_tokens, - total_reasoning_tokens, - has_pricing, - } -} - pub(crate) async fn build_conclusion_from_store( run_store: &dyn RunStore, - run_dir: &Path, status: StageStatus, failure_reason: Option, run_duration_ms: u64, final_git_commit_sha: Option, ) -> Conclusion { - let checkpoint = match run_store.get_checkpoint().await { - Ok(checkpoint) => checkpoint, - Err(err) => { - tracing::warn!(error = %err, "Failed to load checkpoint from store while building conclusion"); - Checkpoint::load(&run_dir.join("checkpoint.json")).ok() - } - }; - let stage_durations = match run_store.list_events().await { - Ok(events) => crate::extract_stage_durations_from_events(&events), - Err(err) => { - tracing::warn!(error = %err, "Failed to load events from store while building conclusion"); - extract_stage_durations(run_dir) - } - }; + let checkpoint = run_store.get_checkpoint().await.ok().flatten(); + let stage_durations = run_store + .list_events() + .await + .map(|events| crate::extract_stage_durations_from_events(&events)) + .unwrap_or_default(); build_conclusion_from_parts( checkpoint.as_ref(), @@ -360,7 +273,6 @@ pub async fn finalize( classify_engine_result(&outcome); let conclusion = build_conclusion_from_store( options.run_store.as_ref(), - &options.run_dir, final_status, failure_reason, duration_ms, @@ -479,9 +391,7 @@ mod tests { ) .await .unwrap(); - let run_store: Arc = Arc::new( - fabro_store::DiskProjectingRunStore::new(inner_store, run_dir.clone()), - ); + let run_store: Arc = inner_store; let retroed = Retroed { graph: Graph::new("test"), outcome: Ok(Outcome::success()), diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs index 18853c3a4..909bca736 100644 --- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs @@ -771,10 +771,7 @@ mod tests { ) .await .unwrap(); - Arc::new(fabro_store::DiskProjectingRunStore::new( - inner, - run_dir.clone(), - )) + inner }, dry_run: false, emitter, @@ -849,10 +846,7 @@ mod tests { ) .await .unwrap(); - Arc::new(fabro_store::DiskProjectingRunStore::new( - inner, - run_dir.clone(), - )) + inner }, dry_run: false, emitter, @@ -891,7 +885,7 @@ mod tests { .await .unwrap(); - assert!(run_dir.join("sandbox.json").exists()); + assert!(initialized.run_store.get_sandbox().await.unwrap().is_some()); assert_eq!(initialized.run_options.run_dir, run_dir); } } diff --git a/lib/crates/fabro-workflow/src/pipeline/mod.rs b/lib/crates/fabro-workflow/src/pipeline/mod.rs index 707d32b08..302a2b546 100644 --- a/lib/crates/fabro-workflow/src/pipeline/mod.rs +++ b/lib/crates/fabro-workflow/src/pipeline/mod.rs @@ -13,8 +13,7 @@ pub use execute::execute; pub use fabro_types::PullRequestRecord; pub(crate) use finalize::build_conclusion_from_store; pub use finalize::{ - build_conclusion, classify_engine_result, finalize, persist_terminal_outcome, - write_finalize_commit, + classify_engine_result, finalize, persist_terminal_outcome, write_finalize_commit, }; pub use initialize::initialize; pub use parse::parse; diff --git a/lib/crates/fabro-workflow/src/pipeline/persist.rs b/lib/crates/fabro-workflow/src/pipeline/persist.rs index 51c2fa1a7..eb9772850 100644 --- a/lib/crates/fabro-workflow/src/pipeline/persist.rs +++ b/lib/crates/fabro-workflow/src/pipeline/persist.rs @@ -3,16 +3,10 @@ use std::path::Path; use fabro_store::RunStore; use crate::error::FabroError; -use crate::records::{RunRecord, RunRecordExt}; use super::types::{PersistOptions, Persisted, Validated}; -const GRAPH_FILE_NAME: &str = "workflow.fabro"; -const LEGACY_GRAPH_FILE_NAME: &str = "graph.fabro"; - -/// PERSIST phase: create run directory, write workflow.fabro and run.json to disk. -/// -/// Overwrites `run_record.graph` with the validated graph before saving. +/// PERSIST phase: create the run directory and return durable metadata for store persistence. pub(crate) fn persist( validated: Validated, mut options: PersistOptions, @@ -21,10 +15,6 @@ pub(crate) fn persist( options.run_record.graph = graph.clone(); std::fs::create_dir_all(&options.run_dir)?; - if !source.is_empty() { - std::fs::write(options.run_dir.join(GRAPH_FILE_NAME), &source)?; - } - options.run_record.save(&options.run_dir)?; Ok(Persisted::new( graph, @@ -35,34 +25,6 @@ pub(crate) fn persist( )) } -/// Load a previously persisted run from disk. -/// -/// `run.json` is authoritative for graph + config; `workflow.fabro` provides the -/// original DOT source string when present. -pub(crate) fn load(run_dir: &Path) -> Result { - let run_record = RunRecord::load(run_dir)?; - let graph = run_record.graph.clone(); - let source = match std::fs::read_to_string(run_dir.join(GRAPH_FILE_NAME)) { - Ok(source) => source, - Err(err) if err.kind() == std::io::ErrorKind::NotFound => { - match std::fs::read_to_string(run_dir.join(LEGACY_GRAPH_FILE_NAME)) { - Ok(source) => source, - Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(), - Err(err) => return Err(err.into()), - } - } - Err(err) => return Err(err.into()), - }; - - Ok(Persisted::new( - graph, - source, - Vec::new(), - run_dir.to_path_buf(), - run_record, - )) -} - pub(crate) async fn load_from_store( run_store: &dyn RunStore, run_dir: &Path, @@ -96,6 +58,7 @@ mod tests { use chrono::Utc; use fabro_config::FabroSettings; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; + use fabro_store::{InMemoryStore, Store}; use fabro_types::fixtures; use super::*; @@ -166,59 +129,33 @@ mod tests { } } + async fn seeded_store( + run_dir: &Path, + record: &RunRecord, + source: Option<&str>, + ) -> std::sync::Arc { + let store = InMemoryStore::default(); + let run_store = store + .create_run( + &record.run_id, + record.created_at, + Some(run_dir.to_string_lossy().as_ref()), + ) + .await + .unwrap(); + run_store.put_run(record).await.unwrap(); + if let Some(source) = source { + run_store.put_graph(source).await.unwrap(); + } + run_store + } + #[test] - fn persist_creates_run_dir_and_writes_graph_and_record() { + fn persist_creates_run_dir_without_writing_legacy_files() { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); let (graph, source) = graph_and_source(); let persisted = persist( - Validated::new(graph.clone(), source.clone(), vec![]), - PersistOptions { - run_dir: run_dir.clone(), - run_record: sample_record(different_graph()), - }, - ) - .unwrap(); - - assert!(run_dir.is_dir()); - assert_eq!( - std::fs::read_to_string(run_dir.join(GRAPH_FILE_NAME)).unwrap(), - source - ); - assert!(run_dir.join(RunRecord::file_name()).exists()); - assert_eq!(persisted.run_dir(), run_dir.as_path()); - assert_eq!( - serde_json::to_value(persisted.run_record().graph.clone()).unwrap(), - serde_json::to_value(graph).unwrap() - ); - } - - #[test] - fn persist_skips_graph_file_when_source_is_empty() { - let temp = tempfile::tempdir().unwrap(); - let run_dir = temp.path().join("run"); - let (graph, _source) = graph_and_source(); - - persist( - Validated::new(graph, String::new(), vec![]), - PersistOptions { - run_dir: run_dir.clone(), - run_record: sample_record(different_graph()), - }, - ) - .unwrap(); - - assert!(!run_dir.join(GRAPH_FILE_NAME).exists()); - assert!(run_dir.join(RunRecord::file_name()).exists()); - } - - #[test] - fn persist_overwrites_run_record_graph_with_validated_graph() { - let temp = tempfile::tempdir().unwrap(); - let run_dir = temp.path().join("run"); - let (graph, source) = graph_and_source(); - - persist( Validated::new(graph.clone(), source, vec![]), PersistOptions { run_dir: run_dir.clone(), @@ -227,17 +164,41 @@ mod tests { ) .unwrap(); - let saved = RunRecord::load(&run_dir).unwrap(); - assert_eq!(saved.graph.name, graph.name); - assert!(saved.graph.nodes.contains_key("exit")); + assert!(run_dir.is_dir()); + assert!(!run_dir.join("workflow.fabro").exists()); + assert!(!run_dir.join("run.json").exists()); + assert_eq!(persisted.run_dir(), run_dir.as_path()); assert_eq!( - serde_json::to_value(saved.graph).unwrap(), + serde_json::to_value(persisted.run_record().graph.clone()).unwrap(), serde_json::to_value(graph).unwrap() ); } #[test] - fn persist_roundtrips_full_run_record_fields_through_load() { + fn persist_overwrites_run_record_graph_with_validated_graph() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run"); + let (graph, source) = graph_and_source(); + + let persisted = persist( + Validated::new(graph.clone(), source, vec![]), + PersistOptions { + run_dir: run_dir.clone(), + run_record: sample_record(different_graph()), + }, + ) + .unwrap(); + + assert_eq!(persisted.run_record().graph.name, graph.name); + assert!(persisted.run_record().graph.nodes.contains_key("exit")); + assert_eq!( + serde_json::to_value(persisted.run_record().graph.clone()).unwrap(), + serde_json::to_value(graph).unwrap() + ); + } + + #[tokio::test] + async fn load_from_store_roundtrips_full_run_record_fields() { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); let (graph, source) = graph_and_source(); @@ -245,7 +206,7 @@ mod tests { expected.graph = graph.clone(); persist( - Validated::new(graph, source, vec![]), + Validated::new(graph, source.clone(), vec![]), PersistOptions { run_dir: run_dir.clone(), run_record: expected.clone(), @@ -253,12 +214,14 @@ mod tests { ) .unwrap(); - let loaded = Persisted::load(&run_dir).unwrap(); + let run_store = seeded_store(&run_dir, &expected, Some(&source)).await; + let loaded = load_from_store(run_store.as_ref(), &run_dir).await.unwrap(); assert_eq!( serde_json::to_value(loaded.run_record()).unwrap(), serde_json::to_value(expected).unwrap() ); + assert_eq!(loaded.source(), source); assert!(loaded.diagnostics().is_empty()); } @@ -281,92 +244,38 @@ mod tests { assert!(matches!(err, FabroError::Io(_))); } - #[test] - fn load_roundtrips_persisted_workflow() { - let temp = tempfile::tempdir().unwrap(); - let run_dir = temp.path().join("run"); - let (graph, source) = graph_and_source(); - let mut expected = sample_record(different_graph()); - expected.graph = graph.clone(); - - let persisted = persist( - Validated::new(graph, source.clone(), vec![]), - PersistOptions { - run_dir: run_dir.clone(), - run_record: expected.clone(), - }, - ) - .unwrap(); - let loaded = Persisted::load(&run_dir).unwrap(); - - assert_eq!(loaded.source(), source); - assert_eq!(loaded.run_dir(), run_dir.as_path()); - assert_eq!( - serde_json::to_value(loaded.run_record()).unwrap(), - serde_json::to_value(expected).unwrap() - ); - assert_eq!( - serde_json::to_value(loaded.graph()).unwrap(), - serde_json::to_value(persisted.graph()).unwrap() - ); - } - - #[test] - fn load_uses_empty_source_when_graph_file_is_missing() { + #[tokio::test] + async fn load_from_store_uses_empty_source_when_graph_missing() { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); std::fs::create_dir_all(&run_dir).unwrap(); let (graph, _source) = graph_and_source(); let mut record = sample_record(different_graph()); record.graph = graph; - record.save(&run_dir).unwrap(); - let loaded = Persisted::load(&run_dir).unwrap(); + let run_store = seeded_store(&run_dir, &record, None).await; + let loaded = load_from_store(run_store.as_ref(), &run_dir).await.unwrap(); assert!(loaded.source().is_empty()); } - #[test] - fn load_reads_graph_from_run_json_and_source_from_graph_file() { + #[tokio::test] + async fn load_from_store_reads_graph_from_run_record_and_source_from_store() { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); std::fs::create_dir_all(&run_dir).unwrap(); - let (graph, _) = graph_and_source(); + let (graph, source) = graph_and_source(); let mut record = sample_record(different_graph()); record.graph = graph.clone(); - record.save(&run_dir).unwrap(); - std::fs::write( - run_dir.join(GRAPH_FILE_NAME), - "digraph mismatch { a -> b; }", - ) - .unwrap(); - let loaded = Persisted::load(&run_dir).unwrap(); + let run_store = seeded_store(&run_dir, &record, Some(&source)).await; + let loaded = load_from_store(run_store.as_ref(), &run_dir).await.unwrap(); assert_eq!( serde_json::to_value(loaded.graph()).unwrap(), serde_json::to_value(graph).unwrap() ); - assert_eq!(loaded.source(), "digraph mismatch { a -> b; }"); - } - - #[test] - fn load_reads_graph_source_from_graph_file() { - let temp = tempfile::tempdir().unwrap(); - let run_dir = temp.path().join("run"); - let (graph, source) = graph_and_source(); - - persist( - Validated::new(graph, source.clone(), vec![]), - PersistOptions { - run_dir: run_dir.clone(), - run_record: sample_record(different_graph()), - }, - ) - .unwrap(); - - let loaded = Persisted::load(&run_dir).unwrap(); assert_eq!(loaded.source(), source); } } diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 46b0e4df2..65c2f7636 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -8,16 +8,13 @@ use tracing::{debug, info}; use fabro_github::{self as github_app, GitHubAppCredentials, ssh_url_to_https}; use fabro_graphviz::parser; use fabro_llm::generate::{GenerateParams, generate}; -use fabro_retro::RetroExt; use fabro_util::text::strip_goal_decoration; +use super::types::{Concluded, Finalized, PullRequestOptions}; use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; use crate::outcome::{StageStatus, format_cost as outcome_format_cost}; -use crate::records::{Conclusion, ConclusionExt, RunRecord, RunRecordExt}; +use crate::records::{Conclusion, RunRecord, RunRecordExt}; use fabro_retro::retro::Retro; -use tokio::fs::read_to_string; - -use super::types::{Concluded, Finalized, PullRequestOptions}; /// Derive a PR title from the workflow goal. /// @@ -199,8 +196,7 @@ fn parse_dot_summary(dot: &str) -> (String, usize, usize) { } } -/// Read the workflow graph source from `run_dir/workflow.fabro`. -/// Falls back to `graph.fabro` / `graph.dot` for older runs. +#[cfg(test)] fn read_dot_source(run_dir: &Path) -> Option { let workflow_fabro_path = run_dir.join("workflow.fabro"); if let Ok(content) = std::fs::read_to_string(&workflow_fabro_path) { @@ -307,9 +303,10 @@ async fn load_pull_request_diff(run_store: Option<&dyn RunStore>, run_dir: &Path .ok() .flatten() .unwrap_or_default(), - None => read_to_string(run_dir.join("final.patch")) - .await - .unwrap_or_default(), + None => { + let _ = run_dir; + String::new() + } } } @@ -336,7 +333,7 @@ pub async fn build_pr_body( }) .ok() .flatten(), - None => Conclusion::load(&run_dir.join("conclusion.json")).ok(), + None => None, } } else { None @@ -351,7 +348,7 @@ pub async fn build_pr_body( }) .ok() .flatten(), - None => Retro::load(run_dir).ok(), + None => None, }; let run_record = match run_store { Some(run_store) => run_store @@ -362,7 +359,7 @@ pub async fn build_pr_body( }) .ok() .flatten(), - None => RunRecord::load(run_dir).ok(), + None => None, }; let dot_source = match run_store { Some(run_store) => run_store @@ -373,7 +370,7 @@ pub async fn build_pr_body( }) .ok() .flatten(), - None => read_dot_source(run_dir), + None => None, }; // Build LLM prompt diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index 47226f12c..fbe3f0e83 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -1,7 +1,7 @@ use std::sync::Arc; use fabro_agent::SessionEvent; -use fabro_retro::retro::{Retro, derive_retro, extract_stage_durations}; +use fabro_retro::retro::{Retro, derive_retro}; use fabro_retro::retro_agent::{dry_run_narrative, run_retro_agent}; use super::types::{Executed, RetroOptions, Retroed}; @@ -36,8 +36,8 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { let stage_durations = match options.run_store.list_events().await { Ok(events) => crate::extract_stage_durations_from_events(&events), Err(err) => { - tracing::warn!(error = %err, "Could not load events from store, falling back to disk"); - extract_stage_durations(&options.run_dir) + tracing::warn!(error = %err, "Could not load events from store, skipping stage durations"); + Default::default() } }; let mut retro = derive_retro( @@ -185,7 +185,7 @@ mod tests { fixtures::RUN_1 } - fn write_checkpoint(run_dir: &std::path::Path) -> Checkpoint { + fn build_checkpoint() -> Checkpoint { let context = Context::new(); context.set("response.work", serde_json::json!("done")); let mut outcomes = HashMap::new(); @@ -201,7 +201,6 @@ mod tests { HashMap::new(), HashMap::new(), ); - checkpoint.save(&run_dir.join("checkpoint.json")).unwrap(); checkpoint } @@ -217,9 +216,7 @@ mod tests { ) .await .unwrap(); - let run_store: Arc = Arc::new( - fabro_store::DiskProjectingRunStore::new(inner, run_dir.to_path_buf()), - ); + let run_store: Arc = inner; run_store.put_checkpoint(checkpoint).await.unwrap(); run_store } @@ -245,7 +242,8 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); std::fs::create_dir_all(&run_dir).unwrap(); - let checkpoint = write_checkpoint(&run_dir); + let checkpoint = build_checkpoint(); + let run_store = test_run_store(&run_dir, &checkpoint).await; let emitter = Arc::new(EventEmitter::default()); let sandbox: Arc = Arc::new(fabro_agent::LocalSandbox::new( @@ -255,7 +253,7 @@ mod tests { graph: Graph::new("test"), outcome: Ok(crate::outcome::Outcome::success()), run_options: test_run_options(&run_dir), - run_store: test_run_store(&run_dir, &checkpoint).await, + run_store: Arc::clone(&run_store), hook_runner: None, emitter: Arc::clone(&emitter), sandbox: Arc::clone(&sandbox), @@ -270,7 +268,7 @@ mod tests { executed, &RetroOptions { run_id: test_run_id(), - run_store: test_run_store(&run_dir, &checkpoint).await, + run_store, workflow_name: "test".to_string(), goal: "Ship it".to_string(), run_dir: run_dir.clone(), @@ -286,7 +284,7 @@ mod tests { ) .await; - assert!(run_dir.join("retro.json").exists()); + assert!(retroed.run_store.get_retro().await.unwrap().is_some()); assert!(retroed.retro.is_some()); } @@ -295,7 +293,7 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); std::fs::create_dir_all(&run_dir).unwrap(); - let checkpoint = write_checkpoint(&run_dir); + let checkpoint = build_checkpoint(); let emitter = Arc::new(EventEmitter::default()); let seen = Arc::new(Mutex::new(Vec::new())); diff --git a/lib/crates/fabro-workflow/src/pipeline/types.rs b/lib/crates/fabro-workflow/src/pipeline/types.rs index 121060907..5bd31ab5c 100644 --- a/lib/crates/fabro-workflow/src/pipeline/types.rs +++ b/lib/crates/fabro-workflow/src/pipeline/types.rs @@ -194,11 +194,6 @@ impl Persisted { ) } - /// Load a previously persisted run from disk. - pub fn load(run_dir: &Path) -> Result { - super::persist::load(run_dir) - } - pub async fn load_from_store( run_store: &dyn RunStore, run_dir: &Path, diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs index cf3cc624f..8aae2f8b1 100644 --- a/lib/crates/fabro-workflow/src/run_lookup.rs +++ b/lib/crates/fabro-workflow/src/run_lookup.rs @@ -7,10 +7,8 @@ use fabro_store::{ListRunsQuery, Store}; use fabro_types::RunId; use serde::Serialize; -use crate::records::{ - Conclusion, ConclusionExt, RunRecord, RunRecordExt, StartRecord, StartRecordExt, -}; -use crate::run_status::{RunStatus, RunStatusRecord, RunStatusRecordExt, StatusReason}; +use crate::records::{RunRecord, RunRecordExt, StartRecord, StartRecordExt}; +use crate::run_status::{RunStatus, StatusReason}; #[derive(Debug, Clone, Serialize)] pub struct RunInfo { @@ -263,27 +261,7 @@ impl StatusInfo { } fn read_status(run_dir: &Path) -> StatusInfo { - if let Ok(record) = RunStatusRecord::load(&run_dir.join("status.json")) { - if record.status.is_terminal() { - if let Ok(conclusion) = Conclusion::load(&run_dir.join("conclusion.json")) { - return StatusInfo { - status: record.status, - reason: record.reason, - end_time: Some(conclusion.timestamp), - duration_ms: Some(conclusion.duration_ms), - total_cost: conclusion.total_cost, - }; - } - } - return StatusInfo { - status: record.status, - reason: record.reason, - end_time: None, - duration_ms: None, - total_cost: None, - }; - } - + let _ = run_dir; StatusInfo::simple(RunStatus::Dead) } diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs index a37d31ba4..a4986669a 100644 --- a/lib/crates/fabro-workflow/src/test_support.rs +++ b/lib/crates/fabro-workflow/src/test_support.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use chrono::Utc; use fabro_agent::Sandbox; use fabro_graphviz::graph::Graph as GvGraph; -use fabro_store::{DiskProjectingRunStore, InMemoryStore, RunStore, Store}; +use fabro_store::{InMemoryStore, Store}; use fabro_types::run::RunRecord; use crate::error::Result; @@ -48,10 +48,7 @@ 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(), - )); + let run_store = inner_store; run_store .put_run(&RunRecord { run_id: run_options.run_id,