diff --git a/lib/crates/fabro-cli/src/commands/pr/create.rs b/lib/crates/fabro-cli/src/commands/pr/create.rs index 3713eb98b..af805d1f6 100644 --- a/lib/crates/fabro-cli/src/commands/pr/create.rs +++ b/lib/crates/fabro-cli/src/commands/pr/create.rs @@ -34,23 +34,19 @@ async fn create_from( let store = store::build_store(storage_dir)?; let run = resolve_run_combined(store.as_ref(), base, &args.run_id).await?; let run_dir = run.path.clone(); - let run_store = store::open_run_reader(storage_dir, &run.run_id) - .await? - .context("Failed to open run store")?; + let run_store = store::open_run_reader(storage_dir, &run.run_id).await?; + let state = run_store.state().await?; - let record = run_store - .get_run() - .await? + let record = state + .run .context("Failed to load run record from store")?; - let start = run_store - .get_start() - .await? + let start = state + .start .context("Failed to load start record from store")?; - let conclusion = run_store - .get_conclusion() - .await? + let conclusion = state + .conclusion .context("Failed to load conclusion from store — is the run finished?")?; match conclusion.status { @@ -63,9 +59,8 @@ async fn create_from( .as_deref() .context("Run has no run_branch — was it run with git push enabled?")?; - let diff = run_store - .get_final_patch() - .await? + let diff = state + .final_patch .context("Failed to load final patch from store — no diff available")?; if diff.trim().is_empty() { bail!("final.patch is empty — nothing to create a PR for"); @@ -120,7 +115,7 @@ async fn create_from( &model, true, None, - Some(run_store.as_ref()), + run_store.as_ref(), &run_dir, None, ) diff --git a/lib/crates/fabro-cli/src/commands/pr/list.rs b/lib/crates/fabro-cli/src/commands/pr/list.rs index 698076518..30cbdcfc0 100644 --- a/lib/crates/fabro-cli/src/commands/pr/list.rs +++ b/lib/crates/fabro-cli/src/commands/pr/list.rs @@ -49,9 +49,11 @@ async fn list_from( let mut entries: Vec<(String, PullRequestRecord)> = Vec::new(); for run in &runs { - if let Ok(Some(run_store)) = store.open_run_reader(&run.run_id).await { - if let Ok(Some(record)) = run_store.get_pull_request().await { - entries.push((run.run_id.to_string(), record)); + if let Ok(run_store) = store.open_run_reader(&run.run_id).await { + if let Ok(state) = run_store.state().await { + if let Some(record) = state.pull_request { + entries.push((run.run_id.to_string(), record)); + } } } } diff --git a/lib/crates/fabro-cli/src/commands/pr/mod.rs b/lib/crates/fabro-cli/src/commands/pr/mod.rs index a165b9fe7..8032f4611 100644 --- a/lib/crates/fabro-cli/src/commands/pr/mod.rs +++ b/lib/crates/fabro-cli/src/commands/pr/mod.rs @@ -37,10 +37,9 @@ pub(crate) async fn load_pr_record( let store = store::build_store(storage_dir)?; let run = resolve_run_combined(store.as_ref(), base, run_id).await?; let run_dir = run.path; - let run_store = store::open_run_reader(storage_dir, &run.run_id) - .await? - .context("Failed to open run store")?; - let record = run_store.get_pull_request().await?.with_context(|| { + let run_store = store::open_run_reader(storage_dir, &run.run_id).await?; + let state = run_store.state().await?; + let record = state.pull_request.with_context(|| { format!("No pull request found in store. Create one first with: fabro pr create {run_id}") })?; Ok((record, run_dir)) diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index 1d6181dda..d625fe1bb 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -50,13 +50,13 @@ pub(crate) async fn attach_run( 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 { - Ok(Some(run_store)) => match run_store.list_events().await { + Ok(run_store) => match run_store.list_events().await { Ok(events) => { let verbose = run_store - .get_run() + .state() .await .ok() - .flatten() + .and_then(|state| state.run) .is_some_and(|record| record.settings.verbose_enabled()); let event_lines = events .iter() @@ -83,7 +83,6 @@ pub(crate) async fn attach_run( ); } }, - Ok(None) => {} Err(err) => { tracing::warn!( run_id = %run_id, @@ -157,13 +156,14 @@ async fn attach_run_store( } // Wait briefly for a terminal status or conclusion for _ in 0..20 { - if run_store.get_conclusion().await.ok().flatten().is_some() - || run_store - .get_status() - .await - .ok() - .flatten() - .is_some_and(|record| record.status.is_terminal()) + if run_store + .state() + .await + .ok() + .is_some_and(|state| { + state.conclusion.is_some() + || state.status.is_some_and(|record| record.status.is_terminal()) + }) { break; } @@ -237,11 +237,10 @@ async fn attach_run_store( } let terminal_status = run_store - .get_status() + .state() .await .ok() - .flatten() - .map(|record| record.status) + .and_then(|state| state.status.map(|record| record.status)) .filter(|status| status.is_terminal()); let child_alive_via_handle = engine_guard.as_mut().and_then(|guard| { @@ -749,24 +748,29 @@ fn determine_exit_code(conclusion_path: &Path, status_record: Option ExitCode { let deadline = Instant::now() + ATTACH_FINAL_STATUS_GRACE; loop { - if let Ok(Some(conclusion)) = run_store.get_conclusion().await { - let success = matches!( - conclusion.status, - StageStatus::Success | StageStatus::PartialSuccess - ); - return if success { - ExitCode::from(0) - } else { - ExitCode::from(1) - }; - } + match run_store.state().await { + Ok(state) => { + if let Some(conclusion) = state.conclusion { + let success = matches!( + conclusion.status, + StageStatus::Success | StageStatus::PartialSuccess + ); + return if success { + ExitCode::from(0) + } else { + ExitCode::from(1) + }; + } - match run_store.get_status().await { - Ok(Some(record)) if matches!(record.status, RunStatus::Succeeded) => { - return ExitCode::from(0); + match state.status { + Some(record) if matches!(record.status, RunStatus::Succeeded) => { + return ExitCode::from(0); + } + Some(record) if record.status.is_terminal() => return ExitCode::from(1), + Some(_) | None => {} + } } - Ok(Some(record)) if record.status.is_terminal() => return ExitCode::from(1), - Ok(Some(_) | None) | Err(_) => {} + Err(_) => {} } if Instant::now() >= deadline { diff --git a/lib/crates/fabro-cli/src/commands/run/cp.rs b/lib/crates/fabro-cli/src/commands/run/cp.rs index ddc4eb3d0..4f0299216 100644 --- a/lib/crates/fabro-cli/src/commands/run/cp.rs +++ b/lib/crates/fabro-cli/src/commands/run/cp.rs @@ -130,12 +130,11 @@ async fn load_sandbox( ) -> Result> { let store = store::build_store(storage_dir)?; let run = resolve_run_combined(store.as_ref(), base, run_prefix).await?; - let run_store = store::open_run_reader(storage_dir, &run.run_id) - .await? - .context("Failed to open run store")?; + let run_store = store::open_run_reader(storage_dir, &run.run_id).await?; let record = run_store - .get_sandbox() + .state() .await? + .sandbox .context("Failed to load sandbox record from store")?; info!(run_id = %run_prefix, provider = %record.provider, "Connecting to sandbox"); diff --git a/lib/crates/fabro-cli/src/commands/run/detached.rs b/lib/crates/fabro-cli/src/commands/run/detached.rs index b6ff68f4f..f476121b8 100644 --- a/lib/crates/fabro-cli/src/commands/run/detached.rs +++ b/lib/crates/fabro-cli/src/commands/run/detached.rs @@ -29,10 +29,7 @@ pub(crate) async fn execute( None => load_user_settings()?.storage_dir(), }; let store = store::build_store(&storage_dir)?; - let run_store = store - .open_run(&run_id) - .await? - .ok_or_else(|| anyhow!("Run {run_id} not found in store"))?; + let run_store = store.open_run(&run_id).await?; let run_record = run_store .get_run() .await? diff --git a/lib/crates/fabro-cli/src/commands/run/diff.rs b/lib/crates/fabro-cli/src/commands/run/diff.rs index 67e58f481..08a41ae0c 100644 --- a/lib/crates/fabro-cli/src/commands/run/diff.rs +++ b/lib/crates/fabro-cli/src/commands/run/diff.rs @@ -18,9 +18,7 @@ pub(crate) async fn run(args: DiffArgs, globals: &GlobalArgs) -> Result<()> { let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id) - .await? - .context("Failed to open run store")?; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; let patch = resolve_diff(&run.path, run_store.as_ref(), &args).await?; @@ -57,17 +55,13 @@ async fn resolve_diff( run_store: &dyn fabro_store::RunStore, args: &DiffArgs, ) -> Result { + let state = run_store.state().await?; if let Some(ref node_id) = args.node { - if let Ok(visits) = run_store.list_node_visits(node_id).await { - if let Some(visit) = visits.into_iter().max() { - if let Ok(node) = run_store - .get_node(&fabro_store::NodeVisitRef { node_id, visit }) - .await - { - if let Some(patch) = node.diff { - debug!(node_id, visit, "Reading per-node diff from store"); - return Ok(patch); - } + if let Some(visit) = state.list_node_visits(node_id).into_iter().max() { + if let Some(node) = state.node(&fabro_store::NodeVisitRef { node_id, visit }) { + if let Some(patch) = node.diff.clone() { + debug!(node_id, visit, "Reading per-node diff from projected state"); + return Ok(patch); } } } @@ -75,9 +69,8 @@ async fn resolve_diff( bail!("No diff found for node '{node_id}' — check the node ID and try again"); } - let start = run_store - .get_start() - .await? + let start = state + .start .context("Failed to load start record from store")?; let base_sha = start @@ -85,12 +78,12 @@ async fn resolve_diff( .as_deref() .ok_or_else(|| anyhow::anyhow!("This run was not git-checkpointed; no diff available"))?; - if let Ok(Some(patch)) = run_store.get_final_patch().await { + if let Some(patch) = state.final_patch { debug!("Reading final.patch from store"); return Ok(patch); } - let run_concluded = run_store.get_conclusion().await?.is_some(); + let run_concluded = state.conclusion.is_some(); if run_concluded { bail!( "Run completed but no final.patch exists — the run may not have produced any changes" @@ -98,9 +91,8 @@ async fn resolve_diff( } debug!("No final.patch found; attempting live diff from sandbox"); - let record = run_store - .get_sandbox() - .await? + let record = state + .sandbox .context("Failed to load sandbox record from store")?; info!(provider = %record.provider, "Reconnecting to sandbox for live diff"); diff --git a/lib/crates/fabro-cli/src/commands/run/fork.rs b/lib/crates/fabro-cli/src/commands/run/fork.rs index d5b699c95..6808986f8 100644 --- a/lib/crates/fabro-cli/src/commands/run/fork.rs +++ b/lib/crates/fabro-cli/src/commands/run/fork.rs @@ -21,7 +21,7 @@ pub(crate) async fn run(args: &ForkArgs, styles: &Styles, globals: &GlobalArgs) let store = Store::new(repo); let run_store = open_run_reader(&cli_settings.storage_dir(), &run_id).await?; - let timeline = build_timeline_or_rebuild(&store, run_store.as_deref(), &run_id).await?; + let timeline = build_timeline_or_rebuild(&store, Some(run_store.as_ref()), &run_id).await?; if args.list { if globals.json { diff --git a/lib/crates/fabro-cli/src/commands/run/logs.rs b/lib/crates/fabro-cli/src/commands/run/logs.rs index ed39bde0c..93047df2b 100644 --- a/lib/crates/fabro-cli/src/commands/run/logs.rs +++ b/lib/crates/fabro-cli/src/commands/run/logs.rs @@ -31,9 +31,7 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs) None => None, }; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id) - .await? - .with_context(|| format!("Run '{}' not found in store", run.run_id))?; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; let (all_lines, last_seq) = match run_store.list_events().await { Ok(events) => { let last_seq = events.last().map_or(0, |event| event.seq); @@ -190,20 +188,11 @@ async fn follow_store_logs( } async fn run_concluded(run_store: &dyn RunStore, _run_dir: &Path) -> Result { - if run_store - .get_conclusion() + let state = run_store + .state() .await - .context("Failed to read conclusion from store while following logs")? - .is_some() - { - return Ok(true); - } - - Ok(run_store - .get_status() - .await - .context("Failed to read status from store while following logs")? - .is_some_and(|record| record.status.is_terminal())) + .context("Failed to read run state from store while following logs")?; + Ok(state.conclusion.is_some() || state.status.is_some_and(|record| record.status.is_terminal())) } async fn flush_remaining_store_events( diff --git a/lib/crates/fabro-cli/src/commands/run/output.rs b/lib/crates/fabro-cli/src/commands/run/output.rs index cc9e5aa1b..dddc5acde 100644 --- a/lib/crates/fabro-cli/src/commands/run/output.rs +++ b/lib/crates/fabro-cli/src/commands/run/output.rs @@ -81,18 +81,13 @@ pub(crate) async fn print_run_summary( let (run_store, conclusion, pr_url) = match run_id.parse() { Ok(parsed_run_id) => { let run_store = store::open_run_reader(storage_dir, &parsed_run_id).await?; - let conclusion = match run_store.as_deref() { - Some(run_store) => run_store.get_conclusion().await?, - None => None, - }; - let pr_url = match run_store.as_deref() { - Some(run_store) => run_store - .get_pull_request() - .await? - .map(|record: PullRequestRecord| record.html_url), - None => None, - }; - (run_store, conclusion, pr_url) + let run_state = run_store.state().await?; + let conclusion = run_state.conclusion.clone(); + let pr_url = run_state + .pull_request + .as_ref() + .map(|record: &PullRequestRecord| record.html_url.clone()); + (Some(run_store), conclusion, pr_url) } Err(_) => (None, None, None), }; @@ -208,7 +203,7 @@ pub(crate) async fn print_final_output( styles: &Styles, ) { let checkpoint = match run_store { - Some(run_store) => run_store.get_checkpoint().await.ok().flatten(), + Some(run_store) => run_store.state().await.ok().and_then(|state| state.checkpoint), None => None, }; let Some(checkpoint) = checkpoint else { diff --git a/lib/crates/fabro-cli/src/commands/run/preview.rs b/lib/crates/fabro-cli/src/commands/run/preview.rs index 8118e454c..126b3d605 100644 --- a/lib/crates/fabro-cli/src/commands/run/preview.rs +++ b/lib/crates/fabro-cli/src/commands/run/preview.rs @@ -13,12 +13,11 @@ pub(crate) async fn run(args: PreviewArgs, globals: &GlobalArgs) -> Result<()> { let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id) - .await? - .context("Failed to open run store")?; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; let record = run_store - .get_sandbox() + .state() .await? + .sandbox .context("Failed to load sandbox record from store")?; validate_daytona_provider(&record, "Preview URLs")?; diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index b116369b2..01597c625 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs @@ -45,7 +45,7 @@ pub(crate) async fn run(args: &RewindArgs, styles: &Styles, globals: &GlobalArgs .await .ok(); - let timeline = build_timeline_or_rebuild(&store, run_store.as_deref(), &run_id).await?; + let timeline = build_timeline_or_rebuild(&store, Some(run_store.as_ref()), &run_id).await?; if args.list || args.target.is_none() { if globals.json { @@ -113,45 +113,28 @@ async fn reset_rewound_run_state( .open_run_reader(run_id) .await .map_err(|err| anyhow::anyhow!("failed to open durable store run before rewind: {err}"))?; - let store_run_record = if let Some(run_store) = existing_run_store.as_ref() { - run_store.get_run().await.ok().flatten() - } else { - None - }; - let store_start_record = if let Some(run_store) = existing_run_store.as_ref() { - run_store.get_start().await.ok().flatten() - } else { - None - }; - let store_graph = if let Some(run_store) = existing_run_store.as_ref() { - run_store.get_graph().await.ok().flatten() - } else { - None - }; + let store_run_record = existing_run_store.get_run().await.ok().flatten(); + let store_start_record = existing_run_store.get_start().await.ok().flatten(); + let store_graph = existing_run_store.get_graph().await.ok().flatten(); let run_record = store_run_record .or_else(|| RunRecord::load(run_dir).ok()) .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")?; - let previous_status = if let Some(run_store) = existing_run_store.as_ref() { - run_store - .get_status() - .await - .ok() - .flatten() - .map(|status| status.status.to_string()) - } else { - None - }; + let previous_status = existing_run_store + .get_status() + .await + .ok() + .flatten() + .map(|status| status.status.to_string()); let _ = std::fs::remove_file(run_dir.join("detached_failure.json")); let run_store = durable_store .open_run(run_id) .await - .map_err(|err| anyhow::anyhow!("failed to open durable store run for rewind reset: {err}"))? - .context("failed to reset durable store run after rewind: missing run store")?; + .map_err(|err| anyhow::anyhow!("failed to open durable store run for rewind reset: {err}"))?; run_store .reset_for_rewind() .await diff --git a/lib/crates/fabro-cli/src/commands/run/ssh.rs b/lib/crates/fabro-cli/src/commands/run/ssh.rs index 11a2c7011..5b5c2a916 100644 --- a/lib/crates/fabro-cli/src/commands/run/ssh.rs +++ b/lib/crates/fabro-cli/src/commands/run/ssh.rs @@ -17,12 +17,11 @@ pub(crate) async fn run(args: SshArgs, globals: &GlobalArgs) -> Result<()> { let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id) - .await? - .context("Failed to open run store")?; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; let record = run_store - .get_sandbox() + .state() .await? + .sandbox .context("Failed to load sandbox record from store")?; validate_daytona_provider(&record, "SSH access")?; diff --git a/lib/crates/fabro-cli/src/commands/run/start.rs b/lib/crates/fabro-cli/src/commands/run/start.rs index 4799aac68..9fd95c2a8 100644 --- a/lib/crates/fabro-cli/src/commands/run/start.rs +++ b/lib/crates/fabro-cli/src/commands/run/start.rs @@ -1,6 +1,6 @@ use std::path::Path; -use anyhow::{Result, anyhow, bail}; +use anyhow::{Result, bail}; use chrono::Utc; use fabro_types::RunId; use fabro_workflow::run_status::RunStatus; @@ -83,10 +83,8 @@ async fn ensure_startable_run(storage_dir: &Path, run_id: &RunId) -> Result<()> bail!("an engine process is still running for this run — cannot start"); } - let run_store = store::open_run_reader(storage_dir, run_id) - .await? - .ok_or_else(|| anyhow!("Cannot start run: run {run_id} not found in store"))?; - if let Some(record) = run_store.get_status().await? { + let run_store = store::open_run_reader(storage_dir, run_id).await?; + if let Some(record) = run_store.state().await?.status { if !matches!(record.status, RunStatus::Submitted | RunStatus::Starting) { bail!( "cannot start run: status is {:?}, expected submitted", diff --git a/lib/crates/fabro-cli/src/commands/run/wait.rs b/lib/crates/fabro-cli/src/commands/run/wait.rs index f4d053a8a..6f18f7143 100644 --- a/lib/crates/fabro-cli/src/commands/run/wait.rs +++ b/lib/crates/fabro-cli/src/commands/run/wait.rs @@ -34,9 +34,8 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs) let final_status = loop { let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id) - .await? - .ok_or_else(|| anyhow::anyhow!("Run {} not found in store", run_info.run_id))?; - let status = run_store.get_status().await?.map(|record| record.status); + .await?; + let status = run_store.state().await?.status.map(|record| record.status); let status = status.unwrap_or_else(|| { if started_waiting_at.elapsed() < WAIT_STARTUP_GRACE { RunStatus::Submitted @@ -65,9 +64,8 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs) }; let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id) - .await? - .ok_or_else(|| anyhow::anyhow!("Run {} not found in store", run_info.run_id))?; - let conclusion = run_store.get_conclusion().await?; + .await?; + let conclusion = run_store.state().await?.conclusion; if globals.json { let json_value = build_json_output(final_status, &run_info.run_id, conclusion.as_ref()); diff --git a/lib/crates/fabro-cli/src/commands/runs/inspect.rs b/lib/crates/fabro-cli/src/commands/runs/inspect.rs index 8fcfe9bf2..ef73bb86e 100644 --- a/lib/crates/fabro-cli/src/commands/runs/inspect.rs +++ b/lib/crates/fabro-cli/src/commands/runs/inspect.rs @@ -1,6 +1,6 @@ use std::path::{Path, PathBuf}; -use anyhow::{Result, anyhow}; +use anyhow::Result; use fabro_types::RunId; use serde::Serialize; @@ -28,9 +28,7 @@ pub(crate) async fn run(args: &InspectArgs, globals: &GlobalArgs) -> Result<()> let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id) - .await? - .ok_or_else(|| anyhow!("Run {} not found in store", run.run_id))?; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; let output = inspect_run_store(&run.run_id, &run.path, run.status, run_store.as_ref()).await; let json = serde_json::to_string_pretty(&[output])?; println!("{json}"); @@ -43,6 +41,29 @@ async fn inspect_run_store( status: RunStatus, run_store: &dyn fabro_store::RunStore, ) -> InspectOutput { + if let Ok(state) = run_store.state().await { + if let Some(snapshot) = state.to_snapshot() { + return InspectOutput { + run_id: run_id.to_string(), + run_dir: run_dir.to_path_buf(), + status: state.status.as_ref().map_or(status, |record| record.status), + run_record: serde_json::to_value(snapshot.run).ok(), + start_record: snapshot + .start + .and_then(|record| serde_json::to_value(record).ok()), + conclusion: snapshot + .conclusion + .and_then(|record| serde_json::to_value(record).ok()), + checkpoint: snapshot + .checkpoint + .and_then(|record| serde_json::to_value(record).ok()), + sandbox: snapshot + .sandbox + .and_then(|record| serde_json::to_value(record).ok()), + }; + } + } + if let Ok(Some(snapshot)) = run_store.get_snapshot().await { return InspectOutput { run_id: run_id.to_string(), diff --git a/lib/crates/fabro-cli/src/commands/runs/rm.rs b/lib/crates/fabro-cli/src/commands/runs/rm.rs index c39afe8aa..600e73051 100644 --- a/lib/crates/fabro-cli/src/commands/runs/rm.rs +++ b/lib/crates/fabro-cli/src/commands/runs/rm.rs @@ -116,7 +116,7 @@ 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<()> { let run_store = match store.open_run_reader(&run.run_id).await { - Ok(run_store) => run_store, + Ok(run_store) => Some(run_store), Err(err) => { warn!( run_id = %run.run_id, @@ -183,9 +183,8 @@ async fn load_sandbox_record( run_store: Option<&dyn fabro_store::RunStore>, ) -> Option { if let Some(run_store) = run_store { - match run_store.get_sandbox().await { - Ok(Some(record)) => return Some(record), - Ok(None) => {} + match run_store.state().await { + Ok(state) => return state.sandbox, Err(err) => { warn!(error = %err, "failed to load sandbox record from store"); } diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index 895b6c9b6..ced3466a1 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -2,7 +2,7 @@ use std::io::{ErrorKind, Write}; use std::path::{Component, Path, PathBuf}; use anyhow::{Context, Result, bail}; -use fabro_store::{NodeVisitRef, RunSnapshot, RunStore}; +use fabro_store::{NodeVisitRef, RunSnapshot, RunState, RunStore}; use fabro_workflow::run_lookup::{resolve_run_combined, runs_base}; use serde::Serialize; #[cfg(test)] @@ -18,14 +18,7 @@ pub(crate) async fn dump_command(args: &StoreDumpArgs, globals: &GlobalArgs) -> let base = runs_base(&cli_settings.storage_dir()); let store = store::build_store(&cli_settings.storage_dir())?; let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id) - .await? - .with_context(|| { - format!( - "run {} is not in the store (it may be a legacy filesystem-only run)", - run.run_id - ) - })?; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; let file_count = export_run(run_store.as_ref(), &args.output).await?; if globals.json { @@ -45,9 +38,9 @@ pub(crate) async fn dump_command(args: &StoreDumpArgs, globals: &GlobalArgs) -> } pub(crate) async fn export_run(run_store: &dyn RunStore, output_dir: &Path) -> Result { - let snapshot = run_store - .get_snapshot() - .await? + let state = run_store.state().await?; + let snapshot = state + .to_snapshot() .context("run has no data in the store")?; let output_state = inspect_output_dir(output_dir)?; @@ -66,7 +59,7 @@ pub(crate) async fn export_run(run_store: &dyn RunStore, output_dir: &Path) -> R })?; let staging_path = staging_dir.path().to_path_buf(); - let file_count = export_run_to_dir(run_store, &snapshot, &staging_path).await?; + let file_count = export_run_to_dir(run_store, &state, &snapshot, &staging_path).await?; if matches!(output_state, OutputDirState::ExistingEmpty) { std::fs::remove_dir(output_dir) @@ -86,6 +79,7 @@ pub(crate) async fn export_run(run_store: &dyn RunStore, output_dir: &Path) -> R async fn export_run_to_dir( run_store: &dyn RunStore, + state: &RunState, snapshot: &RunSnapshot, output_dir: &Path, ) -> Result { @@ -152,11 +146,11 @@ async fn export_run_to_dir( file_count += usize::from(write_optional_text_file( &output_dir.join("retro").join("prompt.md"), - run_store.get_retro_prompt().await?.as_deref(), + state.retro_prompt.as_deref(), )?); file_count += usize::from(write_optional_text_file( &output_dir.join("retro").join("response.md"), - run_store.get_retro_response().await?.as_deref(), + state.retro_response.as_deref(), )?); write_events_jsonl( @@ -165,12 +159,12 @@ async fn export_run_to_dir( )?; file_count += 1; - for (seq, checkpoint) in run_store.list_checkpoints().await? { + for (seq, checkpoint) in &state.checkpoints { write_json_file( &output_dir .join("checkpoints") .join(format!("{seq:04}.json")), - &checkpoint, + checkpoint, )?; file_count += 1; } diff --git a/lib/crates/fabro-cli/src/store.rs b/lib/crates/fabro-cli/src/store.rs index 2f1dad234..39e1344fc 100644 --- a/lib/crates/fabro-cli/src/store.rs +++ b/lib/crates/fabro-cli/src/store.rs @@ -21,7 +21,7 @@ pub(crate) fn build_store(storage_dir: &Path) -> Result> { pub(crate) async fn open_run_reader( storage_dir: &Path, run_id: &RunId, -) -> Result>> { +) -> Result> { build_store(storage_dir)? .open_run_reader(run_id) .await diff --git a/lib/crates/fabro-cli/tests/it/cmd/diff.rs b/lib/crates/fabro-cli/tests/it/cmd/diff.rs index b7e1d0147..f4ea62215 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/diff.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/diff.rs @@ -117,7 +117,7 @@ fn diff_completed_run_reads_store_final_patch_without_disk_file() { 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(); + let run_store = store.open_run(&run_id).await.unwrap(); run_store.get_final_patch().await.unwrap().unwrap() }) }); @@ -126,7 +126,7 @@ fn diff_completed_run_reads_store_final_patch_without_disk_file() { 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(); + let run_store = store.open_run(&run_id).await.unwrap(); run_store.put_final_patch(&patch).await.unwrap(); }); }); @@ -180,7 +180,7 @@ fn diff_node_reads_store_patch_without_disk_file() { 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(); + let run_store = store.open_run(&run_id).await.unwrap(); run_store .get_node(&fabro_store::NodeVisitRef { node_id: "step_one", @@ -197,7 +197,7 @@ fn diff_node_reads_store_patch_without_disk_file() { 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(); + let run_store = store.open_run(&run_id).await.unwrap(); run_store .put_node_diff( &fabro_store::NodeVisitRef { diff --git a/lib/crates/fabro-cli/tests/it/cmd/pr_view.rs b/lib/crates/fabro-cli/tests/it/cmd/pr_view.rs index ae157c2d8..d059bf1d3 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/pr_view.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/pr_view.rs @@ -76,7 +76,7 @@ fn pr_view_reads_pull_request_from_store_without_pull_request_json() { 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(); + let run_store = store.open_run(&run_id).await.unwrap(); run_store .put_pull_request(&PullRequestRecord { html_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(), diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index e0f6c2761..7784583e7 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -159,10 +159,12 @@ 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_store(&run.run_dir) - .and_then(|store| block_on(store.list_events()).ok()) - .is_none_or(|events| events.is_empty()) - { + while { + let store = run_store(&run.run_dir); + block_on(store.list_events()) + .ok() + .is_none_or(|events| events.is_empty()) + } { assert!( Instant::now() < deadline, "timed out waiting for store events for {run_id}" @@ -280,9 +282,10 @@ worktree_mode = "never" ); let run = run_local_workflow(context, &workspace_dir, "run.toml"); + let store = run_store(&run.run_dir); assert!( - run_store(&run.run_dir) - .and_then(|store| block_on(store.get_sandbox()).ok()) + block_on(store.get_sandbox()) + .ok() .flatten() .is_some() ); @@ -371,8 +374,8 @@ 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) = run_store(run_dir) - .and_then(|store| block_on(store.get_status()).ok()) + if let Some(status) = block_on(run_store(run_dir).get_status()) + .ok() .flatten() .map(|record| record.status.to_string()) { @@ -478,25 +481,30 @@ fn block_on(future: impl std::future::Future) -> T { .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()?); +fn run_store(run_dir: &Path) -> Arc { + let runs_dir = run_dir.parent().expect("run dir should have parent"); + let storage_dir = runs_dir.parent().expect("runs dir should have parent"); + let run_id: RunId = infer_run_id(run_dir).parse().expect("run id should parse"); + let object_store = Arc::new( + LocalFileSystem::new_with_prefix(storage_dir.join("store")) + .expect("test store path should be accessible"), + ); let store = Arc::new(SlateStore::new(object_store, "", Duration::from_millis(1))); - block_on(store.open_run_reader(&run_id)).ok().flatten() + block_on(store.open_run_reader(&run_id)).expect("run store should exist") } pub(crate) fn run_snapshot(run_dir: &Path) -> RunSnapshot { - run_store(run_dir) - .and_then(|store| block_on(store.get_snapshot()).ok()) + let store = run_store(run_dir); + block_on(store.get_snapshot()) + .ok() .flatten() .expect("run store snapshot should exist") } pub(crate) fn run_events(run_dir: &Path) -> Vec { - run_store(run_dir) - .and_then(|store| block_on(store.list_events()).ok()) + let store = run_store(run_dir); + block_on(store.list_events()) + .ok() .expect("run store events should exist") } diff --git a/lib/crates/fabro-cli/tests/it/scenario/mod.rs b/lib/crates/fabro-cli/tests/it/scenario/mod.rs index 0b4d5e379..6bfcfcbae 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/mod.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/mod.rs @@ -23,9 +23,9 @@ fn block_on(future: impl std::future::Future) -> T { .block_on(future) } -fn run_store(run_dir: &Path) -> Option> { - let runs_dir = run_dir.parent()?; - let storage_dir = runs_dir.parent()?; +fn run_store(run_dir: &Path) -> Arc { + let runs_dir = run_dir.parent().expect("run dir should have parent"); + let storage_dir = runs_dir.parent().expect("runs dir should have parent"); let run_id: RunId = std::fs::read_to_string(run_dir.join("id.txt")) .ok() .map(|id| id.trim().to_string()) @@ -34,17 +34,22 @@ fn run_store(run_dir: &Path) -> Option> { .file_name() .map(|name| name.to_string_lossy().to_string()) .and_then(|name| name.rsplit('-').next().map(ToOwned::to_owned)) - })? + }) + .expect("run dir should contain resolvable run id") .parse() - .ok()?; - let object_store = Arc::new(LocalFileSystem::new_with_prefix(storage_dir.join("store")).ok()?); + .expect("run id should parse"); + let object_store = Arc::new( + LocalFileSystem::new_with_prefix(storage_dir.join("store")) + .expect("test store path should be accessible"), + ); let store = Arc::new(SlateStore::new(object_store, "", Duration::from_millis(1))); - block_on(store.open_run_reader(&run_id)).ok().flatten() + block_on(store.open_run_reader(&run_id)).expect("run store should exist") } pub(super) fn run_snapshot(run_dir: &Path) -> RunSnapshot { - run_store(run_dir) - .and_then(|store| block_on(store.get_snapshot()).ok()) + let store = run_store(run_dir); + block_on(store.get_snapshot()) + .ok() .flatten() .expect("run store snapshot should exist") } diff --git a/lib/crates/fabro-retro/src/retro_agent.rs b/lib/crates/fabro-retro/src/retro_agent.rs index 3e63350af..1f26115ba 100644 --- a/lib/crates/fabro-retro/src/retro_agent.rs +++ b/lib/crates/fabro-retro/src/retro_agent.rs @@ -137,7 +137,7 @@ pub fn build_retro_prompt(retro_data_dir: &str) -> String { /// files via tool access, then calls `submit_retro` with its analysis. pub async fn run_retro_agent( sandbox: &Arc, - run_store: Option<&dyn RunStore>, + run_store: &dyn RunStore, run_dir: &Path, llm_client: &Client, provider: Provider, @@ -286,32 +286,24 @@ pub fn dry_run_narrative() -> RetroNarrative { } async fn write_retro_prompt( - run_store: Option<&dyn RunStore>, + run_store: &dyn RunStore, retro_dir: &Path, prompt: &str, ) -> anyhow::Result<()> { - if let Some(store) = run_store { - if let Err(err) = store.put_retro_prompt(prompt).await { - tracing::warn!(error = %err, "Failed to save retro prompt to store"); - std::fs::write(retro_dir.join("prompt.md"), prompt)?; - } - } else { + if let Err(err) = run_store.put_retro_prompt(prompt).await { + tracing::warn!(error = %err, "Failed to save retro prompt to store"); std::fs::write(retro_dir.join("prompt.md"), prompt)?; } Ok(()) } async fn write_retro_response( - run_store: Option<&dyn RunStore>, + run_store: &dyn RunStore, retro_dir: &Path, response: &str, ) -> anyhow::Result<()> { - if let Some(store) = run_store { - if let Err(err) = store.put_retro_response(response).await { - tracing::warn!(error = %err, "Failed to save retro response to store"); - std::fs::write(retro_dir.join("response.md"), response)?; - } - } else { + if let Err(err) = run_store.put_retro_response(response).await { + tracing::warn!(error = %err, "Failed to save retro response to store"); std::fs::write(retro_dir.join("response.md"), response)?; } Ok(()) @@ -393,7 +385,7 @@ fn build_profile(provider: Provider, model: &str) -> Box { async fn upload_data_files( sandbox: &Arc, - run_store: Option<&dyn RunStore>, + run_store: &dyn RunStore, _run_dir: &Path, target_dir: &str, ) -> anyhow::Result<()> { @@ -403,11 +395,7 @@ async fn upload_data_files( .await .map_err(|e| anyhow::anyhow!("Failed to create retro data dir: {e}"))?; - let Some(store) = run_store else { - anyhow::bail!("retro analysis now requires a run store"); - }; - - let progress_content = match store.list_events().await { + let progress_content = match run_store.list_events().await { Ok(envelopes) => { let lines: Vec = envelopes .into_iter() @@ -428,7 +416,7 @@ async fn upload_data_files( .map_err(|e| anyhow::anyhow!("Failed to upload progress.jsonl: {e}"))?; } - let checkpoint_content = store + let checkpoint_content = run_store .get_checkpoint() .await .map_err(|e| anyhow::anyhow!("Failed to load checkpoint from store: {e}"))? @@ -436,7 +424,7 @@ async fn upload_data_files( .transpose()?; upload_file(sandbox, target_dir, "checkpoint.json", checkpoint_content).await?; - let run_content = store + let run_content = run_store .get_run() .await .map_err(|e| anyhow::anyhow!("Failed to load run metadata from store: {e}"))? @@ -444,7 +432,7 @@ async fn upload_data_files( .transpose()?; upload_file(sandbox, target_dir, "run.json", run_content).await?; - let start_content = store + let start_content = run_store .get_start() .await .map_err(|e| anyhow::anyhow!("Failed to load start metadata from store: {e}"))? diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 55f6de369..515e6045e 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -680,18 +680,7 @@ async fn execute_run(state: Arc, run_id: RunId) { } let run_store = match state.store.open_run(&run_id).await { - Ok(Some(run_store)) => run_store, - Ok(None) => { - tracing::error!(run_id = %run_id, "Run store missing"); - let mut runs = state.runs.lock().expect("runs lock poisoned"); - if let Some(managed_run) = runs.get_mut(&run_id) { - managed_run.status = RunStatus::Failed; - managed_run.error = Some("Run store missing".to_string()); - clear_live_run_state(managed_run); - } - state.scheduler_notify.notify_one(); - return; - } + Ok(run_store) => run_store, Err(e) => { tracing::error!(run_id = %run_id, error = %e, "Failed to open run store"); let mut runs = state.runs.lock().expect("runs lock poisoned"); @@ -754,10 +743,10 @@ async fn execute_run(state: Arc, run_id: RunId) { }; // Save final checkpoint - let checkpoint = match run_store.get_checkpoint().await { - Ok(checkpoint) => checkpoint, + let checkpoint = match run_store.state().await { + Ok(state) => state.checkpoint, Err(err) => { - tracing::warn!(run_id = %run_id, error = %err, "Failed to load checkpoint from store"); + tracing::warn!(run_id = %run_id, error = %err, "Failed to load run state from store"); None } }; @@ -1069,13 +1058,32 @@ async fn get_checkpoint( Ok(id) => id, Err(response) => return response, }; - let runs = state.runs.lock().expect("runs lock poisoned"); - match runs.get(&id) { - Some(managed_run) => match &managed_run.checkpoint { - Some(cp) => (StatusCode::OK, Json(cp.clone())).into_response(), - None => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), + let live_checkpoint = { + let runs = state.runs.lock().expect("runs lock poisoned"); + match runs.get(&id) { + Some(managed_run) => managed_run.checkpoint.clone(), + None => return ApiError::not_found("Run not found.").into_response(), + } + }; + if let Some(cp) = live_checkpoint { + return (StatusCode::OK, Json(cp)).into_response(); + } + + match state.store.open_run_reader(&id).await { + Ok(run_store) => match run_store.state().await { + Ok(run_state) => match run_state.checkpoint { + Some(cp) => (StatusCode::OK, Json(cp)).into_response(), + None => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), + }, + Err(err) => { + tracing::warn!(run_id = %id, error = %err, "Failed to load checkpoint state from store"); + (StatusCode::OK, Json(serde_json::json!(null))).into_response() + } }, - None => ApiError::not_found("Run not found.").into_response(), + Err(err) => { + tracing::warn!(run_id = %id, error = %err, "Failed to open run store reader"); + ApiError::not_found("Run not found.").into_response() + } } } @@ -1564,18 +1572,19 @@ async fn get_retro( } 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) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), + Ok(run_store) => match run_store.state().await { + Ok(run_state) => match run_state.retro { + Some(retro) => (StatusCode::OK, Json(retro)).into_response(), + 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"); + tracing::warn!(run_id = %id, error = %err, "Failed to load retro state from store"); (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"); - (StatusCode::OK, Json(serde_json::json!(null))).into_response() + ApiError::not_found("Run not found.").into_response() } } } @@ -1603,15 +1612,27 @@ async fn get_graph( Ok(id) => id, Err(response) => return response, }; - let dot_source = { + let live_dot_source = { let runs = state.runs.lock().expect("runs lock poisoned"); match runs.get(&id) { Some(managed_run) => managed_run.dot_source.clone(), None => return ApiError::not_found("Run not found.").into_response(), } }; + if !live_dot_source.is_empty() { + return render_dot_svg(&live_dot_source).await; + } - render_dot_svg(&dot_source).await + match state.store.open_run_reader(&id).await { + Ok(run_store) => match run_store.state().await { + Ok(run_state) => match run_state.graph_source { + Some(dot_source) => render_dot_svg(&dot_source).await, + None => ApiError::new(StatusCode::NOT_FOUND, "Graph not found.").into_response(), + }, + Err(err) => ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response(), + }, + Err(_) => ApiError::new(StatusCode::NOT_FOUND, "Run not found.").into_response(), + } } #[cfg(test)] diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index 07c3c3fc5..7d6ad1659 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -9,12 +9,14 @@ use futures::Stream; mod error; mod keys; mod memory; +mod run_state; mod runtime; mod slate; mod types; pub use error::{Result, StoreError}; pub use memory::InMemoryStore; +pub use run_state::{NodeState, RunState}; pub use runtime::RuntimeState; pub use slate::SlateStore; pub use types::{ @@ -42,8 +44,8 @@ pub trait Store: Send + Sync { created_at: DateTime, run_dir: Option<&str>, ) -> Result>; - async fn open_run(&self, run_id: &RunId) -> Result>>; - async fn open_run_reader(&self, run_id: &RunId) -> Result>>; + async fn open_run(&self, run_id: &RunId) -> Result>; + async fn open_run_reader(&self, run_id: &RunId) -> Result>; async fn list_runs(&self, query: &ListRunsQuery) -> Result>; async fn delete_run(&self, run_id: &RunId) -> Result<()>; } @@ -146,5 +148,6 @@ pub trait RunStore: Send + Sync { async fn list_assets(&self, node: &NodeVisitRef<'_>) -> Result>; async fn list_all_assets(&self) -> Result>; + async fn state(&self) -> Result; async fn get_snapshot(&self) -> Result>; } diff --git a/lib/crates/fabro-store/src/memory.rs b/lib/crates/fabro-store/src/memory.rs index febc85ffe..690263d55 100644 --- a/lib/crates/fabro-store/src/memory.rs +++ b/lib/crates/fabro-store/src/memory.rs @@ -13,9 +13,10 @@ use tokio_stream::StreamExt as _; use tokio_stream::wrappers::UnboundedReceiverStream; use crate::keys; +use crate::run_state::EventProjectionCache; use crate::{ CatalogRecord, EventEnvelope, EventPayload, ListRunsQuery, NodeOutcomeRecord, NodeSnapshot, - NodeVisitRef, Result, RunSnapshot, RunStore, RunSummary, Store, StoreError, + NodeVisitRef, Result, RunSnapshot, RunState, RunStore, RunSummary, Store, StoreError, }; use fabro_types::{ Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunId, RunRecord, @@ -41,6 +42,7 @@ struct InMemoryRunStore { event_seq: AtomicU32, checkpoint_seq: AtomicU32, watchers: Mutex>>, + projection_cache: Mutex, } impl InMemoryRunStore { @@ -65,6 +67,7 @@ impl InMemoryRunStore { event_seq: AtomicU32::new(1), checkpoint_seq: AtomicU32::new(1), watchers: Mutex::new(Vec::new()), + projection_cache: Mutex::new(EventProjectionCache::default()), }) } @@ -264,6 +267,20 @@ impl InMemoryRunStore { } Ok(()) } + + async fn projected_state(&self) -> Result { + let next_seq = { + let cache = self.projection_cache.lock().await; + cache.last_seq.saturating_add(1) + }; + let events = self.list_events_from_inner(next_seq).await?; + let mut cache = self.projection_cache.lock().await; + for event in &events { + cache.state.apply_event(event)?; + cache.last_seq = event.seq; + } + Ok(cache.state.clone()) + } } #[async_trait] @@ -302,14 +319,14 @@ impl Store for InMemoryStore { Ok(run_store as Arc) } - async fn open_run(&self, run_id: &RunId) -> Result>> { + async fn open_run(&self, run_id: &RunId) -> Result> { let runs = self.runs.lock().await; - Ok(runs - .get(run_id) - .map(|catalog| Arc::clone(&catalog.run_store) as Arc)) + runs.get(run_id) + .map(|catalog| Arc::clone(&catalog.run_store) as Arc) + .ok_or_else(|| StoreError::RunNotFound(run_id.to_string())) } - async fn open_run_reader(&self, run_id: &RunId) -> Result>> { + async fn open_run_reader(&self, run_id: &RunId) -> Result> { self.open_run(run_id).await } @@ -324,8 +341,7 @@ impl Store for InMemoryStore { if !matches_query(&catalog.record.created_at, query) { continue; } - let data = catalog.run_store.snapshot_data().await; - summaries.push(build_run_summary(&catalog.record, &data)?); + summaries.push(catalog.run_store.state().await?.build_summary(&catalog.record)); } summaries.sort_by(|a, b| b.created_at.cmp(&a.created_at)); Ok(summaries) @@ -665,9 +681,21 @@ impl RunStore for InMemoryRunStore { self.list_all_assets_inner().await } - async fn get_snapshot(&self) -> Result> { + async fn state(&self) -> Result { + let mut state = self.projected_state().await?; let data = self.snapshot_data().await; - self.build_snapshot_from_data(&data) + state.merge_legacy( + self.build_snapshot_from_data(&data)?, + read_text(&data, keys::graph())?, + read_text(&data, keys::retro_prompt())?, + read_text(&data, keys::retro_response())?, + self.list_checkpoints_inner().await?, + ); + Ok(state) + } + + async fn get_snapshot(&self) -> Result> { + self.state().await.map(|state| state.to_snapshot()) } } @@ -704,48 +732,6 @@ fn read_text(data: &BTreeMap>, key: &str) -> Result>, -) -> Result { - let run = read_json::(data, keys::run())?; - let start = read_json::(data, keys::start())?; - let status = read_json::(data, keys::status())?; - let conclusion = read_json::(data, keys::conclusion())?; - - let workflow_name = run.as_ref().map(|run| { - if run.graph.name.is_empty() { - "unnamed".to_string() - } else { - run.graph.name.clone() - } - }); - let goal = run.as_ref().and_then(|run| { - let goal = run.graph.goal(); - (!goal.is_empty()).then(|| goal.to_string()) - }); - - Ok(RunSummary { - run_id: record.run_id, - created_at: record.created_at, - db_prefix: record.db_prefix.clone(), - run_dir: record.run_dir.clone(), - workflow_name, - workflow_slug: run.as_ref().and_then(|run| run.workflow_slug.clone()), - goal, - labels: run - .as_ref() - .map(|run| run.labels.clone()) - .unwrap_or_default(), - host_repo_path: run.as_ref().and_then(|run| run.host_repo_path.clone()), - start_time: start.map(|start| start.start_time), - status: status.as_ref().map(|status| status.status), - status_reason: status.and_then(|status| status.reason), - duration_ms: conclusion.as_ref().map(|conclusion| conclusion.duration_ms), - total_cost: conclusion.and_then(|conclusion| conclusion.total_cost), - }) -} - #[cfg(test)] mod tests { use super::*; @@ -914,6 +900,26 @@ mod tests { } } + fn event_payload( + run_id: &str, + ts: &str, + event: &str, + node_id: Option<&str>, + properties: serde_json::Value, + ) -> EventPayload { + let mut value = serde_json::json!({ + "id": format!("evt-{event}-{ts}"), + "ts": ts, + "run_id": test_run_id(run_id).to_string(), + "event": event, + "properties": properties, + }); + if let Some(node_id) = node_id { + value["node_id"] = serde_json::Value::String(node_id.to_string()); + } + EventPayload::new(value, &test_run_id(run_id)).unwrap() + } + #[tokio::test] async fn create_run_put_get_and_snapshot_round_trip() { let store = InMemoryStore::default(); @@ -1102,6 +1108,317 @@ mod tests { assert_eq!(snapshot.pull_request, Some(pull_request)); } + #[tokio::test] + async fn state_projects_event_stream_and_compat_fields() { + let store = InMemoryStore::default(); + let created_at = dt("2026-03-27T12:00:00Z"); + let run = store + .create_run(&test_run_id("run-1"), created_at, None) + .await + .unwrap(); + let run_record = sample_run_record("run-1", created_at); + let retro = sample_retro("run-1"); + + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:00Z", + "run.created", + None, + serde_json::json!({ + "settings": run_record.settings, + "graph": run_record.graph, + "workflow_source": "digraph night_sky {}", + "workflow_slug": run_record.workflow_slug, + "working_directory": run_record.working_directory, + "host_repo_path": run_record.host_repo_path, + "base_branch": run_record.base_branch, + "labels": run_record.labels, + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:05Z", + "run.started", + None, + serde_json::json!({ + "run_branch": "fabro/run/demo", + "base_sha": "abc123" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:06Z", + "run.running", + None, + serde_json::json!({}), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:07Z", + "stage.prompt", + Some("code"), + serde_json::json!({ + "visit": 2, + "text": "Plan the fix", + "mode": "prompt", + "provider": "openai", + "model": "gpt-5.4" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08Z", + "stage.completed", + Some("code"), + serde_json::json!({ + "status": "success", + "notes": "all good", + "response": "Implemented", + "files_touched": ["src/lib.rs"], + "node_visits": {"code": 2} + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:09Z", + "checkpoint.completed", + Some("code"), + serde_json::json!({ + "status": "success", + "current_node": "code", + "completed_nodes": ["plan"], + "context_values": {"artifact": {"kind": "summary"}}, + "next_node_id": "review", + "git_commit_sha": "def456", + "node_visits": {"code": 2}, + "diff": "diff --git a/src/lib.rs b/src/lib.rs" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:10Z", + "sandbox.initialized", + None, + serde_json::json!({ + "provider": "local", + "working_directory": "/tmp/night-sky", + "identifier": "sandbox-1", + "host_working_directory": "/tmp/night-sky" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:11Z", + "retro.started", + None, + serde_json::json!({ + "prompt": "How did it go?" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:12Z", + "retro.completed", + None, + serde_json::json!({ + "response": "Smooth enough", + "retro": retro + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:13Z", + "pull_request.created", + None, + serde_json::json!({ + "pr_url": "https://github.com/fabro-sh/fabro/pull/123", + "pr_number": 123, + "owner": "fabro-sh", + "repo": "fabro", + "base_branch": "main", + "head_branch": "fabro/run/demo", + "title": "Map the constellations", + "draft": false + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:15Z", + "run.completed", + None, + serde_json::json!({ + "duration_ms": 3210, + "artifact_count": 1, + "status": "success", + "total_cost": 1.25, + "final_git_commit_sha": "feedbeef", + "final_patch": "diff --git a/src/lib.rs b/src/lib.rs\n" + }), + )) + .await + .unwrap(); + + let state = run.state().await.unwrap(); + assert_eq!(state.run.as_ref().map(|run| run.run_id), Some(test_run_id("run-1"))); + assert_eq!(state.graph_source.as_deref(), Some("digraph night_sky {}")); + assert_eq!(state.start.as_ref().and_then(|start| start.run_branch.as_deref()), Some("fabro/run/demo")); + assert_eq!(state.status.as_ref().map(|status| status.status), Some(RunStatus::Succeeded)); + assert_eq!(state.checkpoint.as_ref().map(|checkpoint| checkpoint.current_node.as_str()), Some("code")); + assert_eq!(state.checkpoints.len(), 1); + assert_eq!(state.final_patch.as_deref(), Some("diff --git a/src/lib.rs b/src/lib.rs\n")); + assert_eq!(state.retro_prompt.as_deref(), Some("How did it go?")); + assert_eq!(state.retro_response.as_deref(), Some("Smooth enough")); + assert_eq!(state.pull_request.as_ref().map(|pr| pr.number), Some(123)); + assert_eq!(state.sandbox.as_ref().map(|sandbox| sandbox.provider.as_str()), Some("local")); + assert_eq!(state.list_node_visits("code"), vec![2]); + let node = state + .node(&NodeVisitRef { + node_id: "code", + visit: 2, + }) + .unwrap(); + assert_eq!(node.prompt.as_deref(), Some("Plan the fix")); + assert_eq!(node.response.as_deref(), Some("Implemented")); + assert_eq!(node.diff.as_deref(), Some("diff --git a/src/lib.rs b/src/lib.rs")); + assert_eq!( + node.provider_used.as_ref().and_then(|value| value.get("provider")).and_then(|value| value.as_str()), + Some("openai") + ); + } + + #[tokio::test] + async fn state_rewind_keeps_active_projection_only() { + let store = InMemoryStore::default(); + let run = store + .create_run(&test_run_id("run-1"), dt("2026-03-27T12:00:00Z"), None) + .await + .unwrap(); + + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:00Z", + "run.created", + None, + serde_json::json!({ + "settings": Settings::default(), + "graph": Graph::new("night-sky"), + "working_directory": "/tmp/night-sky", + "labels": {} + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:01Z", + "stage.prompt", + Some("code"), + serde_json::json!({ + "visit": 1, + "text": "before rewind" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:02Z", + "pull_request.created", + None, + serde_json::json!({ + "pr_url": "https://github.com/fabro-sh/fabro/pull/123", + "pr_number": 123, + "owner": "fabro-sh", + "repo": "fabro", + "base_branch": "main", + "head_branch": "fabro/run/demo", + "title": "Map the constellations", + "draft": false + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:03Z", + "run.completed", + None, + serde_json::json!({ + "duration_ms": 10, + "artifact_count": 0, + "status": "success", + "final_patch": "old patch" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:04Z", + "run.rewound", + None, + serde_json::json!({ + "target_checkpoint_ordinal": 1, + "target_node_id": "plan", + "target_visit": 1 + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:05Z", + "checkpoint.completed", + Some("plan"), + serde_json::json!({ + "status": "success", + "current_node": "plan", + "completed_nodes": [], + "node_visits": {"plan": 1} + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:06Z", + "run.submitted", + None, + serde_json::json!({}), + )) + .await + .unwrap(); + + let state = run.state().await.unwrap(); + assert_eq!(state.status.as_ref().map(|status| status.status), Some(RunStatus::Submitted)); + assert!(state.conclusion.is_none()); + assert!(state.final_patch.is_none()); + assert!(state.pull_request.is_none()); + assert_eq!(state.checkpoints.len(), 1); + assert_eq!(state.checkpoint.as_ref().map(|checkpoint| checkpoint.current_node.as_str()), Some("plan")); + assert!(state.list_node_ids().is_empty()); + } + #[tokio::test] async fn list_artifact_values_and_all_assets_include_asset_only_visits() { let store = InMemoryStore::default(); @@ -1379,11 +1696,10 @@ mod tests { store.delete_run(&test_run_id("run-1")).await.unwrap(); store.delete_run(&test_run_id("run-1")).await.unwrap(); assert!( - store - .open_run(&test_run_id("run-1")) - .await - .unwrap() - .is_none() + matches!( + store.open_run(&test_run_id("run-1")).await, + Err(StoreError::RunNotFound(_)) + ) ); } diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs new file mode 100644 index 000000000..e1770f6f7 --- /dev/null +++ b/lib/crates/fabro-store/src/run_state.rs @@ -0,0 +1,734 @@ +use std::collections::{BTreeMap, HashMap}; +use std::path::PathBuf; +use std::str::FromStr; + +use chrono::{DateTime, Utc}; +use serde_json::Value; + +use crate::{ + CatalogRecord, EventEnvelope, NodeOutcomeRecord, NodeSnapshot, NodeVisitRef, Result, + RunSnapshot, RunSummary, StoreError, +}; +use fabro_types::{ + Checkpoint, Conclusion, FailureSignature, NodeStatusRecord, Outcome, PullRequestRecord, Retro, + RunId, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, StageStatus, StartRecord, + StatusReason, +}; + +#[derive(Debug, Clone, Default)] +pub struct RunState { + pub run: Option, + pub graph_source: Option, + pub start: Option, + pub status: Option, + pub checkpoint: Option, + pub checkpoints: Vec<(u32, Checkpoint)>, + pub conclusion: Option, + pub retro: Option, + pub retro_prompt: Option, + pub retro_response: Option, + pub sandbox: Option, + pub final_patch: Option, + pub pull_request: Option, + pub nodes: HashMap<(String, u32), NodeState>, + pub last_git_sha: Option, +} + +#[derive(Debug, Clone, Default)] +pub struct NodeState { + pub prompt: Option, + pub response: Option, + pub status: Option, + pub outcome: Option, + pub provider_used: Option, + pub diff: Option, + pub script_invocation: Option, + pub script_timing: Option, + pub parallel_results: Option, + pub stdout: Option, + pub stderr: Option, +} + +#[derive(Debug, Clone, Default)] +pub(crate) struct EventProjectionCache { + pub last_seq: u32, + pub state: RunState, +} + +impl RunState { + pub fn apply_events(events: &[EventEnvelope]) -> Result { + let mut state = Self::default(); + for event in events { + state.apply_event(event)?; + } + Ok(state) + } + + pub fn apply_event(&mut self, event: &EventEnvelope) -> Result<()> { + let value = event.payload.as_value(); + let ts = parse_ts(value)?; + let event_name = value + .get("event") + .and_then(Value::as_str) + .ok_or_else(|| StoreError::InvalidEvent("event payload missing event name".into()))?; + let run_id = parse_run_id(value)?; + let properties = value + .get("properties") + .and_then(Value::as_object) + .cloned() + .unwrap_or_default(); + + match event_name { + "run.created" => { + let settings = required_json::(&properties, "settings")?; + let graph = required_json::(&properties, "graph")?; + let working_directory = + required_string(&properties, "working_directory").map(PathBuf::from)?; + let labels = optional_json::>(&properties, "labels")? + .unwrap_or_default() + .into_iter() + .collect::>(); + self.run = Some(RunRecord { + run_id, + created_at: ts, + settings, + graph, + workflow_slug: optional_string(&properties, "workflow_slug"), + working_directory, + host_repo_path: optional_string(&properties, "host_repo_path"), + base_branch: optional_string(&properties, "base_branch"), + labels, + }); + self.graph_source = optional_string(&properties, "workflow_source"); + } + "run.started" => { + self.start = Some(StartRecord { + run_id, + start_time: ts, + run_branch: optional_string(&properties, "run_branch"), + base_sha: optional_string(&properties, "base_sha"), + }); + } + "run.submitted" => { + self.status = Some(run_status_record(RunStatus::Submitted, &properties, ts)?); + } + "run.starting" => { + self.status = Some(run_status_record(RunStatus::Starting, &properties, ts)?); + } + "run.running" => { + self.status = Some(run_status_record(RunStatus::Running, &properties, ts)?); + } + "run.removing" => { + self.status = Some(run_status_record(RunStatus::Removing, &properties, ts)?); + } + "run.completed" => { + self.status = Some(run_status_record(RunStatus::Succeeded, &properties, ts)?); + self.conclusion = Some(conclusion_from_completed(&properties, ts)?); + self.final_patch = optional_string(&properties, "final_patch"); + self.last_git_sha = optional_string(&properties, "final_git_commit_sha") + .or_else(|| self.last_git_sha.clone()); + } + "run.failed" => { + self.status = Some(run_status_record(RunStatus::Failed, &properties, ts)?); + self.conclusion = Some(conclusion_from_failed(&properties, ts)); + self.last_git_sha = + optional_string(&properties, "git_commit_sha").or_else(|| self.last_git_sha.clone()); + } + "run.rewound" => { + self.reset_for_rewind(); + self.last_git_sha = + optional_string(&properties, "run_commit_sha").or_else(|| self.last_git_sha.clone()); + } + "checkpoint.completed" => { + let checkpoint = checkpoint_from_properties(&properties, ts)?; + self.last_git_sha = checkpoint.git_commit_sha.clone().or_else(|| self.last_git_sha.clone()); + if let Some(node_id) = value.get("node_id").and_then(Value::as_str) { + let visit = checkpoint + .node_visits + .get(node_id) + .and_then(|visit| u32::try_from(*visit).ok()) + .unwrap_or(1); + if let Some(diff) = optional_string(&properties, "diff") { + self.node_mut(node_id, visit).diff = Some(diff); + } + } + self.checkpoint = Some(checkpoint.clone()); + self.checkpoints.push((event.seq, checkpoint)); + } + "sandbox.initialized" => { + self.sandbox = Some(SandboxRecord { + provider: required_string(&properties, "provider")?, + working_directory: required_string(&properties, "working_directory")?, + identifier: optional_string(&properties, "identifier"), + host_working_directory: optional_string(&properties, "host_working_directory"), + container_mount_point: optional_string(&properties, "container_mount_point"), + }); + } + "retro.started" => { + self.retro_prompt = optional_string(&properties, "prompt"); + } + "retro.completed" => { + self.retro_response = optional_string(&properties, "response"); + self.retro = optional_json::(&properties, "retro")?; + } + "pull_request.created" => { + self.pull_request = Some(PullRequestRecord { + html_url: required_string(&properties, "pr_url")?, + number: required_u64(&properties, "pr_number")?, + owner: required_string(&properties, "owner")?, + repo: required_string(&properties, "repo")?, + base_branch: required_string(&properties, "base_branch")?, + head_branch: required_string(&properties, "head_branch")?, + title: required_string(&properties, "title")?, + }); + } + "stage.prompt" => { + let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { + return Ok(()); + }; + let visit = required_u32(&properties, "visit")?; + self.node_mut(node_id, visit).prompt = optional_string(&properties, "text"); + self.node_mut(node_id, visit).provider_used = + provider_used_from_prompt(&properties); + } + "prompt.completed" => { + let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { + return Ok(()); + }; + let visit = self.current_visit_for(node_id).unwrap_or(1); + self.node_mut(node_id, visit).response = optional_string(&properties, "response"); + } + "stage.completed" => { + let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { + return Ok(()); + }; + let visit = stage_visit(node_id, &properties, self).unwrap_or(1); + let response = optional_string(&properties, "response"); + let outcome = stage_outcome_from_properties(&properties)?; + let status = node_status_from_outcome(&outcome, ts); + let node = self.node_mut(node_id, visit); + node.response = response; + node.status = Some(status); + node.outcome = Some(outcome); + } + "stage.failed" => { + let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { + return Ok(()); + }; + let visit = self.current_visit_for(node_id).unwrap_or(1); + let failure = optional_json::(&properties, "failure")?; + let failure_reason = failure.as_ref().map(|detail| detail.message.clone()); + let node = self.node_mut(node_id, visit); + node.status = Some(NodeStatusRecord { + status: StageStatus::Fail, + notes: None, + failure_reason: failure_reason.clone(), + timestamp: ts, + }); + node.outcome = Some(Outcome { + status: StageStatus::Fail, + preferred_label: None, + suggested_next_ids: Vec::new(), + context_updates: HashMap::new(), + jump_to_node: None, + notes: None, + failure, + usage: None, + files_touched: Vec::new(), + duration_ms: None, + }); + } + "agent.session.started" | "agent.cli.started" => { + let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { + return Ok(()); + }; + let visit = required_u32(&properties, "visit")?; + self.node_mut(node_id, visit).provider_used = + provider_used_from_agent_event(event_name, &properties); + } + "command.started" => { + let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { + return Ok(()); + }; + let visit = self.current_visit_for(node_id).unwrap_or(1); + self.node_mut(node_id, visit).script_invocation = + Some(Value::Object(properties.clone())); + } + "command.completed" => { + let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { + return Ok(()); + }; + let visit = self.current_visit_for(node_id).unwrap_or(1); + let node = self.node_mut(node_id, visit); + node.stdout = optional_string(&properties, "stdout"); + node.stderr = optional_string(&properties, "stderr"); + node.script_timing = Some(Value::Object(properties.clone())); + } + _ => {} + } + + Ok(()) + } + + pub fn merge_legacy( + &mut self, + snapshot: Option, + graph_source: Option, + retro_prompt: Option, + retro_response: Option, + checkpoints: Vec<(u32, Checkpoint)>, + ) { + let Some(snapshot) = snapshot else { + self.graph_source = self.graph_source.clone().or(graph_source); + self.retro_prompt = self.retro_prompt.clone().or(retro_prompt); + self.retro_response = self.retro_response.clone().or(retro_response); + if self.checkpoints.is_empty() { + self.checkpoints = checkpoints; + } + if self.checkpoint.is_none() { + self.checkpoint = self.checkpoints.last().map(|(_, checkpoint)| checkpoint.clone()); + } + return; + }; + + if self.run.is_none() { + self.run = Some(snapshot.run); + } + self.start = self.start.clone().or(snapshot.start); + self.status = self.status.clone().or(snapshot.status); + self.checkpoint = self.checkpoint.clone().or(snapshot.checkpoint); + self.conclusion = self.conclusion.clone().or(snapshot.conclusion); + self.retro = self.retro.clone().or(snapshot.retro); + self.graph_source = self.graph_source.clone().or(snapshot.graph).or(graph_source); + self.sandbox = self.sandbox.clone().or(snapshot.sandbox); + self.final_patch = self.final_patch.clone().or(snapshot.final_patch); + self.pull_request = self.pull_request.clone().or(snapshot.pull_request); + self.retro_prompt = self.retro_prompt.clone().or(retro_prompt); + self.retro_response = self.retro_response.clone().or(retro_response); + if self.checkpoints.is_empty() { + self.checkpoints = checkpoints; + } + if self.checkpoint.is_none() { + self.checkpoint = self.checkpoints.last().map(|(_, checkpoint)| checkpoint.clone()); + } + + for node in snapshot.nodes { + let entry = self + .nodes + .entry((node.node_id.clone(), node.visit)) + .or_default(); + merge_node_snapshot(entry, node); + } + } + + pub fn node(&self, node: &NodeVisitRef<'_>) -> Option<&NodeState> { + self.nodes.get(&(node.node_id.to_string(), node.visit)) + } + + pub fn list_node_ids(&self) -> Vec { + let mut ids = self + .nodes + .keys() + .map(|(node_id, _)| node_id.clone()) + .collect::>(); + ids.sort(); + ids.dedup(); + ids + } + + pub fn list_node_visits(&self, node_id: &str) -> Vec { + let mut visits = self + .nodes + .keys() + .filter(|(current_node_id, _)| current_node_id == node_id) + .map(|(_, visit)| *visit) + .collect::>(); + visits.sort_unstable(); + visits.dedup(); + visits + } + + pub fn to_snapshot(&self) -> Option { + let run = self.run.clone()?; + let mut node_keys = self.nodes.keys().cloned().collect::>(); + node_keys.sort(); + let nodes = node_keys + .into_iter() + .filter_map(|(node_id, visit)| { + self.nodes.get(&(node_id.clone(), visit)).map(|node| NodeSnapshot { + node_id, + visit, + prompt: node.prompt.clone(), + response: node.response.clone(), + status: node.status.clone(), + outcome: node.outcome.clone(), + provider_used: node.provider_used.clone(), + diff: node.diff.clone(), + script_invocation: node.script_invocation.clone(), + script_timing: node.script_timing.clone(), + parallel_results: node.parallel_results.clone(), + stdout: node.stdout.clone(), + stderr: node.stderr.clone(), + }) + }) + .collect(); + + Some(RunSnapshot { + run, + start: self.start.clone(), + status: self.status.clone(), + checkpoint: self.checkpoint.clone(), + conclusion: self.conclusion.clone(), + retro: self.retro.clone(), + graph: self.graph_source.clone(), + sandbox: self.sandbox.clone(), + final_patch: self.final_patch.clone(), + pull_request: self.pull_request.clone(), + nodes, + }) + } + + pub fn build_summary(&self, catalog: &CatalogRecord) -> RunSummary { + let workflow_name = self.run.as_ref().map(|run| { + if run.graph.name.is_empty() { + "unnamed".to_string() + } else { + run.graph.name.clone() + } + }); + let goal = self.run.as_ref().and_then(|run| { + let goal = run.graph.goal(); + (!goal.is_empty()).then(|| goal.to_string()) + }); + RunSummary { + run_id: catalog.run_id, + created_at: catalog.created_at, + db_prefix: catalog.db_prefix.clone(), + run_dir: catalog.run_dir.clone(), + workflow_name, + workflow_slug: self.run.as_ref().and_then(|run| run.workflow_slug.clone()), + goal, + labels: self + .run + .as_ref() + .map(|run| run.labels.clone()) + .unwrap_or_default(), + host_repo_path: self.run.as_ref().and_then(|run| run.host_repo_path.clone()), + start_time: self.start.as_ref().map(|start| start.start_time), + status: self.status.as_ref().map(|status| status.status), + status_reason: self.status.as_ref().and_then(|status| status.reason), + duration_ms: self.conclusion.as_ref().map(|conclusion| conclusion.duration_ms), + total_cost: self + .conclusion + .as_ref() + .and_then(|conclusion| conclusion.total_cost), + } + } + + fn node_mut(&mut self, node_id: &str, visit: u32) -> &mut NodeState { + self.nodes + .entry((node_id.to_string(), visit)) + .or_default() + } + + fn current_visit_for(&self, node_id: &str) -> Option { + self.nodes + .keys() + .filter(|(current_node_id, _)| current_node_id == node_id) + .map(|(_, visit)| *visit) + .max() + } + + fn reset_for_rewind(&mut self) { + self.status = None; + self.checkpoint = None; + self.checkpoints.clear(); + self.conclusion = None; + self.retro = None; + self.retro_prompt = None; + self.retro_response = None; + self.sandbox = None; + self.final_patch = None; + self.pull_request = None; + self.nodes.clear(); + } +} + +fn merge_node_snapshot(node: &mut NodeState, snapshot: NodeSnapshot) { + node.prompt = node.prompt.clone().or(snapshot.prompt); + node.response = node.response.clone().or(snapshot.response); + node.status = node.status.clone().or(snapshot.status); + node.outcome = node.outcome.clone().or(snapshot.outcome); + node.provider_used = node.provider_used.clone().or(snapshot.provider_used); + node.diff = node.diff.clone().or(snapshot.diff); + node.script_invocation = node.script_invocation.clone().or(snapshot.script_invocation); + node.script_timing = node.script_timing.clone().or(snapshot.script_timing); + node.parallel_results = node.parallel_results.clone().or(snapshot.parallel_results); + node.stdout = node.stdout.clone().or(snapshot.stdout); + node.stderr = node.stderr.clone().or(snapshot.stderr); +} + +fn parse_ts(value: &Value) -> Result> { + let ts = value + .get("ts") + .and_then(Value::as_str) + .ok_or_else(|| StoreError::InvalidEvent("event payload missing ts".into()))?; + chrono::DateTime::parse_from_rfc3339(ts) + .map(|ts| ts.with_timezone(&Utc)) + .map_err(|err| StoreError::InvalidEvent(format!("invalid event ts: {err}")).into()) +} + +fn parse_run_id(value: &Value) -> Result { + let run_id = value + .get("run_id") + .and_then(Value::as_str) + .ok_or_else(|| StoreError::InvalidEvent("event payload missing run_id".into()))?; + run_id + .parse() + .map_err(|err| StoreError::InvalidEvent(format!("invalid run_id: {err}")).into()) +} + +fn required_string(properties: &serde_json::Map, key: &str) -> Result { + properties + .get(key) + .and_then(Value::as_str) + .map(ToString::to_string) + .ok_or_else(|| StoreError::InvalidEvent(format!("event missing string property {key}")).into()) +} + +fn optional_string(properties: &serde_json::Map, key: &str) -> Option { + properties + .get(key) + .and_then(Value::as_str) + .map(ToString::to_string) +} + +fn required_u64(properties: &serde_json::Map, key: &str) -> Result { + properties + .get(key) + .and_then(Value::as_u64) + .ok_or_else(|| StoreError::InvalidEvent(format!("event missing integer property {key}")).into()) +} + +fn required_u32(properties: &serde_json::Map, key: &str) -> Result { + u32::try_from(required_u64(properties, key)?) + .map_err(|_| StoreError::InvalidEvent(format!("property {key} does not fit in u32")).into()) +} + +fn required_json( + properties: &serde_json::Map, + key: &str, +) -> Result { + let value = properties + .get(key) + .cloned() + .ok_or_else(|| StoreError::InvalidEvent(format!("event missing property {key}")))?; + serde_json::from_value(value) + .map_err(|err| StoreError::InvalidEvent(format!("invalid property {key}: {err}")).into()) +} + +fn optional_json( + properties: &serde_json::Map, + key: &str, +) -> Result> { + properties + .get(key) + .filter(|value| !value.is_null()) + .cloned() + .map(|value| { + serde_json::from_value(value).map_err(|err| { + StoreError::InvalidEvent(format!("invalid property {key}: {err}")).into() + }) + }) + .transpose() +} + +fn parse_reason(properties: &serde_json::Map) -> Result> { + optional_string(properties, "reason") + .map(|reason| { + serde_json::from_value(Value::String(reason)).map_err(|err| { + StoreError::InvalidEvent(format!("invalid status reason: {err}")).into() + }) + }) + .transpose() +} + +fn run_status_record( + status: RunStatus, + properties: &serde_json::Map, + updated_at: DateTime, +) -> Result { + Ok(RunStatusRecord { + status, + reason: parse_reason(properties)?, + updated_at, + }) +} + +fn checkpoint_from_properties( + properties: &serde_json::Map, + timestamp: DateTime, +) -> Result { + let loop_failure_signatures = + optional_json::>(properties, "loop_failure_signatures")? + .unwrap_or_default() + .into_iter() + .map(|(key, value)| (FailureSignature(key), value)) + .collect(); + let restart_failure_signatures = + optional_json::>(properties, "restart_failure_signatures")? + .unwrap_or_default() + .into_iter() + .map(|(key, value)| (FailureSignature(key), value)) + .collect(); + + Ok(Checkpoint { + timestamp, + current_node: required_string(properties, "current_node")?, + completed_nodes: optional_json(properties, "completed_nodes")?.unwrap_or_default(), + node_retries: optional_json(properties, "node_retries")?.unwrap_or_default(), + context_values: optional_json(properties, "context_values")?.unwrap_or_default(), + node_outcomes: optional_json(properties, "node_outcomes")?.unwrap_or_default(), + next_node_id: optional_string(properties, "next_node_id"), + git_commit_sha: optional_string(properties, "git_commit_sha"), + loop_failure_signatures, + restart_failure_signatures, + node_visits: optional_json(properties, "node_visits")?.unwrap_or_default(), + }) +} + +fn conclusion_from_completed( + properties: &serde_json::Map, + timestamp: DateTime, +) -> Result { + let usage = optional_json::(properties, "usage")?; + Ok(Conclusion { + timestamp, + status: StageStatus::from_str(&required_string(properties, "status")?).map_err( + |err| StoreError::InvalidEvent(format!("invalid completed stage status: {err}")), + )?, + duration_ms: required_u64(properties, "duration_ms")?, + failure_reason: None, + final_git_commit_sha: optional_string(properties, "final_git_commit_sha"), + stages: Vec::new(), + total_cost: properties.get("total_cost").and_then(Value::as_f64), + total_retries: 0, + total_input_tokens: usage.as_ref().map_or(0, |usage| usage.input_tokens), + total_output_tokens: usage.as_ref().map_or(0, |usage| usage.output_tokens), + total_cache_read_tokens: usage + .as_ref() + .and_then(|usage| usage.cache_read_tokens) + .unwrap_or(0), + total_cache_write_tokens: usage + .as_ref() + .and_then(|usage| usage.cache_write_tokens) + .unwrap_or(0), + total_reasoning_tokens: usage + .as_ref() + .and_then(|usage| usage.reasoning_tokens) + .unwrap_or(0), + has_pricing: usage.as_ref().is_some_and(|usage| usage.cost.is_some()), + }) +} + +fn conclusion_from_failed( + properties: &serde_json::Map, + timestamp: DateTime, +) -> Conclusion { + Conclusion { + timestamp, + status: StageStatus::Fail, + duration_ms: properties + .get("duration_ms") + .and_then(Value::as_u64) + .unwrap_or_default(), + failure_reason: optional_string(properties, "error"), + final_git_commit_sha: optional_string(properties, "git_commit_sha"), + stages: Vec::new(), + total_cost: None, + total_retries: 0, + total_input_tokens: 0, + total_output_tokens: 0, + total_cache_read_tokens: 0, + total_cache_write_tokens: 0, + total_reasoning_tokens: 0, + has_pricing: false, + } +} + +fn stage_visit(node_id: &str, properties: &serde_json::Map, state: &RunState) -> Option { + properties + .get("node_visits") + .and_then(|value| serde_json::from_value::>(value.clone()).ok()) + .and_then(|visits| visits.get(node_id).copied()) + .and_then(|visit| u32::try_from(visit).ok()) + .or_else(|| state.current_visit_for(node_id)) +} + +fn stage_outcome_from_properties( + properties: &serde_json::Map, +) -> Result { + let status = StageStatus::from_str(&required_string(properties, "status")?).map_err(|err| { + StoreError::InvalidEvent(format!("invalid stage status: {err}")) + })?; + Ok(Outcome { + status, + preferred_label: optional_string(properties, "preferred_label"), + suggested_next_ids: optional_json(properties, "suggested_next_ids")?.unwrap_or_default(), + context_updates: optional_json(properties, "context_updates")?.unwrap_or_default(), + jump_to_node: optional_string(properties, "jump_to_node"), + notes: optional_string(properties, "notes"), + failure: optional_json(properties, "failure")?, + usage: optional_json(properties, "usage")?, + files_touched: optional_json(properties, "files_touched")?.unwrap_or_default(), + duration_ms: properties.get("duration_ms").and_then(Value::as_u64), + }) +} + +fn node_status_from_outcome(outcome: &NodeOutcomeRecord, timestamp: DateTime) -> NodeStatusRecord { + NodeStatusRecord { + status: outcome.status.clone(), + notes: outcome.notes.clone(), + failure_reason: outcome.failure.as_ref().map(|failure| failure.message.clone()), + timestamp, + } +} + +fn provider_used_from_prompt(properties: &serde_json::Map) -> Option { + let mut provider_used = serde_json::Map::new(); + if let Some(mode) = optional_string(properties, "mode") { + provider_used.insert("mode".to_string(), Value::String(mode)); + } + if let Some(provider) = optional_string(properties, "provider") { + provider_used.insert("provider".to_string(), Value::String(provider)); + } + if let Some(model) = optional_string(properties, "model") { + provider_used.insert("model".to_string(), Value::String(model)); + } + (!provider_used.is_empty()).then_some(Value::Object(provider_used)) +} + +fn provider_used_from_agent_event( + event_name: &str, + properties: &serde_json::Map, +) -> Option { + let mut provider_used = serde_json::Map::new(); + provider_used.insert( + "mode".to_string(), + Value::String(if event_name == "agent.cli.started" { + "cli".to_string() + } else { + "agent".to_string() + }), + ); + if let Some(provider) = optional_string(properties, "provider") { + provider_used.insert("provider".to_string(), Value::String(provider)); + } + if let Some(model) = optional_string(properties, "model") { + provider_used.insert("model".to_string(), Value::String(model)); + } + if let Some(command) = optional_string(properties, "command") { + provider_used.insert("command".to_string(), Value::String(command)); + } + Some(Value::Object(provider_used)) +} diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index 40b169240..4269f69ca 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -236,30 +236,28 @@ impl Store for SlateStore { Ok(Arc::new(run_store) as Arc) } - async fn open_run(&self, run_id: &RunId) -> Result>> { - let Some(locator) = - catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await? - else { - return Ok(None); - }; + async fn open_run(&self, run_id: &RunId) -> Result> { + let locator = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id) + .await? + .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; - let Some(run_store) = self.open_run_store(&locator).await? else { - return Ok(None); - }; - Ok(Some(Arc::new(run_store) as Arc)) + let run_store = self + .open_run_store(&locator) + .await? + .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; + Ok(Arc::new(run_store) as Arc) } - async fn open_run_reader(&self, run_id: &RunId) -> Result>> { - let Some(locator) = - catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await? - else { - return Ok(None); - }; + async fn open_run_reader(&self, run_id: &RunId) -> Result> { + let locator = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id) + .await? + .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; - let Some(run_store) = self.open_run_reader_store(&locator).await? else { - return Ok(None); - }; - Ok(Some(Arc::new(run_store) as Arc)) + let run_store = self + .open_run_reader_store(&locator) + .await? + .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; + Ok(Arc::new(run_store) as Arc) } async fn list_runs(&self, query: &ListRunsQuery) -> Result> { @@ -584,7 +582,6 @@ mod tests { let reopened = store .open_run(&test_run_id("run-1")) .await - .unwrap() .unwrap(); let stored = reopened.get_run().await.unwrap().unwrap(); assert_eq!(stored.run_id, test_run_id("run-1")); @@ -594,8 +591,7 @@ mod tests { store .open_run(&test_run_id("run-1")) .await - .unwrap() - .is_none() + .is_err() ); assert!(!object_exists(object_store.clone(), &by_id).await); assert!(!object_exists(object_store.clone(), &by_start).await); @@ -634,8 +630,7 @@ mod tests { store .open_run(&test_run_id("run-1")) .await - .unwrap() - .is_some() + .is_ok() ); assert!( store @@ -680,7 +675,6 @@ mod tests { let reopened = store .open_run(&test_run_id("run-1")) .await - .unwrap() .unwrap(); let next_event = reopened .append_event(&event_payload( @@ -730,8 +724,7 @@ mod tests { store .open_run(&test_run_id("run-1")) .await - .unwrap() - .is_none() + .is_err() ); assert!( store @@ -783,7 +776,6 @@ mod tests { let reopened = store .open_run(&test_run_id("run-1")) .await - .unwrap() .unwrap(); let first_event = run .append_event(&event_payload("run-1", "2026-03-27T12:00:00Z", "Started")) @@ -884,8 +876,7 @@ mod tests { store .open_run(&test_run_id("run-1")) .await - .unwrap() - .is_none() + .is_err() ); assert!(list_paths(object_store, "runs").await.is_empty()); } diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 5d704da64..3c1c0a9dc 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -15,9 +15,10 @@ use tokio::time; use tokio_stream::wrappers::UnboundedReceiverStream; use crate::keys; +use crate::run_state::EventProjectionCache; use crate::{ CatalogRecord, EventEnvelope, EventPayload, NodeOutcomeRecord, NodeSnapshot, NodeVisitRef, - Result, RunSnapshot, RunStore, RunSummary, StoreError, + Result, RunSnapshot, RunState, RunStore, RunSummary, StoreError, }; use fabro_types::{ Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunId, RunRecord, @@ -38,6 +39,7 @@ pub(crate) struct SlateRunStoreInner { event_seq: AtomicU32, checkpoint_seq: AtomicU32, close_lock: Mutex<()>, + projection_cache: Mutex, } enum SlateRunDb { @@ -60,6 +62,7 @@ impl SlateRunStore { event_seq: AtomicU32::new(event_seq), checkpoint_seq: AtomicU32::new(checkpoint_seq), close_lock: Mutex::new(()), + projection_cache: Mutex::new(EventProjectionCache::default()), }), }) } @@ -78,6 +81,7 @@ impl SlateRunStore { event_seq: AtomicU32::new(event_seq), checkpoint_seq: AtomicU32::new(checkpoint_seq), close_lock: Mutex::new(()), + projection_cache: Mutex::new(EventProjectionCache::default()), }), }) } @@ -143,42 +147,74 @@ impl SlateRunStore { where R: DbRead + Sync, { - let run = get_json::<_, RunRecord>(db, keys::run()).await?; - let start = get_json::<_, StartRecord>(db, keys::start()).await?; - let status = get_json::<_, RunStatusRecord>(db, keys::status()).await?; - let conclusion = get_json::<_, Conclusion>(db, keys::conclusion()).await?; - - let workflow_name = run.as_ref().map(|run| { - if run.graph.name.is_empty() { - "unnamed".to_string() - } else { - run.graph.name.clone() + let events = list_events_from(db, 1).await?; + let mut state = RunState::apply_events(&events)?; + let mut nodes = BTreeSet::new(); + let mut iter = db.scan_prefix(b"nodes/").await?; + while let Some(entry) = iter.next().await? { + let key = key_to_string(&entry.key)?; + if let Some((node_id, visit, _)) = keys::parse_node_key(&key) { + nodes.insert((node_id, visit)); } - }); - let goal = run.as_ref().and_then(|run| { - let goal = run.graph.goal(); - (!goal.is_empty()).then(|| goal.to_string()) - }); - - Ok(RunSummary { - run_id: catalog.run_id, - created_at: catalog.created_at, - db_prefix: catalog.db_prefix.clone(), - run_dir: catalog.run_dir.clone(), - workflow_name, - workflow_slug: run.as_ref().and_then(|run| run.workflow_slug.clone()), - goal, - labels: run - .as_ref() - .map(|run| run.labels.clone()) - .unwrap_or_default(), - host_repo_path: run.as_ref().and_then(|run| run.host_repo_path.clone()), - start_time: start.map(|start| start.start_time), - status: status.as_ref().map(|status| status.status), - status_reason: status.and_then(|status| status.reason), - duration_ms: conclusion.as_ref().map(|conclusion| conclusion.duration_ms), - total_cost: conclusion.and_then(|conclusion| conclusion.total_cost), - }) + } + let snapshot = if let Some(run) = get_json::<_, RunRecord>(db, keys::run()).await? { + let mut snapshot_nodes = Vec::new(); + for (node_id, visit) in nodes { + let node = NodeVisitRef { + node_id: &node_id, + visit, + }; + let prompt_key = keys::node_prompt(&node); + let response_key = keys::node_response(&node); + let status_key = keys::node_status(&node); + let outcome_key = keys::node_outcome(&node); + let provider_key = keys::node_provider_used(&node); + let diff_key = keys::node_diff(&node); + let invocation_key = keys::node_script_invocation(&node); + let timing_key = keys::node_script_timing(&node); + let results_key = keys::node_parallel_results(&node); + let stdout_key = keys::node_stdout(&node); + let stderr_key = keys::node_stderr(&node); + snapshot_nodes.push(NodeSnapshot { + node_id: node_id.clone(), + visit, + prompt: get_text(db, &prompt_key).await?, + response: get_text(db, &response_key).await?, + status: get_json(db, &status_key).await?, + outcome: get_json(db, &outcome_key).await?, + provider_used: get_json(db, &provider_key).await?, + diff: get_text(db, &diff_key).await?, + script_invocation: get_json(db, &invocation_key).await?, + script_timing: get_json(db, &timing_key).await?, + parallel_results: get_json(db, &results_key).await?, + stdout: get_text(db, &stdout_key).await?, + stderr: get_text(db, &stderr_key).await?, + }); + } + Some(RunSnapshot { + run, + start: get_json::<_, StartRecord>(db, keys::start()).await?, + status: get_json::<_, RunStatusRecord>(db, keys::status()).await?, + checkpoint: get_json::<_, Checkpoint>(db, keys::checkpoint()).await?, + conclusion: get_json::<_, Conclusion>(db, keys::conclusion()).await?, + retro: get_json::<_, Retro>(db, keys::retro()).await?, + graph: get_text(db, keys::graph()).await?, + sandbox: get_json::<_, SandboxRecord>(db, keys::sandbox()).await?, + final_patch: get_text(db, keys::final_patch()).await?, + pull_request: get_json::<_, PullRequestRecord>(db, keys::pull_request()).await?, + nodes: snapshot_nodes, + }) + } else { + None + }; + state.merge_legacy( + snapshot, + get_text(db, keys::graph()).await?, + get_text(db, keys::retro_prompt()).await?, + get_text(db, keys::retro_response()).await?, + list_checkpoints(db).await?, + ); + Ok(state.build_summary(catalog)) } fn validate_run_record(&self, record: &RunRecord) -> Result<()> { @@ -230,6 +266,20 @@ impl SlateRunStore { stderr: self.inner.db.get_text(&keys::node_stderr(node)).await?, }) } + + async fn projected_state(&self) -> Result { + let next_seq = { + let cache = self.inner.projection_cache.lock().await; + cache.last_seq.saturating_add(1) + }; + let events = self.inner.db.list_events_from(next_seq).await?; + let mut cache = self.inner.projection_cache.lock().await; + for event in &events { + cache.state.apply_event(event)?; + cache.last_seq = event.seq; + } + Ok(cache.state.clone()) + } } #[async_trait] @@ -619,7 +669,25 @@ impl RunStore for SlateRunStore { self.inner.db.list_all_assets().await } + async fn state(&self) -> Result { + let mut state = self.projected_state().await?; + state.merge_legacy( + self.get_snapshot_legacy().await?, + self.get_graph().await?, + self.get_retro_prompt().await?, + self.get_retro_response().await?, + self.list_checkpoints().await?, + ); + Ok(state) + } + async fn get_snapshot(&self) -> Result> { + self.state().await.map(|state| state.to_snapshot()) + } +} + +impl SlateRunStore { + async fn get_snapshot_legacy(&self) -> Result> { let Some(run) = self.get_run().await? else { return Ok(None); }; diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index f36fa973a..dec103a97 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -1510,7 +1510,9 @@ pub(crate) fn normalize_json_value(value: Value) -> Value { } pub fn event_payload_from_redacted_json(line: &str, run_id: &RunId) -> Result { - let value = serde_json::from_str(line).context("Failed to parse redacted event payload")?; + let value = normalize_json_value( + serde_json::from_str(line).context("Failed to parse redacted event payload")?, + ); EventPayload::new(value, run_id).map_err(anyhow::Error::from) } diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 43242ded9..e87cb21d2 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -4,7 +4,6 @@ use std::sync::Arc; use async_trait::async_trait; use fabro_agent::Sandbox; -use fabro_store::NodeVisitRef; use fabro_types::RunId; use crate::context::keys; @@ -257,18 +256,7 @@ impl Handler for AgentHandler { let visit = visit_from_context(context); let stage_dir = node_dir(run_dir, &node.id, visit); fs::create_dir_all(&stage_dir).await?; - let node_ref = NodeVisitRef { - node_id: &node.id, - visit: u32::try_from(visit).unwrap_or(u32::MAX), - }; - if let Some(ref store) = services.run_store { - store - .put_node_prompt(&node_ref, &prompt) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; - } else { - fs::write(stage_dir.join("prompt.md"), &prompt).await?; - } + fs::write(stage_dir.join("prompt.md"), &prompt).await?; // 3. Call LLM backend (agent loop) let thread_id = context.thread_id(); @@ -326,14 +314,7 @@ impl Handler for AgentHandler { }; // 4. Write response to logs - if let Some(ref store) = services.run_store { - store - .put_node_response(&node_ref, &response_text) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; - } else { - fs::write(stage_dir.join("response.md"), &response_text).await?; - } + fs::write(stage_dir.join("response.md"), &response_text).await?; // 7. Build and write status let mut outcome = Outcome::success(); outcome.notes = Some(format!("Stage completed: {}", node.id)); @@ -392,7 +373,7 @@ mod tests { use super::*; use crate::event::EventEmitter; use fabro_graphviz::graph::AttrValue; - use fabro_store::{InMemoryStore, RunStore, Store}; + use fabro_store::{InMemoryStore, NodeVisitRef, RunStore, Store}; use fabro_types::fixtures; use std::sync::Arc; use tempfile::TempDir; @@ -412,7 +393,7 @@ mod tests { .await .unwrap(); let services = EngineServices { - run_store: Some(Arc::clone(&run_store)), + run_store: Arc::clone(&run_store), ..EngineServices::test_default() }; let logger = crate::event::StoreProgressLogger::new(Arc::clone(&run_store)); diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs index 0cae5fe6c..d548afa4a 100644 --- a/lib/crates/fabro-workflow/src/handler/command.rs +++ b/lib/crates/fabro-workflow/src/handler/command.rs @@ -1,8 +1,6 @@ use std::path::Path; use async_trait::async_trait; -use fabro_store::NodeVisitRef; - use crate::context::Context; use crate::context::keys; use crate::error::FabroError; @@ -92,22 +90,11 @@ impl Handler for CommandHandler { let visit = visit_from_context(context); let stage_dir = node_dir(run_dir, &node.id, visit); fs::create_dir_all(&stage_dir).await?; - let node_ref = NodeVisitRef { - node_id: &node.id, - visit: u32::try_from(visit).unwrap_or(u32::MAX), - }; - let invocation = serde_json::json!({ "command": script, "language": language, "timeout_ms": timeout_ms(node), }); - if let Some(ref store) = services.run_store { - store - .put_node_script_invocation(&node_ref, &invocation) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; - } fs::write( stage_dir.join("script_invocation.json"), serde_json::to_string_pretty(&invocation).unwrap(), @@ -142,31 +129,14 @@ impl Handler for CommandHandler { .await .map_err(|e| FabroError::handler(format!("Failed to spawn script: {e}")))?; - if let Some(ref store) = services.run_store { - store - .put_node_stdout(&node_ref, &result.stdout) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; - store - .put_node_stderr(&node_ref, &result.stderr) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; - } else { - fs::write(stage_dir.join("stdout.log"), &result.stdout).await?; - fs::write(stage_dir.join("stderr.log"), &result.stderr).await?; - } + fs::write(stage_dir.join("stdout.log"), &result.stdout).await?; + fs::write(stage_dir.join("stderr.log"), &result.stderr).await?; let timing = serde_json::json!({ "duration_ms": result.duration_ms, "exit_code": if result.timed_out { serde_json::Value::Null } else { serde_json::json!(result.exit_code) }, "timed_out": result.timed_out, }); - if let Some(ref store) = services.run_store { - store - .put_node_script_timing(&node_ref, &timing) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; - } fs::write( stage_dir.join("script_timing.json"), serde_json::to_string_pretty(&timing).unwrap(), @@ -230,6 +200,7 @@ mod tests { use crate::outcome::StageStatus; use fabro_graphviz::graph::AttrValue; use fabro_store::{InMemoryStore, NodeVisitRef, RunStore, Store}; + use fabro_types::fixtures; use std::sync::Arc; use std::time::Duration; @@ -237,6 +208,26 @@ mod tests { EngineServices::test_default() } + async fn make_services_with_run_store() -> ( + EngineServices, + Arc, + crate::event::StoreProgressLogger, + ) { + let store = InMemoryStore::default(); + let run_store = store + .create_run(&fixtures::RUN_1, chrono::Utc::now(), None) + .await + .unwrap(); + let services = EngineServices { + emitter: Arc::new(crate::event::EventEmitter::new(fixtures::RUN_1)), + run_store: Arc::clone(&run_store), + ..EngineServices::test_default() + }; + let logger = crate::event::StoreProgressLogger::new(Arc::clone(&run_store)); + logger.register(services.emitter.as_ref()); + (services, run_store, logger) + } + #[tokio::test] async fn script_handler_no_script() { let handler = CommandHandler; @@ -586,31 +577,28 @@ mod tests { let context = Context::new(); let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); - let store = Arc::new(InMemoryStore::default()); - let run_store = store - .create_run(&fabro_types::fixtures::RUN_1, chrono::Utc::now(), None) - .await - .unwrap(); - let services = EngineServices { - run_store: Some(Arc::clone(&run_store) as Arc), - ..EngineServices::test_default() - }; + let (services, run_store, logger) = make_services_with_run_store().await; handler .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); + logger.flush().await; let snapshot = run_store - .get_node(&NodeVisitRef { + .state() + .await + .unwrap(); + let node = snapshot + .node(&NodeVisitRef { node_id: "script_node", visit: 1, }) - .await + .cloned() .unwrap(); - assert_eq!(snapshot.script_invocation.unwrap()["command"], "echo hello"); - assert_eq!(snapshot.script_timing.unwrap()["exit_code"], 0); + assert_eq!(node.script_invocation.unwrap()["script"], "echo hello"); + assert_eq!(node.script_timing.unwrap()["exit_code"], 0); } #[tokio::test] diff --git a/lib/crates/fabro-workflow/src/handler/fan_in.rs b/lib/crates/fabro-workflow/src/handler/fan_in.rs index 8423f6308..948cb3069 100644 --- a/lib/crates/fabro-workflow/src/handler/fan_in.rs +++ b/lib/crates/fabro-workflow/src/handler/fan_in.rs @@ -226,7 +226,7 @@ async fn llm_evaluate( node_id: &str, emitter: &Arc, sandbox: &Arc, - run_store: Option>, + run_store: Arc, ) -> Result { let results_text = serde_json::to_string_pretty(results).unwrap_or_else(|_| results.to_string()); @@ -244,14 +244,11 @@ async fn llm_evaluate( node_id, visit: u32::try_from(visit).unwrap_or(u32::MAX), }; - if let Some(ref store) = run_store { - store - .put_node_prompt(&node_ref, &full_prompt) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; - } else { - fs::write(stage_dir.join("prompt.md"), &full_prompt).await?; - } + run_store + .put_node_prompt(&node_ref, &full_prompt) + .await + .map_err(|err| FabroError::handler(err.to_string()))?; + fs::write(stage_dir.join("prompt.md"), &full_prompt).await?; // Build a synthetic node for the backend call let eval_node = Node::new("fan_in_eval"); @@ -281,14 +278,11 @@ async fn llm_evaluate( .unwrap_or_else(|| "unknown".to_string()); let response_text = serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string()); - if let Some(ref store) = run_store { - store - .put_node_response(&node_ref, &response_text) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; - } else { - fs::write(stage_dir.join("response.md"), &response_text).await?; - } + run_store + .put_node_response(&node_ref, &response_text) + .await + .map_err(|err| FabroError::handler(err.to_string()))?; + fs::write(stage_dir.join("response.md"), &response_text).await?; Ok(Candidate { id: best_id, status: outcome.status.to_string(), @@ -297,14 +291,11 @@ async fn llm_evaluate( } Ok(CodergenResult::Text { text, .. }) => { // Write response to logs - if let Some(ref store) = run_store { - store - .put_node_response(&node_ref, &text) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; - } else { - fs::write(stage_dir.join("response.md"), &text).await?; - } + run_store + .put_node_response(&node_ref, &text) + .await + .map_err(|err| FabroError::handler(err.to_string()))?; + fs::write(stage_dir.join("response.md"), &text).await?; // The LLM responded with text; try to find a matching candidate ID let text = text.trim().to_string(); diff --git a/lib/crates/fabro-workflow/src/handler/mod.rs b/lib/crates/fabro-workflow/src/handler/mod.rs index 93f62de2d..291b9c93c 100644 --- a/lib/crates/fabro-workflow/src/handler/mod.rs +++ b/lib/crates/fabro-workflow/src/handler/mod.rs @@ -19,6 +19,8 @@ use std::sync::Arc; use async_trait::async_trait; use fabro_agent::Sandbox; use fabro_store::RunStore; +#[cfg(test)] +use fabro_store::Store; use crate::context::Context; use crate::error::FabroError; @@ -34,7 +36,7 @@ pub struct EngineServices { pub registry: Arc, pub emitter: Arc, pub sandbox: Arc, - pub run_store: Option>, + pub run_store: Arc, /// Git state for the current run. Set via `set_git_state` at the start of /// `run_via_core` and read by parallel/fan-in handlers. pub(crate) git_state: std::sync::RwLock>>, @@ -75,7 +77,12 @@ impl EngineServices { sandbox: Arc::new(fabro_agent::LocalSandbox::new( std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")), )), - run_store: None, + run_store: futures::executor::block_on(async { + fabro_store::InMemoryStore::default() + .create_run(&fabro_types::RunId::new(), chrono::Utc::now(), None) + .await + .expect("in-memory test run store should initialize") + }), git_state: std::sync::RwLock::new(None), hook_runner: None, env: HashMap::new(), diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 0e305df36..f44388402 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -481,16 +481,15 @@ impl Handler for ParallelHandler { let visit = visit_from_context(context); let node_dir = node_dir(run_dir, &node.id, visit); let _ = fs::create_dir_all(&node_dir).await; - if let Some(ref store) = services.run_store { - let node_ref = NodeVisitRef { - node_id: &node.id, - visit: u32::try_from(visit).unwrap_or(u32::MAX), - }; - store - .put_node_parallel_results(&node_ref, &serde_json::json!(results_json)) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; - } + let node_ref = NodeVisitRef { + node_id: &node.id, + visit: u32::try_from(visit).unwrap_or(u32::MAX), + }; + services + .run_store + .put_node_parallel_results(&node_ref, &serde_json::json!(results_json)) + .await + .map_err(|err| FabroError::handler(err.to_string()))?; if let Ok(json) = serde_json::to_string_pretty(&results_json) { let _ = fs::write(node_dir.join("parallel_results.json"), json).await; } @@ -687,7 +686,7 @@ mod tests { .await .unwrap(); let services = EngineServices { - run_store: Some(Arc::clone(&run_store) as Arc), + run_store: Arc::clone(&run_store) as Arc, ..EngineServices::test_default() }; let mut node = Node::new("par"); diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index fd7e7b797..5c79bfe6f 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -3,8 +3,6 @@ use std::path::Path; use async_trait::async_trait; use fabro_model::Provider; -use fabro_store::NodeVisitRef; - use crate::context::keys; use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; @@ -93,18 +91,7 @@ impl Handler for PromptHandler { let visit = visit_from_context(context); let stage_dir = node_dir(run_dir, &node.id, visit); fs::create_dir_all(&stage_dir).await?; - let node_ref = NodeVisitRef { - node_id: &node.id, - visit: u32::try_from(visit).unwrap_or(u32::MAX), - }; - if let Some(ref store) = services.run_store { - store - .put_node_prompt(&node_ref, &prompt) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; - } else { - fs::write(stage_dir.join("prompt.md"), &prompt).await?; - } + fs::write(stage_dir.join("prompt.md"), &prompt).await?; let prompt_provider = node .provider() @@ -169,14 +156,7 @@ impl Handler for PromptHandler { }); // 4. Write response to logs - if let Some(ref store) = services.run_store { - store - .put_node_response(&node_ref, &response_text) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; - } else { - fs::write(stage_dir.join("response.md"), &response_text).await?; - } + fs::write(stage_dir.join("response.md"), &response_text).await?; // 5. Build and write status let mut outcome = Outcome::success(); @@ -225,7 +205,7 @@ mod tests { .await .unwrap(); let services = EngineServices { - run_store: Some(Arc::clone(&run_store)), + run_store: Arc::clone(&run_store), ..EngineServices::test_default() }; let logger = crate::event::StoreProgressLogger::new(Arc::clone(&run_store)); diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index d3703c563..1ed6527f2 100644 --- a/lib/crates/fabro-workflow/src/operations/create.rs +++ b/lib/crates/fabro-workflow/src/operations/create.rs @@ -148,8 +148,8 @@ async fn persist_created_run( Err(err) => store .open_run(&record.run_id) .await - .map_err(|open_err| FabroError::engine(open_err.to_string()))? - .ok_or_else(|| FabroError::engine(err.to_string()))?, + .map_err(|open_err| FabroError::engine(open_err.to_string())) + .or_else(|_| Err(FabroError::engine(err.to_string())))?, }; run_store.put_run(record).await.map_err(store_error)?; @@ -717,7 +717,7 @@ mod tests { created.persisted.run_record().workflow_slug.as_deref(), Some("slug") ); - let run_store = store.open_run(&fixtures::RUN_1).await.unwrap().unwrap(); + let run_store = store.open_run(&fixtures::RUN_1).await.unwrap(); assert_eq!( run_store.get_status().await.unwrap().unwrap().status, crate::run_status::RunStatus::Submitted @@ -839,11 +839,7 @@ mod tests { ) .await .unwrap(); - let run_store = store - .open_run_reader(&created.run_id) - .await - .unwrap() - .unwrap(); + let run_store = store.open_run_reader(&created.run_id).await.unwrap(); let events = run_store.list_events().await.unwrap(); assert_eq!( diff --git a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs index c7fa52049..9f599285d 100644 --- a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs +++ b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs @@ -26,9 +26,13 @@ pub async fn rebuild_metadata_branch( bail!("metadata branch already exists for run {run_id}"); } - let run_record = run_store - .get_run() + let state = run_store + .state() .await? + ; + let run_record = state + .run + .clone() .ok_or_else(|| anyhow::anyhow!("run record not found for {run_id}"))?; let sig = Signature::now("Fabro", "noreply@fabro.sh")?; @@ -43,10 +47,10 @@ pub async fn rebuild_metadata_branch( "run.json".to_string(), serde_json::to_vec_pretty(&run_record)?, )); - if let Some(start) = run_store.get_start().await? { + if let Some(start) = state.start.clone() { init_entries.push(("start.json".to_string(), serde_json::to_vec_pretty(&start)?)); } - if let Some(sandbox) = run_store.get_sandbox().await? { + if let Some(sandbox) = state.sandbox.clone() { init_entries.push(( "sandbox.json".to_string(), serde_json::to_vec_pretty(&sandbox)?, @@ -54,7 +58,7 @@ pub async fn rebuild_metadata_branch( } write_entries(&bs, &init_entries, "init run")?; - let mut checkpoints = run_store.list_checkpoints().await?; + let mut checkpoints = state.checkpoints.clone(); backfill_missing_checkpoint_shas(git_store, run_id, &mut checkpoints); for (_seq, checkpoint) in checkpoints { @@ -69,7 +73,9 @@ pub async fn rebuild_metadata_branch( for visit in 1..=max_visit { let visit = u32::try_from(visit) .with_context(|| format!("visit {visit} for node {node_id} exceeds u32"))?; - let node = run_store.get_node(&NodeVisitRef { node_id, visit }).await?; + let Some(node) = state.node(&NodeVisitRef { node_id, visit }).cloned() else { + continue; + }; if let Some(prompt) = node.prompt { entries.push(( @@ -125,7 +131,7 @@ pub async fn rebuild_metadata_branch( write_entries(&bs, &entries, "checkpoint")?; } - if let Some(retro) = run_store.get_retro().await? { + if let Some(retro) = state.retro.clone() { let entries = vec![("retro.json".to_string(), serde_json::to_vec_pretty(&retro)?)]; write_entries(&bs, &entries, "finalize run")?; } diff --git a/lib/crates/fabro-workflow/src/operations/resume.rs b/lib/crates/fabro-workflow/src/operations/resume.rs index e03ca7fb7..fcac5048f 100644 --- a/lib/crates/fabro-workflow/src/operations/resume.rs +++ b/lib/crates/fabro-workflow/src/operations/resume.rs @@ -11,24 +11,20 @@ use super::start::{StartServices, Started, execute_persisted_run}; /// Resume a workflow run from its checkpoint. Errors if no checkpoint is found. pub async fn resume(run_dir: &Path, services: StartServices) -> Result { - if let Some(record) = services + let state = services .run_store - .get_status() + .state() .await - .map_err(|err| FabroError::engine(err.to_string()))? - { + .map_err(|err| FabroError::engine(err.to_string()))?; + + if let Some(record) = state.status { if record.status == RunStatus::Succeeded { return Err(FabroError::Precondition( "run already finished successfully — nothing to resume".to_string(), )); } } - if let Some(conclusion) = services - .run_store - .get_conclusion() - .await - .map_err(|err| FabroError::engine(err.to_string()))? - { + if let Some(conclusion) = state.conclusion { if matches!( conclusion.status, StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped @@ -39,11 +35,8 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result Result { - if services + let state = services .run_store - .get_checkpoint() + .state() .await - .map_err(|err| FabroError::engine(err.to_string()))? - .is_some() - { + .map_err(|err| FabroError::engine(err.to_string()))?; + if state.checkpoint.is_some() { return Err(FabroError::Precondition( "checkpoint.json exists in run directory — did you mean to resume?".to_string(), )); } - if let Some(record) = services - .run_store - .get_status() - .await - .map_err(|err| FabroError::engine(err.to_string()))? - { + if let Some(record) = state.status { if !matches!(record.status, RunStatus::Submitted | RunStatus::Starting) { return Err(FabroError::Precondition(format!( "cannot start run: status is {:?}, expected submitted", @@ -580,7 +574,7 @@ impl RunSession { }; let pr_opts = PullRequestOptions { run_dir: retroed.run_options.run_dir.clone(), - run_store: Some(Arc::clone(&retroed.run_store)), + run_store: Arc::clone(&retroed.run_store), pr_config: self.pr_config, github_app: self.pr_github_app, origin_url: self.pr_origin_url, @@ -976,7 +970,7 @@ mod tests { cancel_token: None, emitter, interviewer: Arc::new(fabro_interview::AutoApproveInterviewer), - run_store: store.open_run(&fixtures::RUN_1).await.unwrap().unwrap(), + run_store: store.open_run(&fixtures::RUN_1).await.unwrap(), github_app: None, on_node: None, registry_override: Some(registry), @@ -1052,7 +1046,7 @@ mod tests { .unwrap(); assert_eq!(started.finalized.conclusion.status, StageStatus::Success); - let run_store = store.open_run(&fixtures::RUN_1).await.unwrap().unwrap(); + let run_store = store.open_run(&fixtures::RUN_1).await.unwrap(); assert!(run_store.get_conclusion().await.unwrap().is_some()); } diff --git a/lib/crates/fabro-workflow/src/pipeline/execute.rs b/lib/crates/fabro-workflow/src/pipeline/execute.rs index 98f95de77..c1349ab56 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute.rs @@ -73,7 +73,7 @@ pub async fn execute(init: Initialized) -> Executed { registry, emitter: Arc::clone(&emitter), sandbox: Arc::clone(&sandbox), - run_store: Some(Arc::clone(&run_store)), + run_store: Arc::clone(&run_store), git_state: std::sync::RwLock::new(git_state), hook_runner: hook_runner.clone(), env, diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index b39282918..2a9d5d080 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -69,7 +69,11 @@ pub(crate) async fn build_conclusion_from_store( run_duration_ms: u64, final_git_commit_sha: Option, ) -> Conclusion { - let checkpoint = run_store.get_checkpoint().await.ok().flatten(); + let checkpoint = run_store + .state() + .await + .ok() + .and_then(|state| state.checkpoint); let stage_durations = run_store .list_events() .await @@ -192,10 +196,12 @@ pub async fn write_finalize_commit( let git_author = run_options.git_author(); let store = MetadataStore::new(repo_path, &git_author); let mut entries = scan_node_files_from_store(run_store).await; - let retro_bytes = match run_store.get_retro().await { - Ok(Some(retro)) => serde_json::to_vec_pretty(&retro).ok(), - _ => None, - }; + let retro_bytes = run_store + .state() + .await + .ok() + .and_then(|state| state.retro) + .and_then(|retro| serde_json::to_vec_pretty(&retro).ok()); if let Some(bytes) = retro_bytes { entries.push(("retro.json".to_string(), bytes)); } diff --git a/lib/crates/fabro-workflow/src/pipeline/persist.rs b/lib/crates/fabro-workflow/src/pipeline/persist.rs index 4e86955f6..2d4ff94c0 100644 --- a/lib/crates/fabro-workflow/src/pipeline/persist.rs +++ b/lib/crates/fabro-workflow/src/pipeline/persist.rs @@ -29,17 +29,15 @@ pub(crate) async fn load_from_store( run_store: &dyn RunStore, run_dir: &Path, ) -> Result { - let run_record = run_store - .get_run() + let state = run_store + .state() .await - .map_err(|err| FabroError::engine(err.to_string()))? + .map_err(|err| FabroError::engine(err.to_string()))?; + let run_record = state + .run .ok_or_else(|| FabroError::Precondition("run record missing from store".to_string()))?; let graph = run_record.graph.clone(); - let source = run_store - .get_graph() - .await - .map_err(|err| FabroError::engine(err.to_string()))? - .unwrap_or_default(); + let source = state.graph_source.unwrap_or_default(); Ok(Persisted::new( graph, diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index f9c658036..130081daf 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -292,21 +292,17 @@ fn emit_run_notice( }); } -async fn load_pull_request_diff(run_store: Option<&dyn RunStore>, run_dir: &Path) -> String { - if let Some(run_store) = run_store { - run_store - .get_final_patch() - .await - .inspect_err(|err| { - tracing::warn!(error = %err, "Failed to load final patch from store for PR"); - }) - .ok() - .flatten() - .unwrap_or_default() - } else { - let _ = run_dir; - String::new() - } +async fn load_pull_request_diff(run_store: &dyn RunStore, run_dir: &Path) -> String { + let _ = run_dir; + run_store + .state() + .await + .inspect_err(|err| { + tracing::warn!(error = %err, "Failed to load final patch from store for PR"); + }) + .ok() + .and_then(|state| state.final_patch) + .unwrap_or_default() } /// Build a complete PR body by combining LLM-generated narrative with @@ -315,7 +311,7 @@ pub async fn build_pr_body( diff: &str, goal: &str, model: &str, - run_store: Option<&dyn RunStore>, + run_store: &dyn RunStore, run_dir: &Path, conclusion: Option<&Conclusion>, ) -> Result { @@ -323,54 +319,28 @@ pub async fn build_pr_body( let plan_text = read_plan_text(run_dir); let loaded_conclusion = if conclusion.is_none() { - match run_store { - Some(run_store) => run_store - .get_conclusion() - .await - .inspect_err(|err| { - tracing::warn!(error = %err, "Failed to load conclusion from store for PR body"); - }) - .ok() - .flatten(), - None => None, - } + run_store + .state() + .await + .inspect_err(|err| { + tracing::warn!(error = %err, "Failed to load conclusion from store for PR body"); + }) + .ok() + .and_then(|state| state.conclusion) } else { None }; let conclusion = conclusion.or(loaded_conclusion.as_ref()); - let retro = match run_store { - Some(run_store) => run_store - .get_retro() - .await - .inspect_err(|err| { - tracing::warn!(error = %err, "Failed to load retro from store for PR body"); - }) - .ok() - .flatten(), - None => None, - }; - let run_record = match run_store { - Some(run_store) => run_store - .get_run() - .await - .inspect_err(|err| { - tracing::warn!(error = %err, "Failed to load run record from store for PR body"); - }) - .ok() - .flatten(), - None => None, - }; - let dot_source = match run_store { - Some(run_store) => run_store - .get_graph() - .await - .inspect_err(|err| { - tracing::warn!(error = %err, "Failed to load graph from store for PR body"); - }) - .ok() - .flatten(), - None => None, - }; + let run_state = run_store + .state() + .await + .inspect_err(|err| { + tracing::warn!(error = %err, "Failed to load run state from store for PR body"); + }) + .ok(); + let retro = run_state.as_ref().and_then(|state| state.retro.clone()); + let run_record = run_state.as_ref().and_then(|state| state.run.clone()); + let dot_source = run_state.as_ref().and_then(|state| state.graph_source.clone()); // Build LLM prompt let system = if plan_text.is_some() { @@ -448,7 +418,7 @@ pub async fn maybe_open_pull_request( model: &str, draft: bool, auto_merge: Option, - run_store: Option<&dyn RunStore>, + run_store: &dyn RunStore, run_dir: &Path, conclusion: Option<&Conclusion>, ) -> Result, String> { @@ -519,12 +489,10 @@ pub async fn maybe_open_pull_request( title, }; - if let Some(run_store) = run_store { - run_store - .put_pull_request(&record) - .await - .map_err(|err| format!("failed to persist pull request in run store: {err}"))?; - } + run_store + .put_pull_request(&record) + .await + .map_err(|err| format!("failed to persist pull request in run store: {err}"))?; Ok(Some(record)) } @@ -554,8 +522,7 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) -> result.status, StageStatus::Success | StageStatus::PartialSuccess ) { - let diff = - load_pull_request_diff(options.run_store.as_deref(), &options.run_dir).await; + let diff = load_pull_request_diff(options.run_store.as_ref(), &options.run_dir).await; if let (Some(base_branch), Some(run_branch), Some(creds), Some(origin)) = ( &run_options.base_branch, pushed_branch.as_deref(), @@ -580,7 +547,7 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) -> &options.model, pr_cfg.draft, auto_merge, - options.run_store.as_deref(), + options.run_store.as_ref(), &options.run_dir, Some(&conclusion), ) @@ -1067,12 +1034,21 @@ mod tests { install_mock_llm(); let tmp = tempfile::tempdir().unwrap(); + let store = InMemoryStore::default(); + let run_store = store + .create_run( + &fixtures::RUN_1, + Utc::now(), + Some(&tmp.path().display().to_string()), + ) + .await + .unwrap(); let conclusion = make_test_conclusion(); let body = build_pr_body( "diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n", "Implement feature", "mock-model", - None, + run_store.as_ref(), tmp.path(), Some(&conclusion), ) @@ -1126,7 +1102,7 @@ mod tests { "diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n", "Implement feature", "mock-model", - Some(run_store.as_ref()), + run_store.as_ref(), tmp.path(), Some(&conclusion), ) @@ -1316,6 +1292,15 @@ mod tests { #[tokio::test] async fn empty_diff_returns_none() { let tmp = tempfile::tempdir().unwrap(); + let store = InMemoryStore::default(); + let run_store = store + .create_run( + &fixtures::RUN_1, + Utc::now(), + Some(&tmp.path().display().to_string()), + ) + .await + .unwrap(); let creds = GitHubAppCredentials { app_id: "123".to_string(), private_key_pem: "unused".to_string(), @@ -1330,7 +1315,7 @@ mod tests { "claude-sonnet-4-20250514", false, None, - None, + run_store.as_ref(), tmp.path(), None, ) @@ -1343,20 +1328,35 @@ mod tests { async fn load_pull_request_diff_uses_store_without_disk_patch() { let tmp = tempfile::tempdir().unwrap(); let store = InMemoryStore::default(); + let created_at = Utc::now(); let run_store = store .create_run( &fixtures::RUN_1, - Utc::now(), + created_at, Some(&tmp.path().display().to_string()), ) .await .unwrap(); + run_store + .put_run(&RunRecord { + run_id: fixtures::RUN_1, + created_at, + settings: Settings::default(), + graph: Graph::new("test"), + workflow_slug: None, + working_directory: tmp.path().to_path_buf(), + host_repo_path: None, + base_branch: None, + labels: std::collections::HashMap::new(), + }) + .await + .unwrap(); run_store .put_final_patch("diff --git a/src/lib.rs b/src/lib.rs\n+fn from_store() {}\n") .await .unwrap(); - let diff = load_pull_request_diff(Some(run_store.as_ref()), tmp.path()).await; + let diff = load_pull_request_diff(run_store.as_ref(), tmp.path()).await; assert!(diff.contains("from_store")); } diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index 1c687c6ec..ccde6ac56 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -8,12 +8,14 @@ use fabro_retro::retro_agent::{ use super::types::{Executed, RetroOptions, Retroed}; use crate::event::WorkflowRunEvent; +#[cfg(test)] +use crate::records::RunRecord; pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { - let cp = match options.run_store.get_checkpoint().await { - Ok(Some(cp)) => cp, + let state = match options.run_store.state().await { + Ok(state) => state, Err(e) => { - tracing::warn!(error = %e, "Could not load checkpoint, skipping retro"); + tracing::warn!(error = %e, "Could not load run state, skipping retro"); if let Some(ref emitter) = options.emitter { emitter.emit(&WorkflowRunEvent::RetroFailed { error: e.to_string(), @@ -22,7 +24,10 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { } return None; } - Ok(None) => { + }; + let cp = match state.checkpoint { + Some(cp) => cp, + None => { tracing::warn!("Could not load checkpoint, skipping retro"); if let Some(ref emitter) = options.emitter { emitter.emit(&WorkflowRunEvent::RetroFailed { @@ -86,7 +91,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { }); run_retro_agent( &options.sandbox, - Some(&*options.run_store), + options.run_store.as_ref(), &options.run_dir, client, options.provider, @@ -213,15 +218,30 @@ mod tests { run_dir: &std::path::Path, checkpoint: &Checkpoint, ) -> Arc { + let created_at = Utc::now(); let inner = InMemoryStore::default() .create_run( &test_run_id(), - Utc::now(), + created_at, Some(run_dir.to_string_lossy().as_ref()), ) .await .unwrap(); let run_store: Arc = inner; + run_store + .put_run(&RunRecord { + run_id: test_run_id(), + created_at, + settings: Settings::default(), + graph: Graph::new("test"), + workflow_slug: None, + working_directory: run_dir.to_path_buf(), + host_repo_path: None, + base_branch: None, + labels: std::collections::HashMap::new(), + }) + .await + .unwrap(); run_store.put_checkpoint(checkpoint).await.unwrap(); run_store } diff --git a/lib/crates/fabro-workflow/src/pipeline/types.rs b/lib/crates/fabro-workflow/src/pipeline/types.rs index 567ccc0c4..23b02b45a 100644 --- a/lib/crates/fabro-workflow/src/pipeline/types.rs +++ b/lib/crates/fabro-workflow/src/pipeline/types.rs @@ -356,7 +356,7 @@ pub struct FinalizeOptions { /// Options for the PULL_REQUEST phase. pub struct PullRequestOptions { pub run_dir: PathBuf, - pub run_store: Option>, + pub run_store: Arc, pub pr_config: Option, pub github_app: Option, pub origin_url: Option,