diff --git a/.config/nextest.toml b/.config/nextest.toml index f1cb6eafb..8434bd242 100644 --- a/.config/nextest.toml +++ b/.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 } diff --git a/Cargo.lock b/Cargo.lock index c6d6bb193..87c74d289 100644 --- a/Cargo.lock +++ b/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]] diff --git a/Cargo.toml b/Cargo.toml index 70c4566b5..dfb622b0d 100644 --- a/Cargo.toml +++ b/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] diff --git a/lib/crates/fabro-retro/Cargo.toml b/lib/crates/fabro-retro/Cargo.toml index 898fc4b39..7450f1fe9 100644 --- a/lib/crates/fabro-retro/Cargo.toml +++ b/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"] } diff --git a/lib/crates/fabro-retro/src/retro_agent.rs b/lib/crates/fabro-retro/src/retro_agent.rs index 7b06dc196..f4ffce5aa 100644 --- a/lib/crates/fabro-retro/src/retro_agent.rs +++ b/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, + 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>> = Arc::new(Mutex::new(None)); @@ -348,6 +350,7 @@ fn build_profile(provider: Provider, model: &str) -> Box { async fn upload_data_files( sandbox: &Arc, + 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 = 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, + run_dir: &Path, + target_dir: &str, + filename: &str, + store_content: Option, +) -> anyhow::Result<()> { + let content = if store_content.is_some() { + store_content + } else { + let source = run_dir.join(filename); + if source.exists() { + Some(std::fs::read_to_string(&source)?) + } else { + None + } + }; + if let Some(content) = content { + sandbox + .write_file(&format!("{target_dir}/{filename}"), &content) + .await + .map_err(|e| anyhow::anyhow!("Failed to upload {filename}: {e}"))?; + } Ok(()) } diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 5d4726b93..b40b1b62b 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/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), } 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(&self, key: &str) -> Result> { 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> { 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> { 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> { 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, } } } diff --git a/lib/crates/fabro-workflows/src/operations/hydrate.rs b/lib/crates/fabro-workflows/src/operations/hydrate.rs index be70f1e85..d08674a3f 100644 --- a/lib/crates/fabro-workflows/src/operations/hydrate.rs +++ b/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?; diff --git a/lib/crates/fabro-workflows/src/operations/start.rs b/lib/crates/fabro-workflows/src/operations/start.rs index c928796aa..02be4f1dc 100644 --- a/lib/crates/fabro-workflows/src/operations/start.rs +++ b/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 = 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, ) -> Result { 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(_))), diff --git a/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs index e6d46b24c..1c7b935f4 100644 --- a/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs @@ -140,8 +140,10 @@ fn test_lifecycle(setup_commands: Vec) -> LifecycleOptions { } } -async fn test_run_store(run_dir: &Path) -> Arc { - crate::operations::open_or_hydrate_run(&InMemoryStore::default(), run_dir) +async fn test_run_store(_run_dir: &Path) -> Arc { + let store: &dyn fabro_store::Store = &InMemoryStore::default(); + store + .create_run("test-run", chrono::Utc::now(), None) .await .unwrap() } diff --git a/lib/crates/fabro-workflows/src/pipeline/finalize.rs b/lib/crates/fabro-workflows/src/pipeline/finalize.rs index 72c54a3c3..b79239ed3 100644 --- a/lib/crates/fabro-workflows/src/pipeline/finalize.rs +++ b/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(); diff --git a/lib/crates/fabro-workflows/src/pipeline/initialize.rs b/lib/crates/fabro-workflows/src/pipeline/initialize.rs index 7157420d9..9e8b54075 100644 --- a/lib/crates/fabro-workflows/src/pipeline/initialize.rs +++ b/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 { diff --git a/lib/crates/fabro-workflows/src/pipeline/mod.rs b/lib/crates/fabro-workflows/src/pipeline/mod.rs index 1880cdd4e..8ddbb9d85 100644 --- a/lib/crates/fabro-workflows/src/pipeline/mod.rs +++ b/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, diff --git a/lib/crates/fabro-workflows/src/pipeline/retro.rs b/lib/crates/fabro-workflows/src/pipeline/retro.rs index bb0618704..1d95bc0df 100644 --- a/lib/crates/fabro-workflows/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflows/src/pipeline/retro.rs @@ -90,6 +90,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { }); run_retro_agent( &options.sandbox, + Some(&*options.run_store), &options.run_dir, client, options.provider, diff --git a/lib/crates/fabro-workflows/src/run_lookup.rs b/lib/crates/fabro-workflows/src/run_lookup.rs index 50018e2a0..b7b2c78d1 100644 --- a/lib/crates/fabro-workflows/src/run_lookup.rs +++ b/lib/crates/fabro-workflows/src/run_lookup.rs @@ -173,7 +173,7 @@ pub async fn scan_runs_combined(store: &dyn Store, base: &Path) -> Result