Fix store migration gaps: retro agent, finalize commit, engine failure, and hydration tolerance

- retro_agent::upload_data_files reads from RunStore first with filesystem
  fallback for progress.jsonl, checkpoint, run record, and start record
- write_finalize_commit reads retro.json from store before falling back to disk
- persist_terminal_engine_failure uses build_conclusion_from_store instead of
  disk-only build_conclusion
- open_or_hydrate_run tolerates malformed checkpoint/conclusion/retro/sandbox
  JSON files during hydration (warns and skips instead of failing)
- Box<DbReader> in SlateRunDb fixes clippy large_enum_variant warning
- Fix tests that called open_or_hydrate_run on dirs without run.json
- Nextest test-groups replace global thread cap for better parallelism
- opt-level=1 for dev dependencies shrinks test binary sizes

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

.config/nextest.toml

@@ -1,12 +1,39 @@
 [profile.default]
-# Cap parallelism: the workspace has ~33 test binaries (30-84 MB each).
-# At num-cpus concurrency the I/O from loading those binaries saturates
-# the system and pushes even trivial tests past the 4s kill timeout.
-# 12 threads keeps wall-clock time the same (~27s) with zero timeouts.
-test-threads = 12
 # Unit tests: flag SLOW after 2s, hard-kill after 4s
 slow-timeout = { period = "2s", terminate-after = 2 }
 
+# Test binary sizes range from 1-84 MB. Without concurrency limits the I/O
+# from loading many large binaries simultaneously saturates the system and
+# pushes trivial tests past the SLOW / kill thresholds.  Test groups cap
+# concurrency for heavy and medium binaries while letting lightweight ones
+# run at full parallelism.
+[test-groups]
+heavy  = { max-threads = 2 }   # 40-84 MB binaries
+medium = { max-threads = 4 }   # 18-30 MB binaries
+
+[[profile.default.overrides]]
+filter = """
+  package(fabro-api)
+  | package(fabro-workflows)
+  | package(fabro-agent)
+  | package(fabro-cli)
+"""
+test-group = 'heavy'
+
+[[profile.default.overrides]]
+filter = """
+  package(fabro-hooks)
+  | package(fabro-llm)
+  | package(fabro-openai-oauth)
+  | package(fabro-tracker)
+  | package(fabro-telemetry)
+  | package(fabro-store)
+  | package(fabro-github)
+  | package(fabro-devcontainer)
+  | package(fabro-sandbox)
+"""
+test-group = 'medium'
+
 [profile.e2e]
 # E2E (ignored) tests: flag SLOW after 10s, hard-kill after 30s
 slow-timeout = { period = "10s", terminate-after = 3 }

Cargo.lock

@@ -1812,12 +1812,14 @@ dependencies = [
  "chrono",
  "fabro-agent",
  "fabro-llm",
+ "fabro-store",
  "fabro-types",
  "fabro-util",
  "serde",
  "serde_json",
  "tempfile",
  "tokio",
+ "tracing",
 ]
 
 [[package]]

Cargo.toml

@@ -105,6 +105,7 @@ strip = true
 
 [profile.dev.package."*"]
 debug = false # Disable debug info for all dependencies
+opt-level = 1  # Shrinks monomorphized generics, reducing test binary size
 
 # regex is extremely slow in debug builds (~10s to compile gitleaks patterns)
 [profile.dev.package.regex]

lib/crates/fabro-retro/Cargo.toml

@@ -16,11 +16,13 @@ anyhow = "1"
 chrono = { workspace = true, features = ["serde"] }
 fabro-agent = { path = "../fabro-agent" }
 fabro-llm = { path = "../fabro-llm" }
+fabro-store = { path = "../fabro-store" }
 fabro-types = { path = "../fabro-types" }
 fabro-util = { path = "../fabro-util" }
 serde.workspace = true
 serde_json.workspace = true
 tokio.workspace = true
+tracing.workspace = true
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["test-util", "macros"] }

lib/crates/fabro-retro/src/retro_agent.rs

