From 75b0e0879a9af8d269972f9bb9cf42b5bc01fdd3 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 2 Apr 2026 06:33:22 -0700 Subject: [PATCH] Stabilize workspace tests and clean warnings Make the full workspace nextest run reliable after the run-store migration, restore legacy test-harness projections needed by workflow integration tests, and clear the remaining fmt/clippy issues in the touched paths. --- .config/nextest.toml | 16 ++- .../fabro-cli/src/commands/run/attach.rs | 20 +-- .../fabro-cli/src/commands/run/launcher.rs | 23 +++- .../fabro-cli/src/commands/run/rewind.rs | 4 +- lib/crates/fabro-cli/src/commands/run/wait.rs | 1 + lib/crates/fabro-cli/src/main.rs | 10 +- lib/crates/fabro-cli/tests/it/cmd/attach.rs | 4 +- lib/crates/fabro-cli/tests/it/cmd/config.rs | 6 +- lib/crates/fabro-cli/tests/it/cmd/diff.rs | 1 - .../fabro-cli/tests/it/cmd/pr_create.rs | 6 +- .../fabro-cli/tests/it/cmd/sandbox_cp.rs | 2 +- .../fabro-cli/tests/it/cmd/system_df.rs | 18 +-- lib/crates/fabro-cli/tests/it/cmd/wait.rs | 4 +- .../fabro-cli/tests/it/scenario/lifecycle.rs | 20 +-- lib/crates/fabro-cli/tests/it/scenario/mod.rs | 38 ++++++ .../fabro-cli/tests/it/scenario/recovery.rs | 81 +++--------- lib/crates/fabro-server/build.rs | 11 +- lib/crates/fabro-server/src/server.rs | 20 +-- lib/crates/fabro-server/src/static_files.rs | 16 +-- lib/crates/fabro-server/src/web_auth.rs | 115 ++++++++++++------ lib/crates/fabro-server/tests/it/api.rs | 4 +- .../fabro-workflow/src/operations/resume.rs | 4 +- .../src/pipeline/pull_request.rs | 13 +- .../fabro-workflow/src/pipeline/retro.rs | 2 +- lib/crates/fabro-workflow/src/test_support.rs | 26 +++- .../fabro-workflow/tests/it/integration.rs | 5 +- 26 files changed, 276 insertions(+), 194 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index ba1335d78..5acbb2efd 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -1,11 +1,23 @@ [profile.default] -# Unit tests: flag SLOW after 2s, hard-kill after 4s -slow-timeout = { period = "3s", terminate-after = 2 } +# Unit tests: flag SLOW after 5s, hard-kill after 15s +slow-timeout = { period = "5s", terminate-after = 3 } [[profile.default.overrides]] filter = "package(fabro-cli) & kind(test)" slow-timeout = { period = "5s", terminate-after = 4 } +[[profile.default.overrides]] +filter = "package(fabro-llm)" +slow-timeout = { period = "10s", terminate-after = 3 } + +[[profile.default.overrides]] +filter = "package(fabro-oauth)" +slow-timeout = { period = "10s", terminate-after = 3 } + +[[profile.default.overrides]] +filter = "package(fabro-tracker)" +slow-timeout = { period = "10s", terminate-after = 3 } + [profile.e2e] # E2E (ignored) tests: flag SLOW after 10s, hard-kill after 30s slow-timeout = { period = "10s", terminate-after = 3 } diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index 35121d849..1d6181dda 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -57,8 +57,7 @@ pub(crate) async fn attach_run( .await .ok() .flatten() - .map(|record| record.settings.verbose_enabled()) - .unwrap_or(false); + .is_some_and(|record| record.settings.verbose_enabled()); let event_lines = events .iter() .map(event_payload_line) @@ -311,11 +310,13 @@ async fn flush_remaining_store_events( saw_new_event = true; } - if !saw_new_event || Instant::now() >= deadline { + if Instant::now() >= deadline { break; } - sleep(Duration::from_millis(100)).await; + if !saw_new_event { + sleep(Duration::from_millis(100)).await; + } } Ok(()) @@ -593,7 +594,7 @@ fn infer_storage_dir(run_dir: &Path) -> Option { } fn infer_run_id(run_dir: &Path) -> Option { - super::launcher::active_launcher_record_for_run(run_dir) + super::launcher::launcher_record_for_run(run_dir) .map(|record| record.run_id) .or_else(|| { std::fs::read_to_string(run_dir.join("id.txt")) @@ -765,7 +766,7 @@ async fn determine_exit_code_with_store(run_store: &dyn RunStore) -> ExitCode { return ExitCode::from(0); } Ok(Some(record)) if record.status.is_terminal() => return ExitCode::from(1), - Ok(Some(_)) | Ok(None) | Err(_) => {} + Ok(Some(_) | None) | Err(_) => {} } if Instant::now() >= deadline { @@ -788,6 +789,7 @@ fn process_alive(pid: u32) -> bool { #[cfg(test)] mod tests { use super::*; + use crate::commands::run::launcher; use chrono::Utc; use fabro_interview::{Answer, AnswerValue}; use fabro_util::terminal::Styles; @@ -897,9 +899,9 @@ mod tests { let run_dir = storage_dir.join("runs").join("20260401-test"); std::fs::create_dir_all(&run_dir).unwrap(); - super::launcher::write_launcher_record( - &super::launcher::launcher_record_path(&storage_dir, &fabro_types::fixtures::RUN_1), - &super::launcher::LauncherRecord { + launcher::write_launcher_record( + &launcher::launcher_record_path(&storage_dir, &fabro_types::fixtures::RUN_1), + &launcher::LauncherRecord { run_id: fabro_types::fixtures::RUN_1, run_dir: run_dir.clone(), pid: u32::MAX, diff --git a/lib/crates/fabro-cli/src/commands/run/launcher.rs b/lib/crates/fabro-cli/src/commands/run/launcher.rs index a1136280e..52e33e254 100644 --- a/lib/crates/fabro-cli/src/commands/run/launcher.rs +++ b/lib/crates/fabro-cli/src/commands/run/launcher.rs @@ -50,8 +50,23 @@ pub(crate) fn remove_launcher_record(path: &Path) { } pub(crate) fn active_launcher_record_for_run(run_dir: &Path) -> Option { + let launcher = launcher_record_for_run(run_dir)?; + let storage_dir = run_dir.parent()?.parent()?; + let path = launcher_record_path(storage_dir, &launcher.run_id); + if launcher_record_is_running(&launcher) { + Some(launcher) + } else { + remove_launcher_record(&path); + None + } +} + +pub(crate) fn launcher_record_for_run(run_dir: &Path) -> Option { if let Ok(run_record) = RunRecord::load(run_dir) { - return active_launcher_record(&run_record.settings.storage_dir(), &run_record.run_id); + return read_launcher_record(&launcher_record_path( + &run_record.settings.storage_dir(), + &run_record.run_id, + )); } let storage_dir = run_dir.parent()?.parent()?; @@ -63,11 +78,7 @@ pub(crate) fn active_launcher_record_for_run(run_dir: &Path) -> Option Styles { Styles::new(false) diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index 879f3fa2a..f283e9990 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -38,7 +38,7 @@ async fn main() { let start = std::time::Instant::now(); let raw_args: Vec = std::env::args().collect(); - let (command_name, result) = main_inner().await; + let (command_name, result) = Box::pin(main_inner()).await; let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap(); let is_error = result.is_err(); @@ -438,6 +438,8 @@ mod tests { let cli = Cli::try_parse_from([ "fabro", "__detached", + "--run-id", + "01ARZ3NDEKTSV4RRFFQ69G5FAV", "--run-dir", "/tmp/fabro/runs/01ABC", "--launcher-path", @@ -446,10 +448,12 @@ mod tests { .expect("should parse"); match *cli.command { Commands::RunCmd(RunCommands::Detached { + run_id, run_dir, launcher_path, resume, }) => { + assert_eq!(run_id, "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap()); assert_eq!(run_dir, std::path::PathBuf::from("/tmp/fabro/runs/01ABC")); assert_eq!( launcher_path, @@ -466,6 +470,8 @@ mod tests { let cli = Cli::try_parse_from([ "fabro", "__detached", + "--run-id", + "01ARZ3NDEKTSV4RRFFQ69G5FAV", "--run-dir", "/tmp/fabro/runs/01ABC", "--launcher-path", @@ -475,10 +481,12 @@ mod tests { .expect("should parse"); match *cli.command { Commands::RunCmd(RunCommands::Detached { + run_id, run_dir, launcher_path, resume, }) => { + assert_eq!(run_id, "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap()); assert_eq!(run_dir, std::path::PathBuf::from("/tmp/fabro/runs/01ABC")); assert_eq!( launcher_path, diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index 9985b069a..b3fd9596a 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -178,7 +178,7 @@ fn attach_before_completion_streams_to_finished_state() { "[DURATION]".to_string(), )); let release_gate = std::thread::spawn(move || { - std::thread::sleep(Duration::from_millis(300)); + std::thread::sleep(Duration::from_secs(1)); gate.release(); }); let mut attach_cmd = context.command(); @@ -194,6 +194,8 @@ fn attach_before_completion_streams_to_finished_state() { ----- stderr ----- Sandbox: local (ready in [TIME]) ✓ start [DURATION] + ✓ wait [DURATION] + ✓ exit [DURATION] "); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/config.rs b/lib/crates/fabro-cli/tests/it/cmd/config.rs index c9dc6d4a7..4be60cd79 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/config.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/config.rs @@ -7,6 +7,8 @@ use fabro_config::user::ExecutionMode; use fabro_test::{fabro_snapshot, test_context}; use predicates::prelude::*; +use super::support::run_snapshot; + #[test] fn help() { let context = test_context!(); @@ -442,8 +444,8 @@ fn create_explicit_workflow_path_uses_project_config_relative_to_workflow() { ) }); - let run_record: serde_json::Value = - serde_json::from_str(&std::fs::read_to_string(run_dir.join("run.json")).unwrap()).unwrap(); + let snapshot = run_snapshot(&run_dir); + let run_record = serde_json::to_value(&snapshot.run).unwrap(); assert_eq!(run_record["settings"]["auto_approve"].as_bool(), Some(true)); assert_eq!( run_record["settings"]["storage_dir"].as_str(), diff --git a/lib/crates/fabro-cli/tests/it/cmd/diff.rs b/lib/crates/fabro-cli/tests/it/cmd/diff.rs index 096cf5bfa..aa63f0673 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/diff.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/diff.rs @@ -83,7 +83,6 @@ fn diff_missing_node_diff_reports_helpful_error() { ----- stdout ----- ----- stderr ----- error: No diff found for node 'missing' — check the node ID and try again - > No such file or directory (os error 2) "); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/pr_create.rs b/lib/crates/fabro-cli/tests/it/cmd/pr_create.rs index 5d45f1130..0c288b2ad 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/pr_create.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/pr_create.rs @@ -67,9 +67,9 @@ fn pr_create_completed_dry_run_without_run_branch_errors() { fn pr_create_uses_store_run_record_without_run_json() { let context = test_context!(); let run = setup_completed_dry_run(&context); - std::fs::remove_file(run.run_dir.join("run.json")).unwrap(); - std::fs::remove_file(run.run_dir.join("start.json")).unwrap(); - std::fs::remove_file(run.run_dir.join("conclusion.json")).unwrap(); + let _ = std::fs::remove_file(run.run_dir.join("run.json")); + let _ = std::fs::remove_file(run.run_dir.join("start.json")); + let _ = std::fs::remove_file(run.run_dir.join("conclusion.json")); let mut cmd = context.command(); cmd.args(["pr", "create", &run.run_id]); diff --git a/lib/crates/fabro-cli/tests/it/cmd/sandbox_cp.rs b/lib/crates/fabro-cli/tests/it/cmd/sandbox_cp.rs index 7568b9404..036f47e58 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/sandbox_cp.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/sandbox_cp.rs @@ -73,7 +73,7 @@ fn sandbox_cp_downloads_file_from_run() { fn sandbox_cp_downloads_file_from_store_without_sandbox_json() { let context = test_context!(); let setup = setup_local_sandbox_run(&context); - std::fs::remove_file(setup.run.run_dir.join("sandbox.json")).unwrap(); + let _ = std::fs::remove_file(setup.run.run_dir.join("sandbox.json")); let dest = context.temp_dir.join("downloaded-from-store.txt"); let mut cmd = context.cp(); cmd.args([ diff --git a/lib/crates/fabro-cli/tests/it/cmd/system_df.rs b/lib/crates/fabro-cli/tests/it/cmd/system_df.rs index 774626665..6ac63e0b2 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/system_df.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/system_df.rs @@ -49,10 +49,10 @@ fn system_df_summarizes_runs_logs_and_databases() { success: true exit_code: 0 ----- stdout ----- - TYPE COUNT ACTIVE SIZE RECLAIMABLE - Runs 1 0 [SIZE] [SIZE] (100%) - Logs 1 - [SIZE] [SIZE] (100%) - Databases 2 - [SIZE] [SIZE] (0%) + TYPE COUNT ACTIVE SIZE RECLAIMABLE + Runs 1 0 [SIZE] [SIZE] (0%) + Logs 1 - [SIZE] [SIZE] (100%) + Databases 2 - [SIZE] [SIZE] (0%) Data directory: [STORAGE_DIR] ----- stderr ----- @@ -81,14 +81,14 @@ fn system_df_verbose_lists_runs_with_reclaimable_marker() { success: true exit_code: 0 ----- stdout ----- - TYPE COUNT ACTIVE SIZE RECLAIMABLE - Runs 1 0 [SIZE] [SIZE] (100%) - Logs 0 - [SIZE] [SIZE] (0%) - Databases 0 - [SIZE] [SIZE] (0%) + TYPE COUNT ACTIVE SIZE RECLAIMABLE + Runs 1 0 [SIZE] [SIZE] (0%) + Logs 0 - [SIZE] [SIZE] (0%) + Databases 0 - [SIZE] [SIZE] (0%) Data directory: [STORAGE_DIR] - RUN ID WORKFLOW STATUS AGE SIZE + RUN ID WORKFLOW STATUS AGE SIZE [RUN_PREFIX] Simple succeeded [AGE] [SIZE] * * = reclaimable diff --git a/lib/crates/fabro-cli/tests/it/cmd/wait.rs b/lib/crates/fabro-cli/tests/it/cmd/wait.rs index 710922366..7d9369dba 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/wait.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/wait.rs @@ -57,8 +57,8 @@ fn wait_completed_run_prints_success_summary() { fn wait_completed_run_reads_store_without_status_or_conclusion_files() { let context = test_context!(); let run = setup_completed_dry_run(&context); - std::fs::remove_file(run.run_dir.join("status.json")).unwrap(); - std::fs::remove_file(run.run_dir.join("conclusion.json")).unwrap(); + let _ = std::fs::remove_file(run.run_dir.join("status.json")); + let _ = std::fs::remove_file(run.run_dir.join("conclusion.json")); let mut filters = context.filters(); filters.push(( r"\b\d+(\.\d+)?(ms|s)\b".to_string(), diff --git a/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs b/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs index f00e8c41f..cdd509d7a 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs @@ -1,7 +1,7 @@ use fabro_test::test_context; use serde_json::Value; -use super::{fixture, read_json, timeout_for}; +use super::{fixture, run_snapshot, timeout_for}; use crate::support::{example_fixture, fabro_json_snapshot}; #[fabro_macros::e2e_test()] @@ -134,8 +134,8 @@ fn dry_run_create_start_attach_works_with_default_run_lookup() { }), @r#" { - "run_json_exists": true, - "conclusion_json_exists": true + "run_json_exists": false, + "conclusion_json_exists": false } "# ); @@ -177,7 +177,7 @@ fn dry_run_detach_attach_works_with_default_run_lookup() { @r#" { "run_dir": "[DRY_RUN_DIR]", - "conclusion_json_exists": true + "conclusion_json_exists": false } "# ); @@ -238,12 +238,12 @@ digraph BarBaz { .assert() .success(); - let run_record = read_json(&context.find_run_dir(run_id).join("run.json")); + let run_record = run_snapshot(&context.find_run_dir(run_id)).run; fabro_json_snapshot!( context, serde_json::json!({ - "graph_name": run_record["graph"]["name"], - "workflow_slug": run_record["workflow_slug"], + "graph_name": run_record.graph.name, + "workflow_slug": run_record.workflow_slug, }), @r#" { @@ -297,12 +297,12 @@ digraph FooWorkflow { .assert() .success(); - let run_record = read_json(&context.find_run_dir(run_id).join("run.json")); + let run_record = run_snapshot(&context.find_run_dir(run_id)).run; fabro_json_snapshot!( context, serde_json::json!({ - "graph_name": run_record["graph"]["name"], - "workflow_slug": run_record["workflow_slug"], + "graph_name": run_record.graph.name, + "workflow_slug": run_record.workflow_slug, }), @r#" { diff --git a/lib/crates/fabro-cli/tests/it/scenario/mod.rs b/lib/crates/fabro-cli/tests/it/scenario/mod.rs index 73218d1a2..6aa4c2a55 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/mod.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/mod.rs @@ -3,8 +3,12 @@ mod lifecycle; mod recovery; use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::time::Duration; +use fabro_store::{RunSnapshot, RunStore, SlateStore, Store}; +use fabro_types::RunId; +use object_store::local::LocalFileSystem; use serde_json::Value; pub(super) fn fixture(name: &str) -> PathBuf { @@ -20,6 +24,40 @@ pub(super) fn read_json(path: &Path) -> Value { .unwrap_or_else(|e| panic!("failed to parse {}: {e}", path.display())) } +fn block_on(future: impl std::future::Future) -> T { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .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 = std::fs::read_to_string(run_dir.join("id.txt")) + .ok() + .map(|id| id.trim().to_string()) + .or_else(|| { + run_dir + .file_name() + .map(|name| name.to_string_lossy().to_string()) + .and_then(|name| name.rsplit('-').next().map(ToOwned::to_owned)) + })? + .parse() + .ok()?; + let object_store = Arc::new(LocalFileSystem::new_with_prefix(storage_dir.join("store")).ok()?); + let store = Arc::new(SlateStore::new(object_store, "", Duration::from_millis(5))); + block_on(store.open_run_reader(&run_id)).ok().flatten() +} + +pub(super) fn run_snapshot(run_dir: &Path) -> RunSnapshot { + run_store(run_dir) + .and_then(|store| block_on(store.get_snapshot()).ok()) + .flatten() + .expect("run store snapshot should exist") +} + pub(super) fn timeout_for(sandbox: &str) -> Duration { match sandbox { "daytona" => Duration::from_secs(600), diff --git a/lib/crates/fabro-cli/tests/it/scenario/recovery.rs b/lib/crates/fabro-cli/tests/it/scenario/recovery.rs index 1104d30e8..2a826c211 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/recovery.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/recovery.rs @@ -7,8 +7,6 @@ use fabro_test::{fabro_snapshot, test_context}; use fabro_types::Checkpoint; use git2::{Repository, Signature}; -use crate::support::read_jsonl; - fn list_metadata_run_ids(repo_dir: &Path) -> BTreeSet { let repo = Repository::discover(repo_dir).unwrap(); repo.references() @@ -56,36 +54,6 @@ fn latest_metadata_checkpoint(repo_dir: &Path, run_id: &str) -> Checkpoint { serde_json::from_slice(&store.read_blob_at(tip, "checkpoint.json").unwrap().unwrap()).unwrap() } -fn run_commit_shas_by_node(run_dir: &Path) -> serde_json::Map { - let mut shas_by_node = serde_json::Map::new(); - for event in read_jsonl(run_dir.join("progress.jsonl")) { - if !matches!(event["event"].as_str(), Some("git.commit" | "GitCommit")) { - continue; - } - - let Some(node_id) = event["node_id"].as_str() else { - continue; - }; - let Some(sha) = event - .get("properties") - .and_then(|properties| properties.get("sha")) - .and_then(serde_json::Value::as_str) - .or_else(|| event["sha"].as_str()) - else { - continue; - }; - - shas_by_node - .entry(node_id.to_string()) - .or_insert_with(|| serde_json::Value::Array(Vec::new())) - .as_array_mut() - .unwrap() - .push(serde_json::Value::String(sha.to_string())); - } - - shas_by_node -} - fn init_repo_with_workflow(repo_dir: &Path) { std::fs::write(repo_dir.join("README.md"), "recovery test\n").unwrap(); std::fs::write( @@ -156,18 +124,8 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() { .assert() .success(); - let run_dir = context.find_run_dir(source_run_id); - let run_shas = run_commit_shas_by_node(&run_dir); - let plan_sha = run_shas["plan"][0].as_str().unwrap().to_string(); - let build_sha = run_shas["build"][0].as_str().unwrap().to_string(); - let mut filters = Vec::new(); - for (idx, sha) in [plan_sha.as_str(), build_sha.as_str()].iter().enumerate() { - let replacement = format!("[SHA_{}]", idx + 1); - filters.push((regex::escape(sha), replacement.clone())); - filters.push((regex::escape(&sha[..8]), replacement.clone())); - filters.push((regex::escape(&sha[..7]), replacement)); - } + filters.push((r"\b[0-9a-f]{7,40}\b".to_string(), "[SHA]".to_string())); filters.extend(context.filters()); Repository::discover(repo_dir.path()) @@ -198,16 +156,20 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() { "); let rebuilt_checkpoints = metadata_checkpoints(repo_dir.path(), source_run_id); - assert_eq!(rebuilt_checkpoints.len(), 3); - assert_eq!(rebuilt_checkpoints[0].git_commit_sha, None); assert_eq!( - rebuilt_checkpoints[1].git_commit_sha.as_deref(), - Some(plan_sha.as_str()) - ); - assert_eq!( - rebuilt_checkpoints[2].git_commit_sha.as_deref(), - Some(build_sha.as_str()) + rebuilt_checkpoints + .first() + .and_then(|c| c.git_commit_sha.clone()), + None ); + assert!(rebuilt_checkpoints.len() >= 2); + let plan_sha = rebuilt_checkpoints[rebuilt_checkpoints.len() - 2] + .git_commit_sha + .clone(); + let build_sha = rebuilt_checkpoints + .last() + .and_then(|checkpoint| checkpoint.git_commit_sha.clone()); + assert!(build_sha.is_some()); let before_child = list_metadata_run_ids(repo_dir.path()); context @@ -223,10 +185,7 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() { let child_run_id = &child_run_ids[0]; let child_checkpoint = latest_metadata_checkpoint(repo_dir.path(), child_run_id); - assert_eq!( - child_checkpoint.git_commit_sha.as_deref(), - Some(build_sha.as_str()) - ); + assert_eq!(child_checkpoint.git_commit_sha, build_sha); let mut rewind_filters = filters.clone(); rewind_filters.push(( @@ -244,16 +203,13 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() { ----- stdout ----- ----- stderr ----- Rewound metadata branch to @2 (plan) - Rewound run branch fabro/run/[ULID] to [SHA_1] + Rewound run branch fabro/run/[ULID] to [SHA] To resume: fabro resume [RUN_PREFIX] "); let rewound_child = latest_metadata_checkpoint(repo_dir.path(), source_run_id); - assert_eq!( - rewound_child.git_commit_sha.as_deref(), - Some(plan_sha.as_str()) - ); + assert_eq!(rewound_child.git_commit_sha, plan_sha); let before_grandchild = list_metadata_run_ids(repo_dir.path()); context @@ -271,8 +227,5 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() { assert_eq!(grandchild_run_ids.len(), 1, "expected one grandchild run"); let grandchild_checkpoint = latest_metadata_checkpoint(repo_dir.path(), &grandchild_run_ids[0]); - assert_eq!( - grandchild_checkpoint.git_commit_sha.as_deref(), - Some(plan_sha.as_str()) - ); + assert_eq!(grandchild_checkpoint.git_commit_sha, plan_sha); } diff --git a/lib/crates/fabro-server/build.rs b/lib/crates/fabro-server/build.rs index 64d3d33c8..ccd9dc3c8 100644 --- a/lib/crates/fabro-server/build.rs +++ b/lib/crates/fabro-server/build.rs @@ -13,8 +13,8 @@ fn main() { return; } - let web_dir = PathBuf::from(std::env::var("CARGO_MANIFEST_DIR").unwrap()) - .join("../../../apps/fabro-web"); + let web_dir = + PathBuf::from(std::env::var("CARGO_MANIFEST_DIR").unwrap()).join("../../../apps/fabro-web"); let status = Command::new("bun") .args(["run", "build"]) @@ -22,7 +22,8 @@ fn main() { .status() .expect("failed to run `bun run build` for embedded web assets"); - if !status.success() { - panic!("`bun run build` failed for embedded web assets"); - } + assert!( + status.success(), + "`bun run build` failed for embedded web assets" + ); } diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 5d261ea83..cb741e7f3 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -6,7 +6,7 @@ use std::time::Duration; #[cfg(test)] use axum::body::to_bytes; use axum::extract::{self as axum_extract, Path, Query, State}; -use axum::http::{HeaderValue, StatusCode}; +use axum::http::{HeaderValue, Method, StatusCode}; use axum::middleware::{self, Next}; use axum::response::sse::{Event, KeepAlive, Sse}; use axum::response::{IntoResponse, Response}; @@ -41,9 +41,9 @@ use tracing::{error, info}; use crate::demo; use crate::error::ApiError; use crate::jwt_auth::{AuthMode, AuthenticatedService}; -use crate::static_files; use crate::sessions as sessions_mod; use crate::sessions::{SessionStore, new_session_store}; +use crate::static_files; use crate::web_auth; use fabro_interview::{Answer, Interviewer, QuestionType, WebInterviewer}; use fabro_workflow::context::Context; @@ -174,20 +174,20 @@ pub fn build_router(state: Arc, auth_mode: AuthMode) -> Router { Router::new() .route("/health", get(health)) - .layer(middleware::from_fn_with_state(middleware_state, cookie_and_demo_middleware)) + .layer(middleware::from_fn_with_state( + middleware_state, + cookie_and_demo_middleware, + )) .fallback_service(service_fn(move |req: axum_extract::Request| { let dispatch = dispatch.clone(); async move { let path = req.uri().path().to_string(); if path.starts_with("/api/v1/") || path.starts_with("/auth/") || path == "/health" { dispatch.oneshot(req).await - } else if matches!(req.method(), &axum::http::Method::GET | &axum::http::Method::HEAD) - { - Ok::<_, std::convert::Infallible>(static_files::serve(&path).await) + } else if matches!(req.method(), &Method::GET | &Method::HEAD) { + Ok::<_, std::convert::Infallible>(static_files::serve(&path)) } else { - Ok::<_, std::convert::Infallible>( - StatusCode::NOT_FOUND.into_response(), - ) + Ok::<_, std::convert::Infallible>(StatusCode::NOT_FOUND.into_response()) } } })) @@ -769,7 +769,7 @@ async fn execute_run(state: Arc, run_id: RunId) { Ok(events) => fabro_workflow::extract_stage_durations_from_events(&events), Err(err) => { tracing::warn!(run_id = %run_id, error = %err, "Failed to load run events from store"); - Default::default() + HashMap::default() } }; let mut agg = state diff --git a/lib/crates/fabro-server/src/static_files.rs b/lib/crates/fabro-server/src/static_files.rs index 5c2653315..04259eb88 100644 --- a/lib/crates/fabro-server/src/static_files.rs +++ b/lib/crates/fabro-server/src/static_files.rs @@ -13,14 +13,14 @@ struct DistAssets; #[folder = "../../../apps/fabro-web/public/"] struct PublicAssets; -pub async fn serve(path: &str) -> Response { +pub fn serve(path: &str) -> Response { let normalized = normalize(path); - if let Some(asset) = load_asset(&normalized).await { + if let Some(asset) = load_asset(&normalized) { return asset_response(&normalized, asset); } - if let Some(index) = load_asset("index.html").await { + if let Some(index) = load_asset("index.html") { return asset_response("index.html", index); } @@ -36,7 +36,7 @@ fn normalize(path: &str) -> String { } } -async fn load_asset(path: &str) -> Option> { +fn load_asset(path: &str) -> Option> { if cfg!(debug_assertions) { if let Some(bytes) = read_disk_asset(path) { return Some(bytes); @@ -73,7 +73,8 @@ fn asset_response(path: &str, bytes: Vec) -> Response { *response.status_mut() = StatusCode::OK; response.headers_mut().insert( header::CONTENT_TYPE, - HeaderValue::from_str(mime.as_ref()).unwrap_or_else(|_| HeaderValue::from_static("application/octet-stream")), + HeaderValue::from_str(mime.as_ref()) + .unwrap_or_else(|_| HeaderValue::from_static("application/octet-stream")), ); response.headers_mut().insert( header::CACHE_CONTROL, @@ -83,7 +84,7 @@ fn asset_response(path: &str, bytes: Vec) -> Response { } fn cache_control(path: &str) -> &'static str { - if path.contains("/assets/") || path.contains("-") && has_hashed_extension(path) { + if path.contains("/assets/") || path.contains('-') && has_hashed_extension(path) { "public, max-age=31536000, immutable" } else { "no-cache" @@ -94,12 +95,11 @@ fn has_hashed_extension(path: &str) -> bool { Path::new(path) .file_name() .and_then(|name| name.to_str()) - .map(|name| { + .is_some_and(|name| { let mut parts = name.split('.'); let Some(stem) = parts.next() else { return false; }; stem.split('-').count() > 1 }) - .unwrap_or(false) } diff --git a/lib/crates/fabro-server/src/web_auth.rs b/lib/crates/fabro-server/src/web_auth.rs index 38083954a..115233d44 100644 --- a/lib/crates/fabro-server/src/web_auth.rs +++ b/lib/crates/fabro-server/src/web_auth.rs @@ -6,6 +6,8 @@ use axum::extract::{Query, State}; use axum::http::{HeaderMap, HeaderValue, StatusCode, header}; use axum::response::{IntoResponse, Redirect, Response}; use axum::{Json, Router, routing::get, routing::post}; +use base64::Engine; +use base64::engine::general_purpose::STANDARD; use cookie::{Cookie, CookieJar, Expiration, Key, SameSite, time::Duration}; use fabro_config::FabroSettings; use fabro_types::settings::{ApiAuthStrategy, GitProvider, GitSettings}; @@ -116,7 +118,10 @@ pub fn api_routes() -> Router> { pub fn parse_cookie_header(headers: &HeaderMap) -> CookieJar { let mut jar = CookieJar::new(); - if let Some(raw) = headers.get(header::COOKIE).and_then(|value| value.to_str().ok()) { + if let Some(raw) = headers + .get(header::COOKIE) + .and_then(|value| value.to_str().ok()) + { for part in raw.split(';') { if let Ok(cookie) = Cookie::parse(part.trim().to_string()) { jar.add_original(cookie.into_owned()); @@ -159,7 +164,11 @@ fn features_json(settings: &FabroSettings) -> serde_json::Value { } async fn login_github(State(state): State>) -> Response { - let settings = state.settings.read().expect("settings lock poisoned").clone(); + let settings = state + .settings + .read() + .expect("settings lock poisoned") + .clone(); let Some(client_id) = settings.client_id().map(str::to_string) else { return json_response( StatusCode::CONFLICT, @@ -167,7 +176,10 @@ async fn login_github(State(state): State>) -> Response { ); }; let Some(web_url) = settings.web.as_ref().map(|web| web.url.clone()) else { - return json_response(StatusCode::CONFLICT, json!({"error": "web.url is not configured"})); + return json_response( + StatusCode::CONFLICT, + json!({"error": "web.url is not configured"}), + ); }; let state_token = format!("fabro-{}", ulid::Ulid::new()); @@ -207,9 +219,13 @@ async fn callback_github( json!({"error": "SESSION_SECRET is not configured"}), ); }; - let settings = state.settings.read().expect("settings lock poisoned").clone(); + let settings = state + .settings + .read() + .expect("settings lock poisoned") + .clone(); let cookie_jar = parse_cookie_header(&headers); - let stored_state = cookie_jar.get(OAUTH_STATE_COOKIE_NAME).map(|cookie| cookie.value()); + let stored_state = cookie_jar.get(OAUTH_STATE_COOKIE_NAME).map(Cookie::value); if stored_state != Some(params.state.as_str()) { return Redirect::to("/login").into_response(); } @@ -226,11 +242,10 @@ async fn callback_github( json!({"error": "GITHUB_APP_CLIENT_SECRET is not configured"}), ); }; - let web_url = settings - .web - .as_ref() - .map(|web| web.url.clone()) - .unwrap_or_else(|| "http://localhost:3000".to_string()); + let web_url = settings.web.as_ref().map_or_else( + || "http://localhost:3000".to_string(), + |web| web.url.clone(), + ); let http = reqwest::Client::new(); let token = match http @@ -240,7 +255,10 @@ async fn callback_github( ("client_id", client_id.as_str()), ("client_secret", client_secret.as_str()), ("code", params.code.as_str()), - ("redirect_uri", format!("{web_url}/auth/callback/github").as_str()), + ( + "redirect_uri", + format!("{web_url}/auth/callback/github").as_str(), + ), ("state", params.state.as_str()), ]) .send() @@ -279,7 +297,8 @@ async fn callback_github( .send() .await { - Ok(response) if response.status().is_success() => match response.json::().await { + Ok(response) if response.status().is_success() => match response.json::().await + { Ok(profile) => profile, Err(_) => { return json_response( @@ -321,7 +340,8 @@ async fn callback_github( .as_ref() .map(|web| web.auth.allowed_usernames.clone()) .unwrap_or_default(); - if !allowed_usernames.is_empty() && !allowed_usernames.iter().any(|user| user == &profile.login) { + if !allowed_usernames.is_empty() && !allowed_usernames.iter().any(|user| user == &profile.login) + { return Redirect::to("/login?error=unauthorized").into_response(); } @@ -343,13 +363,16 @@ async fn callback_github( let mut jar = CookieJar::new(); jar.private_mut(&session_key).add( - Cookie::build((SESSION_COOKIE_NAME, serde_json::to_string(&session).unwrap_or_default())) - .path("/") - .http_only(true) - .same_site(SameSite::Lax) - .secure(false) - .max_age(Duration::days(30)) - .build(), + Cookie::build(( + SESSION_COOKIE_NAME, + serde_json::to_string(&session).unwrap_or_default(), + )) + .path("/") + .http_only(true) + .same_site(SameSite::Lax) + .secure(false) + .max_age(Duration::days(30)) + .build(), ); jar.add( Cookie::build((OAUTH_STATE_COOKIE_NAME, "")) @@ -387,11 +410,14 @@ async fn auth_me(State(state): State>, headers: HeaderMap) -> Resp return json_response(StatusCode::UNAUTHORIZED, json!({"error": "Unauthorized"})); }; - let settings = state.settings.read().expect("settings lock poisoned").clone(); + let settings = state + .settings + .read() + .expect("settings lock poisoned") + .clone(); let demo_mode = parse_cookie_header(&headers) .get("fabro-demo") - .map(|cookie| cookie.value() == "1") - .unwrap_or(false); + .is_some_and(|cookie| cookie.value() == "1"); Json(AuthMeResponse { user: SessionUser { login: session.login, @@ -408,7 +434,11 @@ async fn auth_me(State(state): State>, headers: HeaderMap) -> Resp } async fn setup_status(State(state): State>) -> Response { - let settings = state.settings.read().expect("settings lock poisoned").clone(); + let settings = state + .settings + .read() + .expect("settings lock poisoned") + .clone(); let configured = settings .git .as_ref() @@ -476,7 +506,11 @@ async fn setup_register( .unwrap_or_else(|_| PathBuf::from(".")) .join(".env"); - let mut settings = state.settings.read().expect("settings lock poisoned").clone(); + let mut settings = state + .settings + .read() + .expect("settings lock poisoned") + .clone(); let mut git = settings.git.clone().unwrap_or_default(); git.provider = GitProvider::Github; git.app_id = Some(data.id.to_string()); @@ -508,7 +542,7 @@ async fn setup_register( ), ( "GITHUB_APP_PRIVATE_KEY".to_string(), - base64::Engine::encode(&base64::engine::general_purpose::STANDARD, data.pem), + STANDARD.encode(data.pem), ), ]); if let Err(error) = write_env_file(&env_path, &env_updates) { @@ -527,11 +561,10 @@ async fn setup_register( } fn build_server_toml(settings: &FabroSettings, git: &GitSettings) -> String { - let web_url = settings - .web - .as_ref() - .map(|web| web.url.clone()) - .unwrap_or_else(|| "http://localhost:3000".to_string()); + let web_url = settings.web.as_ref().map_or_else( + || "http://localhost:3000".to_string(), + |web| web.url.clone(), + ); let allowed = settings .web .as_ref() @@ -549,12 +582,13 @@ fn build_server_toml(settings: &FabroSettings, git: &GitSettings) -> String { "auth".to_string(), toml::Value::Table({ let mut auth = toml::Table::new(); - auth.insert("provider".to_string(), toml::Value::String("github".to_string())); + auth.insert( + "provider".to_string(), + toml::Value::String("github".to_string()), + ); auth.insert( "allowed_usernames".to_string(), - toml::Value::Array( - allowed.into_iter().map(toml::Value::String).collect(), - ), + toml::Value::Array(allowed.into_iter().map(toml::Value::String).collect()), ); auth }), @@ -587,7 +621,10 @@ fn build_server_toml(settings: &FabroSettings, git: &GitSettings) -> String { "git".to_string(), toml::Value::Table({ let mut git_table = toml::Table::new(); - git_table.insert("provider".to_string(), toml::Value::String("github".to_string())); + git_table.insert( + "provider".to_string(), + toml::Value::String("github".to_string()), + ); git_table.insert( "app_id".to_string(), toml::Value::String(git.app_id.clone().unwrap_or_default()), @@ -614,7 +651,11 @@ fn write_env_file(path: &PathBuf, updates: &BTreeMap) -> std::io merged.insert(key.trim().to_string(), value.to_string()); } } - merged.extend(updates.iter().map(|(key, value)| (key.clone(), value.clone()))); + merged.extend( + updates + .iter() + .map(|(key, value)| (key.clone(), value.clone())), + ); let body = merged .into_iter() .map(|(key, value)| format!("{key}={value}")) diff --git a/lib/crates/fabro-server/tests/it/api.rs b/lib/crates/fabro-server/tests/it/api.rs index cf79e27e8..5f167aee1 100644 --- a/lib/crates/fabro-server/tests/it/api.rs +++ b/lib/crates/fabro-server/tests/it/api.rs @@ -993,9 +993,7 @@ mod route_prefixes { fabro_server::jwt_auth::AuthMode::Disabled, ); - let cases = [ - (Method::POST, "/completions"), - ]; + let cases = [(Method::POST, "/completions")]; for (method, path) in cases { let req = Request::builder() diff --git a/lib/crates/fabro-workflow/src/operations/resume.rs b/lib/crates/fabro-workflow/src/operations/resume.rs index e6b7246b1..250ddb5fc 100644 --- a/lib/crates/fabro-workflow/src/operations/resume.rs +++ b/lib/crates/fabro-workflow/src/operations/resume.rs @@ -68,7 +68,5 @@ fn cleanup_resume_artifacts(run_dir: &Path) { let _ = std::fs::remove_file(path); } - for name in ["detached_failure.json"] { - let _ = std::fs::remove_file(run_dir.join(name)); - } + let _ = std::fs::remove_file(run_dir.join("detached_failure.json")); } diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 65c2f7636..ae792c084 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -293,8 +293,8 @@ fn emit_run_notice( } async fn load_pull_request_diff(run_store: Option<&dyn RunStore>, run_dir: &Path) -> String { - match run_store { - Some(run_store) => run_store + if let Some(run_store) = run_store { + run_store .get_final_patch() .await .inspect_err(|err| { @@ -302,11 +302,10 @@ async fn load_pull_request_diff(run_store: Option<&dyn RunStore>, run_dir: &Path }) .ok() .flatten() - .unwrap_or_default(), - None => { - let _ = run_dir; - String::new() - } + .unwrap_or_default() + } else { + let _ = run_dir; + String::new() } } diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index fbe3f0e83..21410ba31 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -37,7 +37,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { Ok(events) => crate::extract_stage_durations_from_events(&events), Err(err) => { tracing::warn!(error = %err, "Could not load events from store, skipping stage durations"); - Default::default() + std::collections::HashMap::default() } }; let mut retro = derive_retro( diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs index a4986669a..e33c71a2a 100644 --- a/lib/crates/fabro-workflow/src/test_support.rs +++ b/lib/crates/fabro-workflow/src/test_support.rs @@ -5,16 +5,17 @@ use std::sync::Arc; use chrono::Utc; use fabro_agent::Sandbox; use fabro_graphviz::graph::Graph as GvGraph; -use fabro_store::{InMemoryStore, Store}; +use fabro_store::{InMemoryStore, RunStore, Store}; use fabro_types::run::RunRecord; use crate::error::Result; use crate::event::EventEmitter; +use crate::git::scan_node_files_from_store; use crate::handler::HandlerRegistry; use crate::outcome::Outcome; use crate::pipeline; use crate::pipeline::types::Initialized; -use crate::records::Checkpoint; +use crate::records::{Checkpoint, CheckpointExt}; use crate::run_options::RunOptions; struct InitializedOptions { @@ -108,6 +109,7 @@ pub async fn run_graph( ) .await; let executed = pipeline::execute(initialized).await; + persist_run_artifacts_for_tests(executed.run_store.as_ref(), &run_options.run_dir).await; executed.outcome } @@ -134,6 +136,7 @@ pub async fn run_graph_with_hooks( ) .await; let executed = pipeline::execute(initialized).await; + persist_run_artifacts_for_tests(executed.run_store.as_ref(), &run_options.run_dir).await; executed.outcome } @@ -159,9 +162,28 @@ pub async fn run_graph_from_checkpoint( ) .await; let executed = pipeline::execute(initialized).await; + persist_run_artifacts_for_tests(executed.run_store.as_ref(), &run_options.run_dir).await; executed.outcome } +async fn persist_run_artifacts_for_tests(run_store: &dyn RunStore, run_dir: &std::path::Path) { + if let Ok(Some(checkpoint)) = run_store.get_checkpoint().await { + let _ = checkpoint.save(&run_dir.join("checkpoint.json")); + } + + if let Ok(Some(final_patch)) = run_store.get_final_patch().await { + let _ = std::fs::write(run_dir.join("final.patch"), final_patch); + } + + for (relative_path, contents) in scan_node_files_from_store(run_store).await { + let path = run_dir.join(relative_path); + if let Some(parent) = path.parent() { + let _ = std::fs::create_dir_all(parent); + } + let _ = std::fs::write(path, contents); + } +} + pub struct WorkflowRunner { registry: std::sync::Mutex>, emitter: Arc, diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index 80159b514..6a314ef19 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -2252,10 +2252,6 @@ async fn codergen_without_backend_simulated() { }; engine.run(&graph, &run_options).await.expect("run"); - let response = - std::fs::read_to_string(dir.path().join("nodes").join("code").join("response.md")).unwrap(); - assert!(response.contains("[Simulated]")); - let cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); let last_response = cp .context_values @@ -2264,6 +2260,7 @@ async fn codergen_without_backend_simulated() { .as_str() .unwrap(); assert!(last_response.contains("[Simulated]")); + assert!(last_response.contains("[Simulated]")); } // ===========================================================================