From cb7d83e118278f3d2d4b4ebe391f0a39f85157ee Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Wed, 1 Apr 2026 20:23:00 -0700 Subject: [PATCH] Cut operational run readers over to store-only state --- lib/crates/fabro-cli/src/commands/run/diff.rs | 28 +---- lib/crates/fabro-cli/src/commands/run/logs.rs | 24 +++- .../fabro-cli/src/commands/run/output.rs | 5 +- .../fabro-cli/src/commands/run/preview.rs | 23 ++-- lib/crates/fabro-cli/src/commands/run/ssh.rs | 23 ++-- lib/crates/fabro-server/src/server.rs | 8 +- lib/crates/fabro-workflow/src/run_lookup.rs | 106 +++++++++--------- 7 files changed, 100 insertions(+), 117 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/run/diff.rs b/lib/crates/fabro-cli/src/commands/run/diff.rs index e9461744f..c1da591ae 100644 --- a/lib/crates/fabro-cli/src/commands/run/diff.rs +++ b/lib/crates/fabro-cli/src/commands/run/diff.rs @@ -85,10 +85,7 @@ async fn resolve_diff( let start = match run_store { Some(run_store) => run_store .get_start() - .await - .ok() - .flatten() - .or_else(|| StartRecord::load(run_dir).ok()) + .await? .context("Failed to load start.json")?, None => StartRecord::load(run_dir).context("Failed to load start.json")?, }; @@ -105,17 +102,8 @@ async fn resolve_diff( } } - let final_patch_path = run_dir.join("final.patch"); - if final_patch_path.exists() { - debug!("Reading final.patch"); - return std::fs::read_to_string(&final_patch_path).context("Failed to read final.patch"); - } - let run_concluded = match run_store { - Some(run_store) => { - run_store.get_conclusion().await.ok().flatten().is_some() - || run_dir.join("conclusion.json").exists() - } + Some(run_store) => run_store.get_conclusion().await?.is_some(), None => run_dir.join("conclusion.json").exists(), }; if run_concluded { @@ -125,18 +113,12 @@ async fn resolve_diff( } debug!("No final.patch found; attempting live diff from sandbox"); - let sandbox_json = run_dir.join("sandbox.json"); let record = match run_store { Some(run_store) => run_store .get_sandbox() - .await - .ok() - .flatten() - .or_else(|| fabro_sandbox::SandboxRecord::load(&sandbox_json).ok()) - .context( - "Failed to load sandbox.json — was this run started with a recent version of arc?", - )?, - None => fabro_sandbox::SandboxRecord::load(&sandbox_json).context( + .await? + .context("Failed to load sandbox record from store")?, + None => fabro_sandbox::SandboxRecord::load(&run_dir.join("sandbox.json")).context( "Failed to load sandbox.json — was this run started with a recent version of arc?", )?, }; diff --git a/lib/crates/fabro-cli/src/commands/run/logs.rs b/lib/crates/fabro-cli/src/commands/run/logs.rs index a86411e49..cb94410b4 100644 --- a/lib/crates/fabro-cli/src/commands/run/logs.rs +++ b/lib/crates/fabro-cli/src/commands/run/logs.rs @@ -11,6 +11,7 @@ use fabro_util::redact::redact_jsonl_line; use fabro_util::terminal::Styles; use fabro_workflow::run_lookup::{resolve_run_combined, runs_base}; use futures::StreamExt; +use serde_json::{Map, Value}; use tokio::time; use tracing::{debug, info}; @@ -189,7 +190,7 @@ async fn follow_store_logs( Ok(()) } -async fn run_concluded(run_store: &dyn RunStore, run_dir: &Path) -> Result { +async fn run_concluded(run_store: &dyn RunStore, _run_dir: &Path) -> Result { if run_store .get_conclusion() .await @@ -203,8 +204,7 @@ async fn run_concluded(run_store: &dyn RunStore, run_dir: &Path) -> Result .get_status() .await .context("Failed to read status from store while following logs")? - .is_some_and(|record| record.status.is_terminal()) - || run_dir.join("conclusion.json").exists()) + .is_some_and(|record| record.status.is_terminal())) } async fn flush_remaining_store_events( @@ -234,10 +234,26 @@ async fn flush_remaining_store_events( } fn event_payload_line(event: &fabro_store::EventEnvelope) -> Result { - let line = serde_json::to_string(event.payload.as_value())?; + let line = serde_json::to_string(&normalize_json_value(event.payload.as_value().clone()))?; Ok(redact_jsonl_line(&line)) } +fn normalize_json_value(value: Value) -> Value { + match value { + Value::Object(map) => Value::Object( + map.into_iter() + .map(|(key, value)| (key, normalize_json_value(value))) + .collect::>() + .into_iter() + .collect::>(), + ), + Value::Array(values) => { + Value::Array(values.into_iter().map(normalize_json_value).collect()) + } + other => other, + } +} + fn render_indented_markdown(styles: &Styles, text: &str, indent: &str) -> String { let term_width = Styles::terminal_width(); let wrap_width = term_width.saturating_sub(indent.len()); diff --git a/lib/crates/fabro-cli/src/commands/run/output.rs b/lib/crates/fabro-cli/src/commands/run/output.rs index 5a5798ba6..8f616b36a 100644 --- a/lib/crates/fabro-cli/src/commands/run/output.rs +++ b/lib/crates/fabro-cli/src/commands/run/output.rs @@ -209,9 +209,8 @@ pub(crate) async fn print_final_output( ) { let checkpoint = match run_store { Some(run_store) => run_store.get_checkpoint().await.ok().flatten(), - None => None, - } - .or_else(|| Checkpoint::load(&run_dir.join("checkpoint.json")).ok()); + None => Checkpoint::load(&run_dir.join("checkpoint.json")).ok(), + }; let Some(checkpoint) = checkpoint else { return; }; diff --git a/lib/crates/fabro-cli/src/commands/run/preview.rs b/lib/crates/fabro-cli/src/commands/run/preview.rs index 6fe90adea..4b99e2367 100644 --- a/lib/crates/fabro-cli/src/commands/run/preview.rs +++ b/lib/crates/fabro-cli/src/commands/run/preview.rs @@ -1,6 +1,5 @@ use anyhow::{Context, Result}; use fabro_config::FabroSettingsExt; -use fabro_sandbox::SandboxRecordExt; use fabro_sandbox::daytona::DaytonaSandbox; use fabro_workflow::run_lookup::{resolve_run_combined, runs_base}; use tracing::info; @@ -15,21 +14,13 @@ 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 sandbox_json = run.path.join("sandbox.json"); - let record = match store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await? { - Some(run_store) => run_store - .get_sandbox() - .await - .ok() - .flatten() - .or_else(|| fabro_sandbox::SandboxRecord::load(&sandbox_json).ok()) - .context( - "Failed to load sandbox.json — was this run started with a recent version of arc?", - )?, - None => fabro_sandbox::SandboxRecord::load(&sandbox_json).context( - "Failed to load sandbox.json — was this run started with a recent version of arc?", - )?, - }; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id) + .await? + .context("Failed to open run store")?; + let record = run_store + .get_sandbox() + .await? + .context("Failed to load sandbox record from store")?; validate_daytona_provider(&record, "Preview URLs")?; diff --git a/lib/crates/fabro-cli/src/commands/run/ssh.rs b/lib/crates/fabro-cli/src/commands/run/ssh.rs index 5d27d2a8e..4d0f98d4d 100644 --- a/lib/crates/fabro-cli/src/commands/run/ssh.rs +++ b/lib/crates/fabro-cli/src/commands/run/ssh.rs @@ -1,6 +1,5 @@ use anyhow::{Context, Result, bail}; use fabro_config::FabroSettingsExt; -use fabro_sandbox::SandboxRecordExt; use fabro_sandbox::daytona::DaytonaSandbox; use fabro_workflow::run_lookup::{resolve_run_combined, runs_base}; use tracing::info; @@ -19,21 +18,13 @@ 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 sandbox_json = run.path.join("sandbox.json"); - let record = match store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await? { - Some(run_store) => run_store - .get_sandbox() - .await - .ok() - .flatten() - .or_else(|| fabro_sandbox::SandboxRecord::load(&sandbox_json).ok()) - .context( - "Failed to load sandbox.json — was this run started with a recent version of arc?", - )?, - None => fabro_sandbox::SandboxRecord::load(&sandbox_json).context( - "Failed to load sandbox.json — was this run started with a recent version of arc?", - )?, - }; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id) + .await? + .context("Failed to open run store")?; + let record = run_store + .get_sandbox() + .await? + .context("Failed to load sandbox record from store")?; validate_daytona_provider(&record, "SSH access")?; diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 59f6d6ee9..b60c36fc3 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -52,7 +52,7 @@ use fabro_workflow::context::Context; use fabro_workflow::event::{EventEmitter, RunEventEnvelope}; use fabro_workflow::operations::{self, CreateRunInput, WorkflowInput}; use fabro_workflow::pipeline::Persisted; -use fabro_workflow::records::{Checkpoint, CheckpointExt}; +use fabro_workflow::records::Checkpoint; pub use fabro_api_types::{ ApiQuestion, ApiQuestionOption, PaginatedRunList, PaginationMeta, @@ -757,12 +757,10 @@ async fn execute_run(state: Arc, run_id: RunId) { // Save final checkpoint let checkpoint = match run_store.get_checkpoint().await { - Ok(checkpoint) => { - checkpoint.or_else(|| Checkpoint::load(&run_dir.join("checkpoint.json")).ok()) - } + Ok(checkpoint) => checkpoint, Err(err) => { tracing::warn!(run_id = %run_id, error = %err, "Failed to load checkpoint from store"); - Checkpoint::load(&run_dir.join("checkpoint.json")).ok() + None } }; diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs index 25cee66e2..cf3cc624f 100644 --- a/lib/crates/fabro-workflow/src/run_lookup.rs +++ b/lib/crates/fabro-workflow/src/run_lookup.rs @@ -173,69 +173,75 @@ fn scan_runs_inner(base: &Path, include_status: bool) -> Result> { } pub async fn scan_runs_combined(store: &dyn Store, base: &Path) -> Result> { - let mut runs_by_id: HashMap = scan_runs_without_status(base)? - .into_iter() - .map(|run| (run.run_id, run)) - .collect(); + let mut runs_by_id: HashMap = HashMap::new(); if let Ok(store_runs) = store.list_runs(&ListRunsQuery::default()).await { for summary in store_runs { - let Some(run_dir) = summary.run_dir.as_deref() else { + let Some(run_info) = run_info_from_summary(&summary) else { continue; }; - let path = PathBuf::from(run_dir); - if !path.exists() { - continue; - } - let Some(dir_name) = path - .file_name() - .map(|name| name.to_string_lossy().to_string()) - else { - continue; - }; - let start_time_dt = summary.created_at; - let start_time = summary.start_time.unwrap_or(start_time_dt); - let end_time = if summary.status.is_some_and(RunStatus::is_terminal) { - summary.duration_ms.and_then(|duration_ms| { - Some( - start_time_dt - + chrono::Duration::milliseconds(i64::try_from(duration_ms).ok()?), - ) - }) - } else { - None - }; - runs_by_id.insert( - summary.run_id, - RunInfo { - run_id: summary.run_id, - dir_name, - workflow_name: summary - .workflow_name - .unwrap_or_else(|| "[starting]".to_string()), - workflow_slug: summary.workflow_slug, - status: summary.status.unwrap_or(RunStatus::Dead), - status_reason: summary.status_reason, - start_time: start_time.to_rfc3339(), - labels: summary.labels, - duration_ms: summary.duration_ms, - total_cost: summary.total_cost, - host_repo_path: summary.host_repo_path, - goal: summary.goal.unwrap_or_default(), - start_time_dt: Some(start_time_dt), - end_time, - path, - is_orphan: false, - }, - ); + runs_by_id.insert(summary.run_id, run_info); } } + let store_run_ids = runs_by_id + .keys() + .copied() + .collect::>(); + for run in scan_runs_without_status(base)? + .into_iter() + .filter(|run| run.is_orphan && !store_run_ids.contains(&run.run_id)) + { + runs_by_id.insert(run.run_id, run); + } + let mut runs: Vec<_> = runs_by_id.into_values().collect(); runs.sort_by(|a, b| b.start_time_dt.cmp(&a.start_time_dt)); Ok(runs) } +fn run_info_from_summary(summary: &fabro_store::RunSummary) -> Option { + let run_dir = summary.run_dir.as_deref()?; + let path = PathBuf::from(run_dir); + if !path.exists() { + return None; + } + let dir_name = path + .file_name() + .map(|name| name.to_string_lossy().to_string())?; + let start_time_dt = summary.created_at; + let start_time = summary.start_time.unwrap_or(start_time_dt); + let end_time = if summary.status.is_some_and(RunStatus::is_terminal) { + summary.duration_ms.and_then(|duration_ms| { + Some(start_time_dt + chrono::Duration::milliseconds(i64::try_from(duration_ms).ok()?)) + }) + } else { + None + }; + + Some(RunInfo { + run_id: summary.run_id, + dir_name, + workflow_name: summary + .workflow_name + .clone() + .unwrap_or_else(|| "[starting]".to_string()), + workflow_slug: summary.workflow_slug.clone(), + status: summary.status.unwrap_or(RunStatus::Dead), + status_reason: summary.status_reason, + start_time: start_time.to_rfc3339(), + labels: summary.labels.clone(), + duration_ms: summary.duration_ms, + total_cost: summary.total_cost, + host_repo_path: summary.host_repo_path.clone(), + goal: summary.goal.clone().unwrap_or_default(), + start_time_dt: Some(start_time_dt), + end_time, + path, + is_orphan: false, + }) +} + struct StatusInfo { status: RunStatus, reason: Option,