Cut operational run readers over to store-only state

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?",
         )?,
     };

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<bool> {
+async fn run_concluded(run_store: &dyn RunStore, _run_dir: &Path) -> Result<bool> {
     if run_store
         .get_conclusion()
         .await
@@ -203,8 +204,7 @@ async fn run_concluded(run_store: &dyn RunStore, run_dir: &Path) -> Result<bool>
         .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<String> {
-    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::<std::collections::BTreeMap<_, _>>()
+                .into_iter()
+                .collect::<Map<_, _>>(),
+        ),
+        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());

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;
     };

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")?;
 

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")?;
 

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<AppState>, 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
         }
     };
 

lib/crates/fabro-workflow/src/run_lookup.rs

@@ -173,69 +173,75 @@ fn scan_runs_inner(base: &Path, include_status: bool) -> Result<Vec<RunInfo>> {
 }
 
 pub async fn scan_runs_combined(store: &dyn Store, base: &Path) -> Result<Vec<RunInfo>> {
-    let mut runs_by_id: HashMap<RunId, RunInfo> = scan_runs_without_status(base)?
-        .into_iter()
-        .map(|run| (run.run_id, run))
-        .collect();
+    let mut runs_by_id: HashMap<RunId, RunInfo> = 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::<std::collections::HashSet<_>>();
+    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<RunInfo> {
+    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<StatusReason>,