From 2b6ba07f36ebca924659aaa2430d63cd0c98343b Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 5 Apr 2026 03:41:10 -0400 Subject: [PATCH] fix(cli): restore green server-backed test checkpoint --- .../fabro-cli/src/commands/pr/create.rs | 5 +- lib/crates/fabro-cli/src/commands/pr/list.rs | 2 +- lib/crates/fabro-cli/src/commands/pr/mod.rs | 2 +- .../fabro-cli/src/commands/run/attach.rs | 110 ++++++------ .../fabro-cli/src/commands/run/detached.rs | 3 +- lib/crates/fabro-cli/src/commands/run/diff.rs | 9 +- lib/crates/fabro-cli/src/commands/run/fork.rs | 4 +- .../fabro-cli/src/commands/run/launcher.rs | 7 +- lib/crates/fabro-cli/src/commands/run/logs.rs | 22 ++- .../fabro-cli/src/commands/run/output.rs | 23 +-- .../fabro-cli/src/commands/run/preview.rs | 2 +- .../fabro-cli/src/commands/run/rewind.rs | 22 +-- lib/crates/fabro-cli/src/commands/run/wait.rs | 2 +- .../fabro-cli/src/commands/runs/inspect.rs | 11 +- .../fabro-cli/src/commands/runs/list.rs | 2 +- lib/crates/fabro-cli/src/commands/runs/rm.rs | 24 ++- .../fabro-cli/src/commands/store/dump.rs | 4 +- .../fabro-cli/src/commands/system/df.rs | 5 +- lib/crates/fabro-cli/src/main.rs | 3 +- lib/crates/fabro-cli/src/server_client.rs | 15 +- lib/crates/fabro-cli/src/store.rs | 26 --- lib/crates/fabro-cli/tests/it/cmd/logs.rs | 4 +- lib/crates/fabro-cli/tests/it/cmd/ps.rs | 12 +- lib/crates/fabro-cli/tests/it/cmd/rm.rs | 4 +- lib/crates/fabro-cli/tests/it/cmd/run.rs | 6 +- .../fabro-cli/tests/it/cmd/server_start.rs | 6 +- lib/crates/fabro-cli/tests/it/cmd/start.rs | 12 +- lib/crates/fabro-cli/tests/it/cmd/support.rs | 31 +++- .../fabro-cli/tests/it/cmd/system_prune.rs | 2 +- .../fabro-cli/tests/it/scenario/lifecycle.rs | 6 +- lib/crates/fabro-cli/tests/it/support/mod.rs | 2 +- .../tests/it/workflow/dry_run_examples.rs | 10 +- lib/crates/fabro-server/src/jwt_auth.rs | 4 +- lib/crates/fabro-server/src/server.rs | 157 +++++++++++------- lib/crates/fabro-store/src/keys.rs | 11 +- lib/crates/fabro-store/src/slate/catalog.rs | 5 +- lib/crates/fabro-store/src/slate/mod.rs | 11 +- lib/crates/fabro-store/src/slate/run_store.rs | 44 +++-- lib/crates/fabro-test/src/lib.rs | 18 +- lib/crates/fabro-workflow/src/run_lookup.rs | 2 +- 40 files changed, 362 insertions(+), 288 deletions(-) delete mode 100644 lib/crates/fabro-cli/src/store.rs diff --git a/lib/crates/fabro-cli/src/commands/pr/create.rs b/lib/crates/fabro-cli/src/commands/pr/create.rs index 407190d16..f81bddfc9 100644 --- a/lib/crates/fabro-cli/src/commands/pr/create.rs +++ b/lib/crates/fabro-cli/src/commands/pr/create.rs @@ -4,13 +4,14 @@ use anyhow::{Context, Result, bail}; use fabro_model::Catalog; use fabro_sandbox::daytona::detect_repo_info; use fabro_workflow::outcome::StageStatus; +use fabro_workflow::pull_request::maybe_open_pull_request; use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use tracing::info; use crate::args::{GlobalArgs, PrCreateArgs}; use crate::commands::store::rebuild::rebuild_run_store; -use crate::shared::print_json_pretty; use crate::server_client; +use crate::shared::print_json_pretty; use crate::user_config::load_user_settings_with_globals; pub(super) async fn create_command( @@ -104,7 +105,7 @@ async fn create_from( .model .unwrap_or_else(|| Catalog::builtin().default_from_env().id.clone()); - let record = fabro_workflow::pull_request::maybe_open_pull_request( + let record = maybe_open_pull_request( &creds, &origin_url, base_branch, diff --git a/lib/crates/fabro-cli/src/commands/pr/list.rs b/lib/crates/fabro-cli/src/commands/pr/list.rs index 9eeced37d..6e2e7fc8f 100644 --- a/lib/crates/fabro-cli/src/commands/pr/list.rs +++ b/lib/crates/fabro-cli/src/commands/pr/list.rs @@ -8,8 +8,8 @@ use serde::Serialize; use tracing::info; use crate::args::{GlobalArgs, PrListArgs}; -use crate::shared::print_json_pretty; use crate::server_client; +use crate::shared::print_json_pretty; use crate::user_config::load_user_settings_with_globals; #[derive(Serialize)] diff --git a/lib/crates/fabro-cli/src/commands/pr/mod.rs b/lib/crates/fabro-cli/src/commands/pr/mod.rs index 0f83ae2a3..3bbff5516 100644 --- a/lib/crates/fabro-cli/src/commands/pr/mod.rs +++ b/lib/crates/fabro-cli/src/commands/pr/mod.rs @@ -12,8 +12,8 @@ use fabro_types::PullRequestRecord; use fabro_workflow::run_lookup::resolve_run_from_summaries; use crate::args::{GlobalArgs, PrCommand, PrNamespace}; -use crate::shared::github::build_github_app_credentials; use crate::server_client; +use crate::shared::github::build_github_app_credentials; use crate::user_config::load_user_settings_with_globals; pub(crate) async fn dispatch(ns: PrNamespace, globals: &GlobalArgs) -> Result<()> { diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index 0cc28ce18..7f3849729 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -10,7 +10,7 @@ use fabro_types::{EventBody, RunEvent, RunId}; use fabro_api::types; use fabro_interview::{AnswerValue, ConsoleInterviewer, Question, QuestionOption, QuestionType}; -use fabro_store::{EventEnvelope, RuntimeState}; +use fabro_store::EventEnvelope; use fabro_util::json::normalize_json_value; use fabro_util::terminal::Styles; use fabro_workflow::outcome::StageStatus; @@ -97,8 +97,6 @@ async fn attach_run_server( engine_child: Option, json_output: bool, ) -> Result { - let runtime_state = RuntimeState::new(run_dir); - let mut engine_guard = engine_child.map(EngineChildGuard::new); let is_tty = std::io::stderr().is_terminal(); @@ -145,12 +143,17 @@ async fn attach_run_server( } // Wait briefly for a terminal status or conclusion for _ in 0..20 { - if client.get_run_state(run_id).await.ok().is_some_and(|state| { - state.conclusion.is_some() - || state - .status - .is_some_and(|record| record.status.is_terminal()) - }) { + if client + .get_run_state(run_id) + .await + .ok() + .is_some_and(|state| { + state.conclusion.is_some() + || state + .status + .is_some_and(|record| record.status.is_terminal()) + }) + { break; } sleep(Duration::from_millis(100)).await; @@ -201,9 +204,11 @@ async fn attach_run_server( hide_progress(&mut progress_ui, json_output); let interviewer = ConsoleInterviewer::new(styles); - let answer = - fabro_interview::Interviewer::ask(&interviewer, api_question_to_question(&question)) - .await; + let answer = fabro_interview::Interviewer::ask( + &interviewer, + api_question_to_question(&question), + ) + .await; show_progress(&mut progress_ui, json_output); if answer_requires_reattach(&answer) { @@ -234,14 +239,26 @@ async fn attach_run_server( if let Some(child_alive) = child_alive_via_handle { if !child_alive && !saw_event { - flush_remaining_server_events(client, run_id, next_seq, &mut progress_ui, json_output) - .await?; + flush_remaining_server_events( + client, + run_id, + next_seq, + &mut progress_ui, + json_output, + ) + .await?; break; } } else { if terminal_status.is_some() && !saw_event { - flush_remaining_server_events(client, run_id, next_seq, &mut progress_ui, json_output) - .await?; + flush_remaining_server_events( + client, + run_id, + next_seq, + &mut progress_ui, + json_output, + ) + .await?; break; } @@ -261,8 +278,14 @@ async fn attach_run_server( } }; if !engine_alive { - flush_remaining_server_events(client, run_id, next_seq, &mut progress_ui, json_output) - .await?; + flush_remaining_server_events( + client, + run_id, + next_seq, + &mut progress_ui, + json_output, + ) + .await?; break; } } @@ -318,7 +341,13 @@ async fn submit_server_interview_answer( } }; client - .submit_run_answer(run_id, qid, value, selected_option_key, selected_option_keys) + .submit_run_answer( + run_id, + qid, + value, + selected_option_key, + selected_option_keys, + ) .await?; Ok(true) } @@ -358,7 +387,8 @@ async fn flush_remaining_server_events( } fn is_run_not_found_error(err: &anyhow::Error) -> bool { - err.chain().any(|cause| cause.to_string() == "Run not found.") + err.chain() + .any(|cause| cause.to_string() == "Run not found.") } fn emit_progress_line( @@ -407,8 +437,7 @@ fn restore_empty_run_properties(value: &mut serde_json::Value) { let Some(event_name) = object.get("event").and_then(serde_json::Value::as_str) else { return; }; - if matches!(event_name, "run.submitted" | "run.running") && !object.contains_key("properties") - { + if matches!(event_name, "run.submitted" | "run.running") && !object.contains_key("properties") { let run_id = object.remove("run_id"); let ts = object.remove("ts"); object.insert("properties".to_string(), serde_json::json!({})); @@ -443,28 +472,12 @@ fn infer_run_id(run_dir: &Path) -> Option { }) } -#[allow(clippy::struct_field_names)] -#[derive(Debug, Clone, PartialEq, Eq)] -struct InterviewPaths { - claim_path: PathBuf, - request_path: PathBuf, - response_path: PathBuf, -} - -impl InterviewPaths { - fn from_runtime_state(runtime_state: &RuntimeState) -> Self { - Self { - claim_path: runtime_state.interview_claim_path(), - request_path: runtime_state.interview_request_path(), - response_path: runtime_state.interview_response_path(), - } - } -} - +#[cfg(test)] struct InterviewClaimGuard { claim_path: PathBuf, } +#[cfg(test)] impl InterviewClaimGuard { fn acquire(claim_path: &Path) -> Option { if try_claim_interview_request(claim_path) { @@ -477,6 +490,7 @@ impl InterviewClaimGuard { } } +#[cfg(test)] impl Drop for InterviewClaimGuard { fn drop(&mut self) { let _ = std::fs::remove_file(&self.claim_path); @@ -516,6 +530,7 @@ impl Drop for EngineChildGuard { } } +#[cfg(test)] fn try_claim_interview_request(claim_path: &Path) -> bool { if let Some(parent) = claim_path.parent() { if std::fs::create_dir_all(parent).is_err() { @@ -549,6 +564,7 @@ fn answer_requires_reattach(answer: &fabro_interview::Answer) -> bool { matches!(answer.value, AnswerValue::Aborted | AnswerValue::Skipped) } +#[cfg(test)] fn write_interview_response_atomically( response_path: &Path, answer: &fabro_interview::Answer, @@ -611,13 +627,13 @@ fn process_alive(pid: u32) -> bool { fn event_exit_code(event: &EventEnvelope) -> Option { let run_event = RunEvent::try_from(&event.payload).ok()?; match run_event.body { - EventBody::RunCompleted(props) => Some(if props.status == "success" - || props.status == "partial_success" - { - ExitCode::from(0) - } else { - ExitCode::from(1) - }), + EventBody::RunCompleted(props) => Some( + if props.status == "success" || props.status == "partial_success" { + ExitCode::from(0) + } else { + ExitCode::from(1) + }, + ), EventBody::RunFailed(_) => Some(ExitCode::from(1)), _ => None, } diff --git a/lib/crates/fabro-cli/src/commands/run/detached.rs b/lib/crates/fabro-cli/src/commands/run/detached.rs index be0883b87..9306fe076 100644 --- a/lib/crates/fabro-cli/src/commands/run/detached.rs +++ b/lib/crates/fabro-cli/src/commands/run/detached.rs @@ -3,6 +3,7 @@ use std::time::Duration; use anyhow::{Result, anyhow}; use fabro_types::{RunId, RunStatus}; +use tokio::time::sleep; use crate::server_client; use crate::user_config::load_user_settings; @@ -44,7 +45,7 @@ pub(crate) async fn execute( | RunStatus::Running | RunStatus::Paused | RunStatus::Removing => { - tokio::time::sleep(Duration::from_millis(100)).await; + sleep(Duration::from_millis(100)).await; } } } diff --git a/lib/crates/fabro-cli/src/commands/run/diff.rs b/lib/crates/fabro-cli/src/commands/run/diff.rs index 6b5f0c9af..c52b02473 100644 --- a/lib/crates/fabro-cli/src/commands/run/diff.rs +++ b/lib/crates/fabro-cli/src/commands/run/diff.rs @@ -8,8 +8,9 @@ use fabro_workflow::sandbox_git::GIT_REMOTE; use tracing::{debug, info}; use crate::args::{DiffArgs, GlobalArgs}; -use crate::shared::print_json_pretty; use crate::server_client; +use crate::server_client::RunProjection; +use crate::shared::print_json_pretty; use crate::user_config::load_user_settings_with_globals; pub(crate) async fn run(args: DiffArgs, globals: &GlobalArgs) -> Result<()> { @@ -52,11 +53,7 @@ pub(crate) async fn run(args: DiffArgs, globals: &GlobalArgs) -> Result<()> { Ok(()) } -async fn resolve_diff( - _run_dir: &Path, - state: &crate::server_client::RunProjection, - args: &DiffArgs, -) -> Result { +async fn resolve_diff(_run_dir: &Path, state: &RunProjection, args: &DiffArgs) -> Result { if let Some(ref node_id) = args.node { if let Some(visit) = state.list_node_visits(node_id).into_iter().max() { if let Some(node) = state.node(&fabro_store::StageId::new(node_id, visit)) { diff --git a/lib/crates/fabro-cli/src/commands/run/fork.rs b/lib/crates/fabro-cli/src/commands/run/fork.rs index eb2710345..9fd7f4a98 100644 --- a/lib/crates/fabro-cli/src/commands/run/fork.rs +++ b/lib/crates/fabro-cli/src/commands/run/fork.rs @@ -2,9 +2,7 @@ use anyhow::Context; use anyhow::Result; use fabro_checkpoint::git::Store; use fabro_util::terminal::Styles; -use fabro_workflow::operations::{ - ForkRunInput, RewindTarget, build_timeline_or_rebuild, fork, -}; +use fabro_workflow::operations::{ForkRunInput, RewindTarget, build_timeline_or_rebuild, fork}; use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use git2::Repository; diff --git a/lib/crates/fabro-cli/src/commands/run/launcher.rs b/lib/crates/fabro-cli/src/commands/run/launcher.rs index d8fc34606..b40a39be1 100644 --- a/lib/crates/fabro-cli/src/commands/run/launcher.rs +++ b/lib/crates/fabro-cli/src/commands/run/launcher.rs @@ -1,5 +1,6 @@ use std::path::{Path, PathBuf}; +#[cfg(test)] use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; use fabro_types::RunId; @@ -26,10 +27,7 @@ pub(crate) fn launcher_record_path(storage_dir: &Path, run_id: &RunId) -> PathBu launcher_dir(storage_dir).join(format!("{run_id}.json")) } -pub(crate) fn launcher_log_path(storage_dir: &Path, run_id: &RunId) -> PathBuf { - launcher_dir(storage_dir).join(format!("{run_id}.log")) -} - +#[cfg(test)] pub(crate) fn write_launcher_record(path: &Path, record: &LauncherRecord) -> Result<()> { if let Some(parent) = path.parent() { std::fs::create_dir_all(parent)?; @@ -76,6 +74,7 @@ pub(crate) fn launcher_record_for_run(run_dir: &Path) -> Option None } +#[cfg(test)] pub(crate) fn active_launcher_record(storage_dir: &Path, run_id: &RunId) -> Option { let path = launcher_record_path(storage_dir, run_id); let launcher = read_launcher_record(&path)?; diff --git a/lib/crates/fabro-cli/src/commands/run/logs.rs b/lib/crates/fabro-cli/src/commands/run/logs.rs index a510e6cb7..7697e60aa 100644 --- a/lib/crates/fabro-cli/src/commands/run/logs.rs +++ b/lib/crates/fabro-cli/src/commands/run/logs.rs @@ -154,7 +154,12 @@ async fn follow_store_logs( let mut next_seq = seq; loop { - match time::timeout(Duration::from_millis(200), client.list_run_events(run_id, Some(next_seq), None)).await { + match time::timeout( + Duration::from_millis(200), + client.list_run_events(run_id, Some(next_seq), None), + ) + .await + { Ok(Ok(events)) => { let saw_terminal = events .iter() @@ -172,16 +177,20 @@ async fn follow_store_logs( next_seq = event.seq.saturating_add(1); } if saw_terminal { - flush_remaining_store_events(client, run_id, next_seq, pretty, styles, &mut out) - .await?; + flush_remaining_store_events( + client, run_id, next_seq, pretty, styles, &mut out, + ) + .await?; debug!("Observed terminal event while following logs, stopping follow"); break; } } Err(_) => { if run_concluded(client, run_id).await? { - flush_remaining_store_events(client, run_id, next_seq, pretty, styles, &mut out) - .await?; + flush_remaining_store_events( + client, run_id, next_seq, pretty, styles, &mut out, + ) + .await?; debug!("Run reached terminal status, stopping follow"); break; } @@ -248,8 +257,7 @@ fn restore_empty_run_properties(value: &mut serde_json::Value) { let Some(event_name) = object.get("event").and_then(serde_json::Value::as_str) else { return; }; - if matches!(event_name, "run.submitted" | "run.running") && !object.contains_key("properties") - { + if matches!(event_name, "run.submitted" | "run.running") && !object.contains_key("properties") { let run_id = object.remove("run_id"); let ts = object.remove("ts"); object.insert("properties".to_string(), serde_json::json!({})); diff --git a/lib/crates/fabro-cli/src/commands/run/output.rs b/lib/crates/fabro-cli/src/commands/run/output.rs index 2872391ae..ad3eaccbf 100644 --- a/lib/crates/fabro-cli/src/commands/run/output.rs +++ b/lib/crates/fabro-cli/src/commands/run/output.rs @@ -9,12 +9,12 @@ use fabro_util::terminal::Styles; use fabro_util::text::strip_goal_decoration; use fabro_workflow::artifact_snapshot::collect_artifact_paths; use fabro_workflow::outcome::{StageStatus, format_cost}; -use fabro_workflow::pipeline::{Persisted, Validated}; +use fabro_workflow::pipeline::Validated; use fabro_workflow::records::Conclusion; use indicatif::HumanDuration; -use crate::shared::{format_tokens_human, print_diagnostics, relative_path, tilde_path}; use crate::server_client; +use crate::shared::{format_tokens_human, print_diagnostics, relative_path, tilde_path}; fn print_workflow_header( graph: &Graph, @@ -56,21 +56,6 @@ pub(crate) fn print_workflow_report( print_workflow_header(validated.graph(), validated.diagnostics(), dot_path, styles); } -pub(crate) fn print_workflow_report_from_persisted( - persisted: &Persisted, - dot_path: Option<&Path>, - styles: &Styles, -) { - print_workflow_header(persisted.graph(), persisted.diagnostics(), dot_path, styles); -} - -pub(crate) fn print_diagnostics_from_error( - diagnostics: &[fabro_validate::Diagnostic], - styles: &Styles, -) { - print_diagnostics(diagnostics, styles); -} - pub(crate) async fn print_run_summary( storage_dir: &Path, run_dir: &Path, @@ -104,7 +89,7 @@ pub(crate) async fn print_run_summary( pr_url.as_deref(), styles, ); - print_final_output(checkpoint.as_ref(), run_dir, styles).await; + print_final_output(checkpoint.as_ref(), run_dir, styles); print_assets(run_dir, styles); Ok(()) } @@ -198,7 +183,7 @@ pub(crate) fn print_run_conclusion( } } -pub(crate) async fn print_final_output( +pub(crate) fn print_final_output( checkpoint: Option<&fabro_types::Checkpoint>, _run_dir: &Path, styles: &Styles, diff --git a/lib/crates/fabro-cli/src/commands/run/preview.rs b/lib/crates/fabro-cli/src/commands/run/preview.rs index 9f71c952c..cc245686d 100644 --- a/lib/crates/fabro-cli/src/commands/run/preview.rs +++ b/lib/crates/fabro-cli/src/commands/run/preview.rs @@ -4,8 +4,8 @@ use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use tracing::info; use crate::args::{GlobalArgs, PreviewArgs}; -use crate::shared::{print_json_pretty, validate_daytona_provider}; use crate::server_client; +use crate::shared::{print_json_pretty, validate_daytona_provider}; use crate::user_config::load_user_settings_with_globals; pub(crate) async fn run(args: PreviewArgs, globals: &GlobalArgs) -> Result<()> { diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index f02bed8e3..407870da2 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs @@ -3,15 +3,13 @@ use anyhow::Result; use cli_table::format::{Border, Separator}; use cli_table::{Cell, CellStruct, Color, Style, Table}; use fabro_checkpoint::git::Store; +use fabro_types::run_event::{CheckpointCompletedProps, RunRewoundProps, RunStatusTransitionProps}; +use fabro_types::{EventBody, RunEvent}; use fabro_util::terminal::Styles; -use fabro_types::run_event::{ - CheckpointCompletedProps, RunRewoundProps, RunStatusTransitionProps, -}; use fabro_workflow::git::MetadataStore; use fabro_workflow::operations::{ RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline_or_rebuild, rewind, }; -use fabro_types::{EventBody, RunEvent}; use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use git2::Repository; use serde::Serialize; @@ -19,6 +17,7 @@ use serde::Serialize; use crate::args::{GlobalArgs, RewindArgs}; use crate::commands::store::rebuild::rebuild_run_store; use crate::server_client; +use crate::server_client::ServerStoreClient; use crate::shared::{color_if, print_json_pretty}; use crate::user_config::load_user_settings_with_globals; @@ -97,16 +96,15 @@ pub(crate) fn timeline_entries_json(timeline: &RunTimeline) -> Vec Result<()> { - let state = client - .get_run_state(run_id) - .await - .map_err(|err| anyhow::anyhow!("failed to load durable store state before rewind: {err}"))?; + let state = client.get_run_state(run_id).await.map_err(|err| { + anyhow::anyhow!("failed to load durable store state before rewind: {err}") + })?; let _run_record = state .run @@ -191,11 +189,7 @@ fn restored_checkpoint_event( ) } -fn run_event( - run_id: fabro_types::RunId, - node_id: Option, - body: EventBody, -) -> RunEvent { +fn run_event(run_id: fabro_types::RunId, node_id: Option, body: EventBody) -> RunEvent { RunEvent { id: ulid::Ulid::new().to_string(), ts: chrono::Utc::now(), diff --git a/lib/crates/fabro-cli/src/commands/run/wait.rs b/lib/crates/fabro-cli/src/commands/run/wait.rs index d6232c112..b0ddbd7eb 100644 --- a/lib/crates/fabro-cli/src/commands/run/wait.rs +++ b/lib/crates/fabro-cli/src/commands/run/wait.rs @@ -9,8 +9,8 @@ use fabro_workflow::run_status::RunStatus; use tracing::info; use crate::args::{GlobalArgs, WaitArgs}; -use crate::shared::format_duration_ms; use crate::server_client; +use crate::shared::format_duration_ms; use crate::user_config::load_user_settings_with_globals; #[cfg(test)] diff --git a/lib/crates/fabro-cli/src/commands/runs/inspect.rs b/lib/crates/fabro-cli/src/commands/runs/inspect.rs index 8bac124ca..2e436558d 100644 --- a/lib/crates/fabro-cli/src/commands/runs/inspect.rs +++ b/lib/crates/fabro-cli/src/commands/runs/inspect.rs @@ -9,6 +9,7 @@ use fabro_workflow::run_status::RunStatus; use crate::args::{GlobalArgs, InspectArgs}; use crate::server_client; +use crate::server_client::RunProjection; use crate::user_config::load_user_settings_with_globals; #[derive(Debug, Serialize)] @@ -41,14 +42,18 @@ fn inspect_run_state( run_id: &RunId, run_dir: &Path, status: RunStatus, - state: crate::server_client::RunProjection, + state: RunProjection, ) -> InspectOutput { InspectOutput { run_id: run_id.to_string(), run_dir: run_dir.to_path_buf(), status: state.status.as_ref().map_or(status, |record| record.status), - run_record: state.run.and_then(|record| serde_json::to_value(record).ok()), - start_record: state.start.and_then(|record| serde_json::to_value(record).ok()), + run_record: state + .run + .and_then(|record| serde_json::to_value(record).ok()), + start_record: state + .start + .and_then(|record| serde_json::to_value(record).ok()), conclusion: state .conclusion .and_then(|record| serde_json::to_value(record).ok()), diff --git a/lib/crates/fabro-cli/src/commands/runs/list.rs b/lib/crates/fabro-cli/src/commands/runs/list.rs index 053586267..2979edb5e 100644 --- a/lib/crates/fabro-cli/src/commands/runs/list.rs +++ b/lib/crates/fabro-cli/src/commands/runs/list.rs @@ -11,8 +11,8 @@ use fabro_workflow::run_lookup::{StatusFilter, filter_runs, runs_base, scan_runs use fabro_workflow::run_status::RunStatus; use crate::args::{GlobalArgs, RunsListArgs}; -use crate::shared::{color_if, format_duration_ms, tilde_path}; use crate::server_client; +use crate::shared::{color_if, format_duration_ms, tilde_path}; use crate::user_config::load_user_settings_with_globals; use super::short_run_id; diff --git a/lib/crates/fabro-cli/src/commands/runs/rm.rs b/lib/crates/fabro-cli/src/commands/runs/rm.rs index 118f79111..6b76bc038 100644 --- a/lib/crates/fabro-cli/src/commands/runs/rm.rs +++ b/lib/crates/fabro-cli/src/commands/runs/rm.rs @@ -1,16 +1,17 @@ use std::path::Path; use anyhow::{Context, Result, bail}; -use tracing::warn; - -use crate::args::{GlobalArgs, RunsRemoveArgs}; -use crate::server_client; -use crate::shared::print_json_pretty; -use crate::user_config::load_user_settings_with_globals; use fabro_sandbox::reconnect::reconnect as reconnect_sandbox; use fabro_workflow::event::{Event, to_run_event}; use fabro_workflow::run_lookup::RunInfo; use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; +use tracing::warn; + +use crate::args::{GlobalArgs, RunsRemoveArgs}; +use crate::server_client; +use crate::server_client::RunProjection; +use crate::shared::print_json_pretty; +use crate::user_config::load_user_settings_with_globals; use super::short_run_id; @@ -134,10 +135,7 @@ async fn remove_run_dir_with_cleanup( }; if run_state.is_some() { let run_event = to_run_event(&run_id, &Event::RunRemoving { reason: None }); - if let Err(err) = client - .append_run_event(&run_id, &run_event) - .await - { + if let Err(err) = client.append_run_event(&run_id, &run_event).await { warn!( run_id = %run_id, error = %err, @@ -146,7 +144,7 @@ async fn remove_run_dir_with_cleanup( } } - if let Some(record) = load_sandbox_record(run_state.as_ref()).await { + if let Some(record) = load_sandbox_record(run_state.as_ref()) { if record.provider != "local" { match reconnect_sandbox(&record).await { Ok(sandbox) => { @@ -175,9 +173,7 @@ async fn delete_run_store_state( .with_context(|| format!("failed to delete store state for {}", run.run_id())) } -async fn load_sandbox_record( - run_state: Option<&crate::server_client::RunProjection>, -) -> Option { +fn load_sandbox_record(run_state: Option<&RunProjection>) -> Option { if let Some(run_state) = run_state { return run_state.sandbox.clone(); } diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index 10ff029eb..58b6f30b4 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -1,5 +1,3 @@ -use std::io::ErrorKind; -use std::path::Path; use anyhow::{Context, Result}; #[cfg(test)] use fabro_store::StageId; @@ -8,6 +6,8 @@ use fabro_workflow::run_dump::RunDump; use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; #[cfg(test)] use serde::de::DeserializeOwned; +use std::io::ErrorKind; +use std::path::Path; use crate::args::{GlobalArgs, StoreDumpArgs}; use crate::commands::store::rebuild::rebuild_run_store; diff --git a/lib/crates/fabro-cli/src/commands/system/df.rs b/lib/crates/fabro-cli/src/commands/system/df.rs index 24f7ad10e..8e3cb840d 100644 --- a/lib/crates/fabro-cli/src/commands/system/df.rs +++ b/lib/crates/fabro-cli/src/commands/system/df.rs @@ -10,8 +10,8 @@ use fabro_workflow::run_lookup::{logs_base, runs_base, scan_runs_with_summaries} use fabro_workflow::run_status::RunStatus; use crate::args::{DfArgs, GlobalArgs}; -use crate::shared::{format_size, print_json_pretty}; use crate::server_client; +use crate::shared::{format_size, print_json_pretty}; use crate::user_config::load_user_settings_with_globals; #[derive(Serialize)] @@ -59,11 +59,10 @@ pub(super) async fn df_command(args: &DfArgs, globals: &GlobalArgs) -> Result<() &logs_base_dir, globals, ) - .await } #[allow(clippy::print_stdout)] -async fn df_from( +fn df_from( args: &DfArgs, summaries: &[fabro_store::RunSummary], data_dir: &Path, diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index 910e128b8..051ef6fdd 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -3,11 +3,10 @@ mod args; mod commands; mod logging; +mod server_client; mod shared; #[cfg(feature = "sleep_inhibitor")] mod sleep_inhibitor; -mod server_client; -mod store; mod user_config; use anyhow::Result; diff --git a/lib/crates/fabro-cli/src/server_client.rs b/lib/crates/fabro-cli/src/server_client.rs index 03be42009..3fa4e39f9 100644 --- a/lib/crates/fabro-cli/src/server_client.rs +++ b/lib/crates/fabro-cli/src/server_client.rs @@ -5,11 +5,14 @@ use std::time::Duration; use anyhow::{Context as _, Result, anyhow}; use fabro_api::types; +use fabro_server::bind::Bind; use fabro_store::{EventEnvelope, RunSummary, StageId}; use fabro_types::{ - Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunEvent, RunId, - RunRecord, RunStatusRecord, SandboxRecord, Settings, StartRecord, + Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunEvent, RunId, RunRecord, + RunStatusRecord, SandboxRecord, Settings, StartRecord, }; +use serde::de::DeserializeOwned; +use tokio::time::sleep; use crate::commands::server::start; @@ -98,8 +101,8 @@ pub(crate) async fn connect_server(storage_dir: &Path) -> Result path, - fabro_server::bind::Bind::Tcp(addr) => { + Bind::Unix(path) => path, + Bind::Tcp(addr) => { return Err(anyhow!( "Unsupported server bind for store client auto-connect: {addr}" )); @@ -133,7 +136,7 @@ async fn wait_for_server_ready(http_client: &reqwest::Client) -> Result<()> { } Err(err) => last_error = Some(anyhow!(err)), } - tokio::time::sleep(Duration::from_millis(50)).await; + sleep(Duration::from_millis(50)).await; } Err(last_error.unwrap_or_else(|| anyhow!("server did not become ready in time"))) @@ -327,7 +330,7 @@ where fn convert_type(value: TInput) -> Result where TInput: serde::Serialize, - TOutput: serde::de::DeserializeOwned, + TOutput: DeserializeOwned, { serde_json::from_value(serde_json::to_value(value)?).map_err(Into::into) } diff --git a/lib/crates/fabro-cli/src/store.rs b/lib/crates/fabro-cli/src/store.rs deleted file mode 100644 index a5ea08dcb..000000000 --- a/lib/crates/fabro-cli/src/store.rs +++ /dev/null @@ -1,26 +0,0 @@ -use std::path::Path; -use std::sync::Arc; -use std::time::Duration; - -use anyhow::Result; -use fabro_store::{SlateRunStore, SlateStore}; -use fabro_types::RunId; -use object_store::local::LocalFileSystem; - -pub(crate) fn build_store(storage_dir: &Path) -> Result> { - let store_path = storage_dir.join("store"); - std::fs::create_dir_all(&store_path)?; - let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_path)?); - Ok(Arc::new(SlateStore::new( - object_store, - "", - Duration::from_millis(1), - ))) -} - -pub(crate) async fn open_run_reader(storage_dir: &Path, run_id: &RunId) -> Result { - build_store(storage_dir)? - .open_run_reader(run_id) - .await - .map_err(Into::into) -} diff --git a/lib/crates/fabro-cli/tests/it/cmd/logs.rs b/lib/crates/fabro-cli/tests/it/cmd/logs.rs index c56ea2de4..f3fa054c9 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/logs.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/logs.rs @@ -63,7 +63,7 @@ fn logs_completed_run_outputs_raw_ndjson() { success: true exit_code: 0 ----- stdout ----- - {"event":"run.created","id":"[EVENT_ID]","properties":{"graph":{"attrs":{"goal":{"String":"Run tests and report results"},"rankdir":{"String":"LR"}},"edges":[{"attrs":{},"from":"start","to":"run_tests"},{"attrs":{},"from":"run_tests","to":"report"},{"attrs":{},"from":"report","to":"exit"}],"name":"Simple","nodes":{"exit":{"attrs":{"label":{"String":"Exit"},"shape":{"String":"Msquare"}},"id":"exit"},"report":{"attrs":{"label":{"String":"Report"},"prompt":{"String":"Summarize the test results"}},"id":"report"},"run_tests":{"attrs":{"label":{"String":"Run Tests"},"prompt":{"String":"Run the test suite and report results"}},"id":"run_tests"},"start":{"attrs":{"label":{"String":"Start"},"shape":{"String":"Mdiamond"}},"id":"start"}}},"host_repo_path":"[TEMP_DIR]","labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"run_dir":"[STORAGE_DIR]/runs/20260404-[ULID]","settings":{"auto_approve":true,"dry_run":true,"fabro":{"root":"fabro/"},"features":{"retros":false,"session_sandboxes":false},"goal":"Run tests and report results","hooks":[{"blocking":true,"command":"cargo fmt","event":"post_tool_use","matcher":"write_file|edit_file|apply_patch","name":"cargo-fmt","sandbox":null,"timeout_ms":null}],"labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"llm":{"fallbacks":null,"model":"claude-sonnet-4-6","provider":"anthropic"},"mode":"standalone","no_retro":true,"pull_request":{"auto_merge":false,"draft":false,"enabled":true,"merge_strategy":"squash"},"sandbox":{"daytona":{"auto_stop_interval":30,"labels":{"repo":"fabro-sh/fabro"},"network":null,"skip_clone":false,"snapshot":{"cpu":4,"disk":20,"dockerfile":"FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n","memory":8,"name":"fabro-v6"}},"devcontainer":null,"env":null,"local":null,"preserve":null,"provider":"local"},"storage_dir":"[STORAGE_DIR]","version":1},"workflow_slug":"simple","workflow_source":"digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} + {"event":"run.created","id":"[EVENT_ID]","properties":{"graph":{"attrs":{"goal":{"String":"Run tests and report results"},"rankdir":{"String":"LR"}},"edges":[{"attrs":{},"from":"start","to":"run_tests"},{"attrs":{},"from":"run_tests","to":"report"},{"attrs":{},"from":"report","to":"exit"}],"name":"Simple","nodes":{"exit":{"attrs":{"label":{"String":"Exit"},"shape":{"String":"Msquare"}},"id":"exit"},"report":{"attrs":{"label":{"String":"Report"},"prompt":{"String":"Summarize the test results"}},"id":"report"},"run_tests":{"attrs":{"label":{"String":"Run Tests"},"prompt":{"String":"Run the test suite and report results"}},"id":"run_tests"},"start":{"attrs":{"label":{"String":"Start"},"shape":{"String":"Mdiamond"}},"id":"start"}}},"host_repo_path":"[TEMP_DIR]","labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"run_dir":"[RUN_DIR]","settings":{"auto_approve":true,"dry_run":true,"fabro":{"root":"fabro/"},"features":{"retros":false,"session_sandboxes":false},"goal":"Run tests and report results","hooks":[{"blocking":true,"command":"cargo fmt","event":"post_tool_use","matcher":"write_file|edit_file|apply_patch","name":"cargo-fmt","sandbox":null,"timeout_ms":null}],"labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"llm":{"fallbacks":null,"model":"gpt-5.4","provider":"openai"},"mode":"standalone","no_retro":true,"pull_request":{"auto_merge":false,"draft":false,"enabled":true,"merge_strategy":"squash"},"sandbox":{"daytona":{"auto_stop_interval":30,"labels":{"repo":"fabro-sh/fabro"},"network":null,"skip_clone":false,"snapshot":{"cpu":4,"disk":20,"dockerfile":"FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n","memory":8,"name":"fabro-v6"}},"devcontainer":null,"env":null,"local":null,"preserve":null,"provider":"local"},"storage_dir":"[STORAGE_DIR]","version":1},"workflow_slug":"simple","workflow_source":"digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"run.submitted","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"run.starting","id":"[EVENT_ID]","properties":{"reason":"sandbox_initializing"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"sandbox.initializing","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} @@ -228,7 +228,7 @@ fn logs_follow_detached_run_streams_until_completion() { success: true exit_code: 0 ----- stdout ----- - {"event":"run.created","id":"[EVENT_ID]","properties":{"graph":{"attrs":{"goal":{"String":"Run tests and report results"},"rankdir":{"String":"LR"}},"edges":[{"attrs":{},"from":"start","to":"run_tests"},{"attrs":{},"from":"run_tests","to":"report"},{"attrs":{},"from":"report","to":"exit"}],"name":"Simple","nodes":{"exit":{"attrs":{"label":{"String":"Exit"},"shape":{"String":"Msquare"}},"id":"exit"},"report":{"attrs":{"label":{"String":"Report"},"prompt":{"String":"Summarize the test results"}},"id":"report"},"run_tests":{"attrs":{"label":{"String":"Run Tests"},"prompt":{"String":"Run the test suite and report results"}},"id":"run_tests"},"start":{"attrs":{"label":{"String":"Start"},"shape":{"String":"Mdiamond"}},"id":"start"}}},"host_repo_path":"[TEMP_DIR]","labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"run_dir":"[STORAGE_DIR]/runs/20260404-[ULID]","settings":{"auto_approve":true,"dry_run":true,"fabro":{"root":"fabro/"},"features":{"retros":false,"session_sandboxes":false},"goal":"Run tests and report results","hooks":[{"blocking":true,"command":"cargo fmt","event":"post_tool_use","matcher":"write_file|edit_file|apply_patch","name":"cargo-fmt","sandbox":null,"timeout_ms":null}],"labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"llm":{"fallbacks":null,"model":"claude-sonnet-4-6","provider":"anthropic"},"mode":"standalone","no_retro":true,"pull_request":{"auto_merge":false,"draft":false,"enabled":true,"merge_strategy":"squash"},"sandbox":{"daytona":{"auto_stop_interval":30,"labels":{"repo":"fabro-sh/fabro"},"network":null,"skip_clone":false,"snapshot":{"cpu":4,"disk":20,"dockerfile":"FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n","memory":8,"name":"fabro-v6"}},"devcontainer":null,"env":null,"local":null,"preserve":null,"provider":"local"},"storage_dir":"[STORAGE_DIR]","version":1},"workflow_slug":"simple","workflow_source":"digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} + {"event":"run.created","id":"[EVENT_ID]","properties":{"graph":{"attrs":{"goal":{"String":"Run tests and report results"},"rankdir":{"String":"LR"}},"edges":[{"attrs":{},"from":"start","to":"run_tests"},{"attrs":{},"from":"run_tests","to":"report"},{"attrs":{},"from":"report","to":"exit"}],"name":"Simple","nodes":{"exit":{"attrs":{"label":{"String":"Exit"},"shape":{"String":"Msquare"}},"id":"exit"},"report":{"attrs":{"label":{"String":"Report"},"prompt":{"String":"Summarize the test results"}},"id":"report"},"run_tests":{"attrs":{"label":{"String":"Run Tests"},"prompt":{"String":"Run the test suite and report results"}},"id":"run_tests"},"start":{"attrs":{"label":{"String":"Start"},"shape":{"String":"Mdiamond"}},"id":"start"}}},"host_repo_path":"[TEMP_DIR]","labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"run_dir":"[RUN_DIR]","settings":{"auto_approve":true,"dry_run":true,"fabro":{"root":"fabro/"},"features":{"retros":false,"session_sandboxes":false},"goal":"Run tests and report results","hooks":[{"blocking":true,"command":"cargo fmt","event":"post_tool_use","matcher":"write_file|edit_file|apply_patch","name":"cargo-fmt","sandbox":null,"timeout_ms":null}],"labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"llm":{"fallbacks":null,"model":"gpt-5.4","provider":"openai"},"mode":"standalone","no_retro":true,"pull_request":{"auto_merge":false,"draft":false,"enabled":true,"merge_strategy":"squash"},"sandbox":{"daytona":{"auto_stop_interval":30,"labels":{"repo":"fabro-sh/fabro"},"network":null,"skip_clone":false,"snapshot":{"cpu":4,"disk":20,"dockerfile":"FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n","memory":8,"name":"fabro-v6"}},"devcontainer":null,"env":null,"local":null,"preserve":null,"provider":"local"},"storage_dir":"[STORAGE_DIR]","version":1},"workflow_slug":"simple","workflow_source":"digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"run.submitted","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"run.starting","id":"[EVENT_ID]","properties":{"reason":"sandbox_initializing"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"sandbox.initializing","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} diff --git a/lib/crates/fabro-cli/tests/it/cmd/ps.rs b/lib/crates/fabro-cli/tests/it/cmd/ps.rs index dd5f28f73..3d93165d3 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/ps.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/ps.rs @@ -69,11 +69,13 @@ fn ps_all_json_lists_created_and_completed_runs() { "all runs should belong to the Simple workflow: {runs:#?}" ); assert!( - runs.iter().all(|run| run["labels"]["fabro_test_case"] == context.test_case_id()), + runs.iter() + .all(|run| run["labels"]["fabro_test_case"] == context.test_case_id()), "all runs should be scoped to the current test case: {runs:#?}" ); assert!( - runs.iter().all(|run| run["labels"]["fabro_test_run"] == context.test_run_id()), + runs.iter() + .all(|run| run["labels"]["fabro_test_run"] == context.test_run_id()), "all runs should be scoped to the current test session: {runs:#?}" ); assert!( @@ -152,7 +154,11 @@ fn ps_filters_by_workflow_and_label() { assert!(output.status.success(), "ps should succeed"); let runs: Vec = serde_json::from_slice(&output.stdout).expect("ps JSON should parse"); - assert_eq!(runs.len(), 1, "workflow+label filter should isolate one run"); + assert_eq!( + runs.len(), + 1, + "workflow+label filter should isolate one run" + ); let run = &runs[0]; assert_eq!(run["workflow_name"], "Simple"); assert_eq!(run["status"], "succeeded"); diff --git a/lib/crates/fabro-cli/tests/it/cmd/rm.rs b/lib/crates/fabro-cli/tests/it/cmd/rm.rs index 1465409ff..e411743d2 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/rm.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/rm.rs @@ -1,7 +1,9 @@ use fabro_test::{fabro_snapshot, test_context}; use serde_json::Value; -use super::support::{setup_completed_fast_dry_run, setup_created_fast_dry_run, setup_local_sandbox_run}; +use super::support::{ + setup_completed_fast_dry_run, setup_created_fast_dry_run, setup_local_sandbox_run, +}; #[test] fn help() { diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 1e6fec3e6..a2495218a 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -68,7 +68,7 @@ fn dry_run_simple() { Run: [ULID] Status: SUCCESS Duration: [DURATION] - Run: [STORAGE_DIR]/runs/20260404-[ULID] + Run: [RUN_DIR] === Output === [Simulated] Response for stage: report @@ -348,8 +348,8 @@ fn json_run_implies_auto_approve_for_human_gates() { "goal": "Route through the default approval path", "llm": { "fallbacks": null, - "model": "claude-sonnet-4-6", - "provider": "anthropic" + "model": "gpt-5.4", + "provider": "openai" }, "mode": "standalone", "no_retro": true, diff --git a/lib/crates/fabro-cli/tests/it/cmd/server_start.rs b/lib/crates/fabro-cli/tests/it/cmd/server_start.rs index 2e3617965..48d37974f 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/server_start.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/server_start.rs @@ -100,7 +100,11 @@ fn start_already_running_exits_with_error() { #[test] fn concurrent_autostart_converges_on_one_shared_daemon_and_cleans_up() { - fn run_ps_json(home_dir: &std::path::Path, temp_dir: &std::path::Path, storage_dir: &std::path::Path) -> std::process::Output { + fn run_ps_json( + home_dir: &std::path::Path, + temp_dir: &std::path::Path, + storage_dir: &std::path::Path, + ) -> std::process::Output { std::process::Command::new(env!("CARGO_BIN_EXE_fabro")) .current_dir(temp_dir) .env("NO_COLOR", "1") diff --git a/lib/crates/fabro-cli/tests/it/cmd/start.rs b/lib/crates/fabro-cli/tests/it/cmd/start.rs index 55f6e9db4..d862b61d5 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/start.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/start.rs @@ -53,7 +53,11 @@ fn start_by_run_id_starts_created_run() { .assert() .success(); - context.command().args(["start", &run_id]).assert().success(); + context + .command() + .args(["start", &run_id]) + .assert() + .success(); context .command() .args(["wait", &run_id]) @@ -102,7 +106,11 @@ fn start_by_run_id_starts_created_run_without_run_json_or_status_json() { let run_dir = context.find_run_dir(&run_id); let _ = std::fs::remove_file(run_dir.join("run.json")); - context.command().args(["start", &run_id]).assert().success(); + context + .command() + .args(["start", &run_id]) + .assert() + .success(); let output = context .command() .args(["wait", "--json", &run_id]) diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index dd3659b91..cb77932b1 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -6,8 +6,8 @@ use std::time::{Duration, Instant}; use fabro_store::EventEnvelope; use fabro_test::TestContext; use fabro_types::{ - Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunRecord, - RunStatusRecord, SandboxRecord, StageId, StartRecord, + Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunRecord, RunStatusRecord, + SandboxRecord, StageId, StartRecord, }; use serde_json::Value; use shlex::try_quote; @@ -185,7 +185,13 @@ fn run_completed_dry_run(context: &TestContext, workflow: &Path) -> RunSetup { let mut cmd = context.run_cmd(); cmd.current_dir(&context.temp_dir); cmd.timeout(COMMAND_TIMEOUT); - cmd.args(["--dry-run", "--auto-approve", "--no-retro", "--sandbox", "local"]); + cmd.args([ + "--dry-run", + "--auto-approve", + "--no-retro", + "--sandbox", + "local", + ]); cmd.arg(workflow); let output = cmd.output().expect("command should execute"); if !output.status.success() { @@ -213,7 +219,13 @@ fn run_created_dry_run(context: &TestContext, workflow: &Path) -> RunSetup { let mut cmd = context.create_cmd(); cmd.current_dir(&context.temp_dir); cmd.timeout(COMMAND_TIMEOUT); - cmd.args(["--dry-run", "--auto-approve", "--no-retro", "--sandbox", "local"]); + cmd.args([ + "--dry-run", + "--auto-approve", + "--no-retro", + "--sandbox", + "local", + ]); cmd.arg(workflow); let output = cmd.output().expect("command should execute"); if !output.status.success() { @@ -668,13 +680,18 @@ async fn get_server_json_for_storage( pub(crate) fn run_state(run_dir: &Path) -> RunProjection { let run_id = infer_run_id(run_dir); - block_on(get_server_json(run_dir, &format!("/api/v1/runs/{run_id}/state"))) + block_on(get_server_json( + run_dir, + &format!("/api/v1/runs/{run_id}/state"), + )) } pub(crate) fn run_events(run_dir: &Path) -> Vec { let run_id = infer_run_id(run_dir); - let response: serde_json::Value = - block_on(get_server_json(run_dir, &format!("/api/v1/runs/{run_id}/events"))); + let response: serde_json::Value = block_on(get_server_json( + run_dir, + &format!("/api/v1/runs/{run_id}/events"), + )); serde_json::from_value(response["data"].clone()).expect("event list should parse") } diff --git a/lib/crates/fabro-cli/tests/it/cmd/system_prune.rs b/lib/crates/fabro-cli/tests/it/cmd/system_prune.rs index 3f441073d..175b10a6c 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/system_prune.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/system_prune.rs @@ -58,7 +58,7 @@ fn system_prune_dry_run_lists_matching_runs_without_deleting() { success: true exit_code: 0 ----- stdout ----- - would delete: 20260404-[ULID] (Simple) + would delete: 20260405-[ULID] (Simple) ----- stderr ----- 1 run(s) would be deleted ([SIZE] freed). Pass --yes to confirm. diff --git a/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs b/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs index 417fbb75f..1227efc01 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs @@ -119,7 +119,11 @@ fn dry_run_create_start_attach_works_with_default_run_lookup() { .assert() .success(); - context.command().args(["start", &run_id]).assert().success(); + context + .command() + .args(["start", &run_id]) + .assert() + .success(); context .command() .args(["attach", &run_id]) diff --git a/lib/crates/fabro-cli/tests/it/support/mod.rs b/lib/crates/fabro-cli/tests/it/support/mod.rs index 1c8bb65f8..98d0e0e94 100644 --- a/lib/crates/fabro-cli/tests/it/support/mod.rs +++ b/lib/crates/fabro-cli/tests/it/support/mod.rs @@ -1,7 +1,7 @@ use std::path::{Path, PathBuf}; -use fabro_types::RunId; use fabro_test::TestContext; +use fabro_types::RunId; macro_rules! fabro_json_snapshot { ($context:expr, $value:expr, @$snapshot:literal) => {{ let mut filters = $context.filters(); diff --git a/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs b/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs index a241a9bd3..6b83f0bc5 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs @@ -30,7 +30,7 @@ fn dry_run_branching() { Run: [ULID] Status: SUCCESS Duration: [DURATION] - Run: [STORAGE_DIR]/runs/20260404-[ULID] + Run: [RUN_DIR] === Output === [Simulated] Response for stage: validate @@ -62,7 +62,7 @@ fn dry_run_conditions() { Run: [ULID] Status: SUCCESS Duration: [DURATION] - Run: [STORAGE_DIR]/runs/20260404-[ULID] + Run: [RUN_DIR] === Output === [Simulated] Response for stage: path_b @@ -95,7 +95,7 @@ fn dry_run_parallel() { Run: [ULID] Status: SUCCESS Duration: [DURATION] - Run: [STORAGE_DIR]/runs/20260404-[ULID] + Run: [RUN_DIR] === Output === [Simulated] Response for stage: review @@ -128,7 +128,7 @@ fn dry_run_styled() { Run: [ULID] Status: SUCCESS Duration: [DURATION] - Run: [STORAGE_DIR]/runs/20260404-[ULID] + Run: [RUN_DIR] === Output === [Simulated] Response for stage: critical_review @@ -159,6 +159,6 @@ fn dry_run_legacy_tool() { Run: [ULID] Status: SUCCESS Duration: [DURATION] - Run: [STORAGE_DIR]/runs/20260404-[ULID] + Run: [RUN_DIR] "); } diff --git a/lib/crates/fabro-server/src/jwt_auth.rs b/lib/crates/fabro-server/src/jwt_auth.rs index 4bc9185ef..f2ee59b32 100644 --- a/lib/crates/fabro-server/src/jwt_auth.rs +++ b/lib/crates/fabro-server/src/jwt_auth.rs @@ -77,7 +77,9 @@ pub fn resolve_auth_mode(api_settings: &ApiSettings, allowed_usernames: &[String if api_settings.authentication_strategies.is_empty() && std::env::var("FABRO_LOCAL_NO_AUTH").ok().as_deref() == Some("1") { - warn!("No authentication strategies configured; allowing unauthenticated local daemon access"); + warn!( + "No authentication strategies configured; allowing unauthenticated local daemon access" + ); return AuthMode::Disabled; } diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index a9f78628d..9bad67730 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -51,6 +51,9 @@ use fabro_workflow::event::Emitter; use fabro_workflow::operations::{self, CreateRunInput, WorkflowInput}; use fabro_workflow::pipeline::Persisted; use fabro_workflow::records::Checkpoint; +use fabro_workflow::run_status::RunStatus as WorkflowRunStatus; +#[cfg(test)] +use fabro_workflow::run_status::StatusReason as WorkflowStatusReason; use fabro_api::types::AggregateUsageTotals; pub use fabro_api::types::{ @@ -59,8 +62,8 @@ pub use fabro_api::types::{ CompletionResponse, CompletionToolChoiceMode, CompletionUsage, CreateCompletionRequest, CreateRunRequest, EventEnvelope as ApiEventEnvelope, ModelReference, PaginatedEventList, PaginatedRunList, PaginationMeta, QuestionType as ApiQuestionType, RunError, - RunEvent as ApiRunEvent, RunStatus, RunStatusResponse, StartRunRequest, - SubmitAnswerRequest, TokenUsage, UsageByModel, WriteBlobResponse, + RunEvent as ApiRunEvent, RunStatus, RunStatusResponse, StartRunRequest, SubmitAnswerRequest, + TokenUsage, UsageByModel, WriteBlobResponse, }; pub fn default_page_limit() -> u32 { @@ -183,7 +186,6 @@ impl AppState { } } - /// Build the axum Router with all run endpoints and embedded static assets. pub fn build_router(state: Arc, auth_mode: AuthMode) -> Router { let middleware_state = Arc::clone(&state); @@ -242,7 +244,10 @@ fn demo_routes() -> Router> { .route("/runs/{id}/questions", get(demo::get_questions_stub)) .route("/runs/{id}/questions/{qid}/answer", post(demo::answer_stub)) .route("/runs/{id}/state", get(not_implemented)) - .route("/runs/{id}/events", get(not_implemented).post(not_implemented)) + .route( + "/runs/{id}/events", + get(not_implemented).post(not_implemented), + ) .route("/runs/{id}/attach", get(demo::run_events_stub)) .route("/runs/{id}/blobs", post(not_implemented)) .route("/runs/{id}/blobs/{blobId}", get(not_implemented)) @@ -331,7 +336,10 @@ fn real_routes() -> Router> { .route("/runs/{id}/questions", get(get_questions)) .route("/runs/{id}/questions/{qid}/answer", post(submit_answer)) .route("/runs/{id}/state", get(get_run_state)) - .route("/runs/{id}/events", get(list_run_events).post(append_run_event)) + .route( + "/runs/{id}/events", + get(list_run_events).post(append_run_event), + ) .route("/runs/{id}/attach", get(attach_run_events)) .route("/runs/{id}/blobs", post(write_run_blob)) .route("/runs/{id}/blobs/{blobId}", get(read_run_blob)) @@ -569,17 +577,16 @@ async fn list_board_runs( .into_response() } -async fn list_runs( - _auth: AuthenticatedService, - State(state): State>, -) -> Response { +async fn list_runs(_auth: AuthenticatedService, State(state): State>) -> Response { match state .store .list_runs(&fabro_store::ListRunsQuery::default()) .await { Ok(runs) => (StatusCode::OK, Json(runs)).into_response(), - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(), + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } } } @@ -599,7 +606,9 @@ async fn delete_run( match state.store.delete_run(&id).await { Ok(()) => StatusCode::NO_CONTENT.into_response(), - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(), + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } } } @@ -714,7 +723,10 @@ async fn create_run( }; info!(run_id = %run_id, "Run created"); - let using_dot_source = req.dot_source.as_ref().is_some_and(|value| !value.is_empty()); + let using_dot_source = req + .dot_source + .as_ref() + .is_some_and(|value| !value.is_empty()); let using_local_workflow = req .workflow_path .as_ref() @@ -807,7 +819,7 @@ async fn create_run( created_at, }), ) - .into_response() + .into_response() } async fn start_run( @@ -820,7 +832,7 @@ async fn start_run( Ok(id) => id, Err(response) => return response, }; - let resume = body.map(|Json(req)| req.resume).unwrap_or(false); + let resume = body.is_some_and(|Json(req)| req.resume); { let runs = state.runs.lock().expect("runs lock poisoned"); @@ -842,9 +854,8 @@ async fn start_run( } } - let run_store = match state.store.open_run(&id).await { - Ok(run_store) => run_store, - Err(_) => return ApiError::not_found("Run not found.").into_response(), + let Ok(run_store) = state.store.open_run(&id).await else { + return ApiError::not_found("Run not found.").into_response(); }; let run_state = match run_store.state().await { Ok(state) => state, @@ -863,19 +874,27 @@ async fn start_run( .into_response(); } } else if let Some(record) = run_state.status.as_ref() { - if !matches!(record.status, fabro_workflow::run_status::RunStatus::Submitted | fabro_workflow::run_status::RunStatus::Starting) - { + if !matches!( + record.status, + WorkflowRunStatus::Submitted | WorkflowRunStatus::Starting + ) { return ApiError::new( StatusCode::CONFLICT, - format!("cannot start run: status is {:?}, expected submitted", record.status), + format!( + "cannot start run: status is {:?}, expected submitted", + record.status + ), ) .into_response(); } } let Some(run_record) = run_state.run.as_ref() else { - return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, "run record missing from store") - .into_response(); + return ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "run record missing from store", + ) + .into_response(); }; let run_dir = operations::make_run_dir(&run_record.settings.storage_dir().join("runs"), &id); let dot_source = run_state.graph_source.unwrap_or_default(); @@ -909,7 +928,7 @@ async fn start_run( created_at: id.created_at(), }), ) - .into_response() + .into_response() } /// Execute a single run: transitions queued → starting → running → completed/failed/cancelled. @@ -1174,7 +1193,9 @@ async fn get_run_status( Some(run) => (StatusCode::OK, Json(run)).into_response(), None => ApiError::not_found("Run not found.").into_response(), }, - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(), + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } } } @@ -1305,8 +1326,9 @@ async fn get_run_state( match state.store.open_run_reader(&id).await { Ok(run_store) => match run_store.state().await { Ok(run_state) => Json(run_state).into_response(), - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response(), + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } }, Err(_) => ApiError::not_found("Run not found.").into_response(), } @@ -1324,7 +1346,9 @@ async fn append_run_event( }; let event = match RunEvent::from_value(value.clone()) { Ok(event) => event, - Err(err) => return ApiError::bad_request(format!("Invalid run event: {err}")).into_response(), + Err(err) => { + return ApiError::bad_request(format!("Invalid run event: {err}")).into_response(); + } }; if event.run_id != id { return ApiError::bad_request("Event run_id does not match path run ID.").into_response(); @@ -1336,9 +1360,13 @@ async fn append_run_event( match state.store.open_run(&id).await { Ok(run_store) => match run_store.append_event(&payload).await { - Ok(seq) => Json(AppendEventResponse { seq: i64::from(seq) }).into_response(), - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response(), + Ok(seq) => Json(AppendEventResponse { + seq: i64::from(seq), + }) + .into_response(), + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } }, Err(_) => ApiError::not_found("Run not found.").into_response(), } @@ -1357,7 +1385,10 @@ async fn list_run_events( let since_seq = params.since_seq(); let limit = params.limit(); match state.store.open_run_reader(&id).await { - Ok(run_store) => match run_store.list_events_from_with_limit(since_seq, limit).await { + Ok(run_store) => match run_store + .list_events_from_with_limit(since_seq, limit) + .await + { Ok(mut events) => { let has_more = events.len() > limit; events.truncate(limit); @@ -1375,8 +1406,9 @@ async fn list_run_events( }) .into_response() } - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response(), + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } }, Err(_) => ApiError::not_found("Run not found.").into_response(), } @@ -1427,18 +1459,16 @@ async fn attach_run_events( .into_response(); } }; - let stream = stream.filter_map(|result| { - match result { - Ok(event) => { - let event = api_event_envelope_from_store(&event).ok()?; - let data = serde_json::to_string(&event).ok()?; - let data = redact_jsonl_line(&data); - Some(Ok::( - Event::default().data(data), - )) - } - Err(_) => None, + let stream = stream.filter_map(|result| match result { + Ok(event) => { + let event = api_event_envelope_from_store(&event).ok()?; + let data = serde_json::to_string(&event).ok()?; + let data = redact_jsonl_line(&data); + Some(Ok::( + Event::default().data(data), + )) } + Err(_) => None, }); Sse::new(stream).into_response() @@ -1498,8 +1528,9 @@ async fn write_run_blob( id: blob_id.to_string(), }) .into_response(), - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response(), + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } }, Err(_) => ApiError::not_found("Run not found.").into_response(), } @@ -1522,8 +1553,9 @@ async fn read_run_blob( Ok(run_store) => match run_store.read_blob(&blob_id).await { Ok(Some(bytes)) => octet_stream_response(bytes), Ok(None) => ApiError::not_found("Blob not found.").into_response(), - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response(), + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } }, Err(_) => ApiError::not_found("Run not found.").into_response(), } @@ -1551,8 +1583,9 @@ async fn list_stage_artifacts( .collect(), }) .into_response(), - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response(), + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } }, Err(_) => ApiError::not_found("Run not found.").into_response(), } @@ -1580,8 +1613,9 @@ async fn put_stage_artifact( match state.store.open_run(&id).await { Ok(run_store) => match run_store.put_artifact(&stage_id, &filename, &body).await { Ok(()) => StatusCode::NO_CONTENT.into_response(), - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response(), + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } }, Err(_) => ApiError::not_found("Run not found.").into_response(), } @@ -1609,8 +1643,9 @@ async fn get_stage_artifact( Ok(run_store) => match run_store.get_artifact(&stage_id, &filename).await { Ok(Some(bytes)) => octet_stream_response(bytes), Ok(None) => ApiError::not_found("Artifact not found.").into_response(), - Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response(), + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } }, Err(_) => ApiError::not_found("Run not found.").into_response(), } @@ -3392,8 +3427,8 @@ mod tests { let mut status_record = None; for _ in 0..50 { if let Some(record) = run_store.state().await.unwrap().status { - if record.status == fabro_workflow::run_status::RunStatus::Failed - && record.reason == Some(fabro_workflow::run_status::StatusReason::Cancelled) + if record.status == WorkflowRunStatus::Failed + && record.reason == Some(WorkflowStatusReason::Cancelled) { status_record = Some(record); break; @@ -3403,14 +3438,8 @@ mod tests { } let status_record = status_record.expect("status record should be persisted"); - assert_eq!( - status_record.status, - fabro_workflow::run_status::RunStatus::Failed - ); - assert_eq!( - status_record.reason, - Some(fabro_workflow::run_status::StatusReason::Cancelled) - ); + assert_eq!(status_record.status, WorkflowRunStatus::Failed); + assert_eq!(status_record.reason, Some(WorkflowStatusReason::Cancelled)); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] diff --git a/lib/crates/fabro-store/src/keys.rs b/lib/crates/fabro-store/src/keys.rs index f44242205..eab5bd275 100644 --- a/lib/crates/fabro-store/src/keys.rs +++ b/lib/crates/fabro-store/src/keys.rs @@ -66,7 +66,11 @@ pub(crate) fn parse_event_seq(key: &str) -> Option { } pub(crate) fn parse_blob_id(key: &str) -> Option { - key.rsplit('/').next()?.strip_prefix(BLOBS_PREFIX)?.parse().ok() + key.rsplit('/') + .next()? + .strip_prefix(BLOBS_PREFIX)? + .parse() + .ok() } pub(crate) fn parse_node_artifact_key(key: &str) -> Option<(StageId, String)> { @@ -119,7 +123,10 @@ mod tests { let node = StageId::new("code", 2); let run_id = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap(); let blob_id = RunBlobId::new(&run_id, b"summary"); - assert_eq!(blob_key(&run_id, &blob_id), format!("runs/{run_id}/blobs#{blob_id}")); + assert_eq!( + blob_key(&run_id, &blob_id), + format!("runs/{run_id}/blobs#{blob_id}") + ); assert_eq!( node_artifact(&run_id, &node, "src/main.rs"), "runs/01JT56VE4Z5NZ814GZN2JZD65A/artifacts#nodes#code#visit-2#src/main.rs" diff --git a/lib/crates/fabro-store/src/slate/catalog.rs b/lib/crates/fabro-store/src/slate/catalog.rs index fa26d8253..e8f484400 100644 --- a/lib/crates/fabro-store/src/slate/catalog.rs +++ b/lib/crates/fabro-store/src/slate/catalog.rs @@ -25,8 +25,9 @@ pub(crate) async fn list_run_ids(db: &Db, query: &ListRunsQuery) -> Result Result { - let event_seq = recover_next_seq(&db, &keys::events_prefix(&run_id), keys::parse_event_seq).await?; + let event_seq = + recover_next_seq(&db, &keys::events_prefix(&run_id), keys::parse_event_seq).await?; let (event_tx, _) = broadcast::channel(DEFAULT_EVENT_TAIL_LIMIT.max(16)); Ok(Self { inner: Arc::new(SlateRunStoreInner { @@ -71,7 +72,8 @@ impl SlateRunStore { } pub(crate) async fn open_reader(run_id: RunId, db: Db) -> Result { - let event_seq = recover_next_seq(&db, &keys::events_prefix(&run_id), keys::parse_event_seq).await?; + let event_seq = + recover_next_seq(&db, &keys::events_prefix(&run_id), keys::parse_event_seq).await?; let (event_tx, _) = broadcast::channel(DEFAULT_EVENT_TAIL_LIMIT.max(16)); Ok(Self { inner: Arc::new(SlateRunStoreInner { @@ -96,7 +98,7 @@ impl SlateRunStore { } } - pub(crate) fn into_read_only(&self) -> Self { + pub(crate) fn read_only_clone(&self) -> Self { Self { inner: Arc::clone(&self.inner), read_only: true, @@ -203,10 +205,13 @@ impl SlateRunStore { seq, payload: payload.clone(), }; - self.inner.db.put( - keys::event_key(&self.inner.run_id, seq, Utc::now().timestamp_millis()), - serde_json::to_vec(payload)?, - ).await?; + self.inner + .db + .put( + keys::event_key(&self.inner.run_id, seq, Utc::now().timestamp_millis()), + serde_json::to_vec(payload)?, + ) + .await?; self.cache_event(&event).await?; Ok(seq) } @@ -293,7 +298,10 @@ impl SlateRunStore { } self.inner .db - .put(keys::node_artifact(&self.inner.run_id, node, filename), data) + .put( + keys::node_artifact(&self.inner.run_id, node, filename), + data, + ) .await?; Ok(()) } @@ -350,7 +358,9 @@ async fn list_events_from(db: &R, run_id: &RunId, start_seq: u32) -> Result(db: &R, run_id: &RunId) -> Result> where R: DbRead + Sync, { - let mut iter = db.scan_prefix(keys::blobs_prefix(run_id).as_bytes()).await?; + let mut iter = db + .scan_prefix(keys::blobs_prefix(run_id).as_bytes()) + .await?; let mut blob_ids = Vec::new(); while let Some(entry) = iter.next().await? { let key = key_to_string(&entry.key)?; @@ -404,9 +416,7 @@ async fn list_all_artifacts(db: &R, run_id: &RunId) -> Result(db: &R, run_id: &RunId, stage_id: &StageId) -> Result> +async fn list_artifacts_for_stage( + db: &R, + run_id: &RunId, + stage_id: &StageId, +) -> Result> where R: DbRead + Sync, { diff --git a/lib/crates/fabro-test/src/lib.rs b/lib/crates/fabro-test/src/lib.rs index 323f1967c..3db30c751 100644 --- a/lib/crates/fabro-test/src/lib.rs +++ b/lib/crates/fabro-test/src/lib.rs @@ -204,7 +204,7 @@ fn shorten_session_id(id: &str) -> String { let trimmed = id.trim(); let shortened: String = trimmed .chars() - .filter(|ch| ch.is_ascii_alphanumeric()) + .filter(char::is_ascii_alphanumeric) .take(12) .collect(); if shortened.is_empty() { @@ -339,7 +339,10 @@ fn reap_stale_session_roots(fabro_bin: &Path, mode: SessionMode) { if !root.is_dir() { continue; } - let file_name = root.file_name().and_then(|name| name.to_str()).unwrap_or(""); + let file_name = root + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or(""); let expected_prefix = match mode { SessionMode::Nextest => "n-", SessionMode::Process => "p-", @@ -494,10 +497,7 @@ impl TestContext { cmd.env("HOME", &self.home_dir); cmd.env("FABRO_NO_UPGRADE_CHECK", "true"); cmd.env("FABRO_STORAGE_DIR", &self.storage_dir); - cmd.env( - "FABRO_SERVER_MAX_CONCURRENT_RUNS", - "64", - ); + cmd.env("FABRO_SERVER_MAX_CONCURRENT_RUNS", "64"); cmd } @@ -1283,7 +1283,11 @@ mod tests { let _guard = EnvGuard::set("NEXTEST_RUN_ID", None); let (_, run_id, paths) = session_paths(); assert_eq!(run_id, format!("process-{}", current_pid())); - assert!(paths.root.ends_with(Path::new("fx").join(format!("p-{}", current_pid())))); + assert!( + paths + .root + .ends_with(Path::new("fx").join(format!("p-{}", current_pid()))) + ); assert_eq!(paths.storage_dir, paths.root.join("storage")); } diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs index cf15ab78f..8413572f3 100644 --- a/lib/crates/fabro-workflow/src/run_lookup.rs +++ b/lib/crates/fabro-workflow/src/run_lookup.rs @@ -199,7 +199,7 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result Result> {