@@ -10,6 +10,7 @@ use fabro_agent::{
 use fabro_llm::client::Client;
 use fabro_llm::provider::Provider;
 use fabro_llm::types::ToolDefinition;
+use fabro_store::RunStore;
 use fabro_util::redact::redact_jsonl_line;
 use tokio::sync::broadcast::Receiver;
 use tokio::task::JoinHandle;
@@ -118,6 +119,7 @@ const SUBMIT_RETRO_SCHEMA: &str = r#"{
 /// files via tool access, then calls `submit_retro` with its analysis.
 pub async fn run_retro_agent(
     sandbox: &Arc<dyn Sandbox>,
+    run_store: Option<&dyn RunStore>,
     run_dir: &Path,
     llm_client: &Client,
     provider: Provider,
@@ -127,7 +129,7 @@ pub async fn run_retro_agent(
     // Upload data files into sandbox (needed for Daytona; no-op effect for local
     // since the agent can also read from the original paths via tools).
     let retro_data_dir = "/tmp/retro_data";
-    upload_data_files(sandbox, run_dir, retro_data_dir).await?;
+    upload_data_files(sandbox, run_store, run_dir, retro_data_dir).await?;
 
     // Build provider profile with the submit_retro tool
     let captured: Arc<Mutex<Option<RetroNarrative>>> = Arc::new(Mutex::new(None));
@@ -348,6 +350,7 @@ fn build_profile(provider: Provider, model: &str) -> Box<dyn AgentProfile> {
 
 async fn upload_data_files(
     sandbox: &Arc<dyn Sandbox>,
+    run_store: Option<&dyn RunStore>,
     run_dir: &Path,
     target_dir: &str,
 ) -> anyhow::Result<()> {
@@ -357,23 +360,125 @@ async fn upload_data_files(
         .await
         .map_err(|e| anyhow::anyhow!("Failed to create retro data dir: {e}"))?;
 
-    let files = [
-        "progress.jsonl",
-        "checkpoint.json",
-        "run.json",
-        "start.json",
-    ];
-    for filename in &files {
-        let source = run_dir.join(filename);
-        if source.exists() {
-            let content = std::fs::read_to_string(&source)?;
-            sandbox
-                .write_file(&format!("{target_dir}/{filename}"), &content)
-                .await
-                .map_err(|e| anyhow::anyhow!("Failed to upload {filename}: {e}"))?;
+    // progress.jsonl — try store first, fall back to filesystem
+    let progress_content = if let Some(store) = run_store {
+        match store.list_events().await {
+            Ok(envelopes) => {
+                let lines: Vec<String> = envelopes
+                    .into_iter()
+                    .filter_map(|env| serde_json::to_string(env.payload.as_value()).ok())
+                    .collect();
+                if lines.is_empty() {
+                    None
+                } else {
+                    Some(lines.join("\n") + "\n")
+                }
+            }
+            Err(e) => {
+                tracing::debug!(error = %e, "Could not read events from store, falling back to filesystem");
+                None
+            }
         }
+    } else {
+        None
+    };
+    let progress_content = if progress_content.is_some() {
+        progress_content
+    } else {
+        let source = run_dir.join("progress.jsonl");
+        if source.exists() {
+            Some(std::fs::read_to_string(&source)?)
+        } else {
+            None
+        }
+    };
+    if let Some(content) = progress_content {
+        sandbox
+            .write_file(&format!("{target_dir}/progress.jsonl"), &content)
+            .await
+            .map_err(|e| anyhow::anyhow!("Failed to upload progress.jsonl: {e}"))?;
     }
 
+    // checkpoint.json — try store first, fall back to filesystem
+    let checkpoint_content = if let Some(store) = run_store {
+        match store.get_checkpoint().await {
+            Ok(Some(cp)) => serde_json::to_string_pretty(&cp).ok(),
+            Ok(None) => None,
+            Err(e) => {
+                tracing::debug!(error = %e, "Could not read checkpoint from store, falling back to filesystem");
+                None
+            }
+        }
+    } else {
+        None
+    };
+    upload_file_with_fallback(
+        sandbox,
+        run_dir,
+        target_dir,
+        "checkpoint.json",
+        checkpoint_content,
+    )
+    .await?;
+
+    // run.json — try store first, fall back to filesystem
+    let run_content = if let Some(store) = run_store {
+        match store.get_run().await {
+            Ok(Some(run)) => serde_json::to_string_pretty(&run).ok(),
+            Ok(None) => None,
+            Err(e) => {
+                tracing::debug!(error = %e, "Could not read run from store, falling back to filesystem");
+                None
+            }
+        }
+    } else {
+        None
+    };
+    upload_file_with_fallback(sandbox, run_dir, target_dir, "run.json", run_content).await?;
+
+    // start.json — try store first, fall back to filesystem
+    let start_content = if let Some(store) = run_store {
+        match store.get_start().await {
+            Ok(Some(start)) => serde_json::to_string_pretty(&start).ok(),
+            Ok(None) => None,
+            Err(e) => {
+                tracing::debug!(error = %e, "Could not read start from store, falling back to filesystem");
+                None
+            }
+        }
+    } else {
+        None
+    };
+    upload_file_with_fallback(sandbox, run_dir, target_dir, "start.json", start_content).await?;
+
+    Ok(())
+}
+
+/// Upload a single file to the sandbox. If `store_content` is `Some`, use it directly;
+/// otherwise fall back to reading from `run_dir/filename` on the filesystem.
+async fn upload_file_with_fallback(
+    sandbox: &Arc<dyn Sandbox>,
+    run_dir: &Path,
+    target_dir: &str,
+    filename: &str,
+    store_content: Option<String>,
+) -> anyhow::Result<()> {
+    let content = if store_content.is_some() {
+        store_content
+    } else {
+        let source = run_dir.join(filename);
+        if source.exists() {
+            Some(std::fs::read_to_string(&source)?)
+        } else {
+            None
+        }
+    };
+    if let Some(content) = content {
+        sandbox
+            .write_file(&format!("{target_dir}/{filename}"), &content)
+            .await
+            .map_err(|e| anyhow::anyhow!("Failed to upload {filename}: {e}"))?;
+    }
     Ok(())
 }
 

lib/crates/fabro-store/src/slate/run_store.rs

@@ -42,7 +42,7 @@ pub(crate) struct SlateRunStoreInner {
 
 enum SlateRunDb {
     Writer(slatedb::Db),
-    Reader(DbReader),
+    Reader(Box<DbReader>),
 }
 
 impl SlateRunStore {
@@ -74,7 +74,7 @@ impl SlateRunStore {
                 created_at: record.created_at,
                 db_prefix: record.db_prefix,
                 run_dir: record.run_dir,
-                db: SlateRunDb::Reader(db),
+                db: SlateRunDb::Reader(Box::new(db)),
                 event_seq: AtomicU32::new(event_seq),
                 checkpoint_seq: AtomicU32::new(checkpoint_seq),
                 close_lock: Mutex::new(()),
@@ -518,7 +518,7 @@ impl SlateRunDb {
     async fn get_json<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>> {
         match self {
             Self::Writer(db) => get_json(db, key).await,
-            Self::Reader(db) => get_json(db, key).await,
+            Self::Reader(db) => get_json(db.as_ref(), key).await,
         }
     }
 
@@ -529,7 +529,7 @@ impl SlateRunDb {
     async fn get_text(&self, key: &str) -> Result<Option<String>> {
         match self {
             Self::Writer(db) => get_text(db, key).await,
-            Self::Reader(db) => get_text(db, key).await,
+            Self::Reader(db) => get_text(db.as_ref(), key).await,
         }
     }
 
@@ -564,14 +564,14 @@ impl SlateRunDb {
     async fn list_events_from(&self, start_seq: u32) -> Result<Vec<EventEnvelope>> {
         match self {
             Self::Writer(db) => list_events_from(db, start_seq).await,
-            Self::Reader(db) => list_events_from(db, start_seq).await,
+            Self::Reader(db) => list_events_from(db.as_ref(), start_seq).await,
         }
     }
 
     async fn list_checkpoints(&self) -> Result<Vec<(u32, Checkpoint)>> {
         match self {
             Self::Writer(db) => list_checkpoints(db).await,
-            Self::Reader(db) => list_checkpoints(db).await,
+            Self::Reader(db) => list_checkpoints(db.as_ref()).await,
         }
     }
 }

lib/crates/fabro-workflows/src/operations/hydrate.rs

@@ -47,23 +47,45 @@ pub async fn open_or_hydrate_run(
     if let Some(start) = load_start_record(run_dir)? {
         run_store.put_start(&start).await.map_err(store_error)?;
     }
-    if let Some(checkpoint) = load_checkpoint(run_dir)? {
-        run_store
-            .put_checkpoint(&checkpoint)
-            .await
-            .map_err(store_error)?;
+    match load_checkpoint(run_dir) {
+        Ok(Some(checkpoint)) => {
+            run_store
+                .put_checkpoint(&checkpoint)
+                .await
+                .map_err(store_error)?;
+        }
+        Ok(None) => {}
+        Err(err) => {
+            tracing::warn!(error = %err, "Skipping malformed checkpoint.json during hydration")
+        }
     }
-    if let Some(conclusion) = load_conclusion(run_dir)? {
-        run_store
-            .put_conclusion(&conclusion)
-            .await
-            .map_err(store_error)?;
+    match load_conclusion(run_dir) {
+        Ok(Some(conclusion)) => {
+            run_store
+                .put_conclusion(&conclusion)
+                .await
+                .map_err(store_error)?;
+        }
+        Ok(None) => {}
+        Err(err) => {
+            tracing::warn!(error = %err, "Skipping malformed conclusion.json during hydration")
+        }
     }
-    if let Some(retro) = load_retro(run_dir)? {
-        run_store.put_retro(&retro).await.map_err(store_error)?;
+    match load_retro(run_dir) {
+        Ok(Some(retro)) => {
+            run_store.put_retro(&retro).await.map_err(store_error)?;
+        }
+        Ok(None) => {}
+        Err(err) => tracing::warn!(error = %err, "Skipping malformed retro.json during hydration"),
     }
-    if let Some(sandbox) = load_sandbox_record(run_dir)? {
-        run_store.put_sandbox(&sandbox).await.map_err(store_error)?;
+    match load_sandbox_record(run_dir) {
+        Ok(Some(sandbox)) => {
+            run_store.put_sandbox(&sandbox).await.map_err(store_error)?;
+        }
+        Ok(None) => {}
+        Err(err) => {
+            tracing::warn!(error = %err, "Skipping malformed sandbox.json during hydration")
+        }
     }
 
     hydrate_events(run_dir, &record.run_id, run_store.as_ref()).await?;

lib/crates/fabro-workflows/src/operations/start.rs

@@ -25,8 +25,8 @@ use crate::handler::HandlerRegistry;
 use crate::outcome::{Outcome, StageStatus};
 use crate::pipeline::{
     self, DevcontainerSpec, FinalizeOptions, Finalized, InitOptions, LlmSpec, Persisted,
-    PullRequestOptions, RetroOptions, SandboxEnvSpec, build_conclusion, classify_engine_result,
-    persist_terminal_outcome,
+    PullRequestOptions, RetroOptions, SandboxEnvSpec, build_conclusion_from_store,
+    classify_engine_result, persist_terminal_outcome,
 };
 use crate::records::{Checkpoint, Conclusion, ConclusionExt, RunRecord, RunRecordExt};
 use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
@@ -199,13 +199,15 @@ async fn persist_terminal_engine_failure(
     let engine_result: Result<Outcome, FabroError> = Err(error.clone());
     let (final_status, failure_reason, run_status, status_reason) =
         classify_engine_result(&engine_result);
-    let conclusion = build_conclusion(
+    let conclusion = build_conclusion_from_store(
+        run_store,
         run_dir,
         final_status,
         failure_reason,
         u64::try_from(duration.as_millis()).unwrap(),
         None,
-    );
+    )
+    .await;
     persist_terminal_outcome(run_dir, &conclusion, run_status, status_reason);
     if let Err(err) = run_store.put_conclusion(&conclusion).await {
         tracing::warn!(error = %err, "Failed to save terminal engine failure conclusion to store");
@@ -812,7 +814,7 @@ fn write_failure_conclusion(
     _reason: Option<StatusReason>,
 ) -> Result<Conclusion, FabroError> {
     if run_dir.join("conclusion.json").exists() {
-        return Conclusion::load(&run_dir.join("conclusion.json")).map_err(Into::into);
+        return Conclusion::load(&run_dir.join("conclusion.json"));
     }
 
     let conclusion = build_failure_conclusion(message);
@@ -1015,13 +1017,27 @@ mod tests {
         let registry = Arc::new(test_registry());
 
         persisted_workflow(MINIMAL_DOT, &run_dir);
-        std::fs::write(run_dir.join("checkpoint.json"), "{}").unwrap();
+        let services = test_start_services(&run_dir, emitter, registry).await;
 
-        let result = start(
-            &run_dir,
-            test_start_services(&run_dir, emitter, registry).await,
-        )
-        .await;
+        // Write a checkpoint to the store (not disk) so start() sees it
+        let checkpoint = Checkpoint::from_context(
+            &Context::new(),
+            "start",
+            vec!["start".to_string()],
+            HashMap::new(),
+            HashMap::new(),
+            Some("exit".to_string()),
+            HashMap::new(),
+            HashMap::new(),
+            HashMap::new(),
+        );
+        services
+            .run_store
+            .put_checkpoint(&checkpoint)
+            .await
+            .unwrap();
+
+        let result = start(&run_dir, services).await;
 
         assert!(
             matches!(&result, Err(crate::error::FabroError::Precondition(_))),

lib/crates/fabro-workflows/src/pipeline/execute/tests.rs

@@ -140,8 +140,10 @@ fn test_lifecycle(setup_commands: Vec<String>) -> LifecycleOptions {
     }
 }
 
-async fn test_run_store(run_dir: &Path) -> Arc<dyn fabro_store::RunStore> {
-    crate::operations::open_or_hydrate_run(&InMemoryStore::default(), run_dir)
+async fn test_run_store(_run_dir: &Path) -> Arc<dyn fabro_store::RunStore> {
+    let store: &dyn fabro_store::Store = &InMemoryStore::default();
+    store
+        .create_run("test-run", chrono::Utc::now(), None)
         .await
         .unwrap()
 }

lib/crates/fabro-workflows/src/pipeline/finalize.rs

@@ -262,7 +262,11 @@ pub fn persist_terminal_outcome(
 ///
 /// This captures the last diff.patch (written after the final checkpoint) and retro.json.
 /// Best-effort: errors are logged as warnings.
-pub async fn write_finalize_commit(run_options: &RunOptions, run_dir: &Path) {
+pub async fn write_finalize_commit(
+    run_options: &RunOptions,
+    run_dir: &Path,
+    run_store: &dyn RunStore,
+) {
     let (Some(meta_branch), Some(repo_path)) = (
         run_options
             .git
@@ -275,8 +279,12 @@ pub async fn write_finalize_commit(run_options: &RunOptions, run_dir: &Path) {
 
     let store = MetadataStore::new(repo_path, &run_options.git_author);
     let mut entries = scan_node_files(run_dir);
-    if let Ok(retro_bytes) = std::fs::read(run_dir.join("retro.json")) {
-        entries.push(("retro.json".to_string(), retro_bytes));
+    let retro_bytes = match run_store.get_retro().await {
+        Ok(Some(retro)) => serde_json::to_vec_pretty(&retro).ok(),
+        _ => std::fs::read(run_dir.join("retro.json")).ok(),
+    };
+    if let Some(bytes) = retro_bytes {
+        entries.push(("retro.json".to_string(), bytes));
     }
     let refs: Vec<(&str, &[u8])> = entries
         .iter()
@@ -360,7 +368,7 @@ pub async fn finalize(
     )
     .await;
 
-    write_finalize_commit(&run_options, &options.run_dir).await;
+    write_finalize_commit(&run_options, &options.run_dir, options.run_store.as_ref()).await;
 
     if options.preserve_sandbox {
         let info = sandbox.sandbox_info();

lib/crates/fabro-workflows/src/pipeline/initialize.rs

@@ -737,12 +737,13 @@ mod tests {
             persisted,
             InitOptions {
                 run_id: "run-test".to_string(),
-                run_store: crate::operations::open_or_hydrate_run(
-                    &InMemoryStore::default(),
-                    &run_dir,
-                )
-                .await
-                .unwrap(),
+                run_store: {
+                    let store: &dyn fabro_store::Store = &InMemoryStore::default();
+                    store
+                        .create_run("test-run", chrono::Utc::now(), None)
+                        .await
+                        .unwrap()
+                },
                 dry_run: false,
                 emitter,
                 sandbox: SandboxSpec::Local {
@@ -806,12 +807,13 @@ mod tests {
             persisted,
             InitOptions {
                 run_id: "run-test".to_string(),
-                run_store: crate::operations::open_or_hydrate_run(
-                    &InMemoryStore::default(),
-                    &run_dir,
-                )
-                .await
-                .unwrap(),
+                run_store: {
+                    let store: &dyn fabro_store::Store = &InMemoryStore::default();
+                    store
+                        .create_run("test-run", chrono::Utc::now(), None)
+                        .await
+                        .unwrap()
+                },
                 dry_run: false,
                 emitter,
                 sandbox: SandboxSpec::Local {

lib/crates/fabro-workflows/src/pipeline/mod.rs

@@ -10,6 +10,7 @@ pub(crate) mod types;
 mod validate;
 
 pub use execute::execute;
+pub(crate) use finalize::build_conclusion_from_store;
 pub use finalize::{
     build_conclusion, classify_engine_result, finalize, persist_terminal_outcome,
     write_finalize_commit,

lib/crates/fabro-workflows/src/pipeline/retro.rs

@@ -90,6 +90,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
             });
         run_retro_agent(
             &options.sandbox,
+            Some(&*options.run_store),
             &options.run_dir,
             client,
             options.provider,

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

@@ -173,7 +173,7 @@ pub async fn scan_runs_combined(store: &dyn Store, base: &Path) -> Result<Vec<Ru
             };
             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(|status| status.is_terminal()) {
+            let end_time = if summary.status.is_some_and(RunStatus::is_terminal) {
                 summary.duration_ms.and_then(|duration_ms| {
                     Some(
                         start_time_dt