From e92a2ba9dbaf7175d8c195d6d23c03867cdc28aa Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 21 Apr 2026 09:10:57 -0400 Subject: [PATCH] refactor(server): drop duplicate run status adapters Use the unified status family directly in server.rs and remove the manual conversion helpers that only existed to bridge duplicate API/domain types. --- lib/crates/fabro-server/src/server.rs | 198 +++++++++----------------- 1 file changed, 66 insertions(+), 132 deletions(-) diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 81fe6e50f..11d48aa48 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -25,18 +25,17 @@ use bytes::Bytes; pub use fabro_api::types::{ AggregateBilling, AggregateBillingTotals, ApiQuestion, ApiQuestionOption, AppendEventResponse, ArtifactEntry, ArtifactListResponse, BilledTokenCounts as ApiBilledTokenCounts, BillingByModel, - BillingStageRef, BlockedReason as ApiBlockedReason, CompletionContentPart, CompletionMessage, - CompletionMessageRole, CompletionResponse, CompletionToolChoiceMode, CompletionUsage, - CreateCompletionRequest, CreateSecretRequest, DeleteSecretRequest, DiskUsageResponse, - DiskUsageRunRow, DiskUsageSummaryRow, EventEnvelope as ApiEventEnvelope, ModelReference, - PaginatedEventList, PaginatedRunList, PaginationMeta, PreflightResponse, PreviewUrlRequest, - PreviewUrlResponse, PruneRunEntry, PruneRunsRequest, PruneRunsResponse, - QuestionType as ApiQuestionType, RenderWorkflowGraphDirection, RenderWorkflowGraphRequest, - RunArtifactEntry, RunArtifactListResponse, RunBilling, RunBillingStage, RunBillingTotals, - RunControlAction as ApiRunControlAction, RunError, RunManifest, RunStage, RunStatus, - RunStatusResponse, SandboxFileEntry, SandboxFileListResponse, SecretType as ApiSecretType, - ServerSettings, SshAccessRequest, SshAccessResponse, StageStatus as ApiStageStatus, - StartRunRequest, StatusReason as ApiStatusReason, SubmitAnswerRequest, SystemFeatures, + BillingStageRef, CompletionContentPart, CompletionMessage, CompletionMessageRole, + CompletionResponse, CompletionToolChoiceMode, CompletionUsage, CreateCompletionRequest, + CreateSecretRequest, DeleteSecretRequest, DiskUsageResponse, DiskUsageRunRow, + DiskUsageSummaryRow, EventEnvelope as ApiEventEnvelope, ModelReference, PaginatedEventList, + PaginatedRunList, PaginationMeta, PreflightResponse, PreviewUrlRequest, PreviewUrlResponse, + PruneRunEntry, PruneRunsRequest, PruneRunsResponse, QuestionType as ApiQuestionType, + RenderWorkflowGraphDirection, RenderWorkflowGraphRequest, RunArtifactEntry, + RunArtifactListResponse, RunBilling, RunBillingStage, RunBillingTotals, RunError, RunManifest, + RunStage, RunStatusResponse, SandboxFileEntry, SandboxFileListResponse, + SecretType as ApiSecretType, ServerSettings, SshAccessRequest, SshAccessResponse, + StageStatus as ApiStageStatus, StartRunRequest, SubmitAnswerRequest, SystemFeatures, SystemInfoResponse, SystemRunCounts, WriteBlobResponse, }; use fabro_auth::parse_credential_secret; @@ -86,9 +85,7 @@ use fabro_workflow::records::Checkpoint; use fabro_workflow::run_lookup::{ RunInfo, StatusFilter, filter_runs, scan_runs_with_summaries, scratch_base, }; -use fabro_workflow::run_status::{ - RunStatus as WorkflowRunStatus, StatusReason as WorkflowStatusReason, -}; +use fabro_workflow::run_status::{RunStatus, StatusReason}; use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, Validation}; use object_store::memory::InMemory as MemoryObjectStore; use rand::TryRngCore; @@ -2724,16 +2721,14 @@ fn test_secret_store_path() -> PathBuf { dir.join("secrets.json") } -fn board_column(status: WorkflowRunStatus) -> Option<&'static str> { +fn board_column(status: RunStatus) -> Option<&'static str> { match status { - WorkflowRunStatus::Submitted | WorkflowRunStatus::Queued | WorkflowRunStatus::Starting => { - Some("initializing") - } - WorkflowRunStatus::Running | WorkflowRunStatus::Paused => Some("running"), - WorkflowRunStatus::Blocked => Some("blocked"), - WorkflowRunStatus::Succeeded => Some("succeeded"), - WorkflowRunStatus::Failed | WorkflowRunStatus::Dead => Some("failed"), - WorkflowRunStatus::Removing | WorkflowRunStatus::Archived => None, + RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting => Some("initializing"), + RunStatus::Running | RunStatus::Paused => Some("running"), + RunStatus::Blocked => Some("blocked"), + RunStatus::Succeeded => Some("succeeded"), + RunStatus::Failed | RunStatus::Dead => Some("failed"), + RunStatus::Removing | RunStatus::Archived => None, } } @@ -2788,9 +2783,9 @@ fn summary_to_api_run_summary(summary: fabro_store::RunSummary) -> serde_json::V "repository": { "name": repository }, "start_time": summary.start_time.map(|time| time.to_rfc3339()), "status": summary.status, - "status_reason": summary.status_reason.map(api_status_reason), - "blocked_reason": summary.blocked_reason.map(api_blocked_reason), - "pending_control": summary.pending_control.map(api_pending_control), + "status_reason": summary.status_reason, + "blocked_reason": summary.blocked_reason, + "pending_control": summary.pending_control, "duration_ms": summary.duration_ms, "elapsed_secs": elapsed_secs(summary.duration_ms), "total_usd_micros": summary.total_usd_micros, @@ -2922,7 +2917,7 @@ async fn list_runs( let include_archived = params.include_archived; let items = runs .into_iter() - .filter(|summary| include_archived || summary.status != WorkflowRunStatus::Archived) + .filter(|summary| include_archived || summary.status != RunStatus::Archived) .map(summary_to_api_run_summary) .collect::>(); let (data, has_more) = paginate_items(items, ¶ms.pagination()); @@ -3380,29 +3375,26 @@ struct LiveWorkerProcess { fn failure_for_incomplete_run( pending_control: Option, terminated_message: String, -) -> (WorkflowError, Option) { +) -> (WorkflowError, Option) { if pending_control == Some(RunControlAction::Cancel) { - ( - WorkflowError::Cancelled, - Some(WorkflowStatusReason::Cancelled), - ) + (WorkflowError::Cancelled, Some(StatusReason::Cancelled)) } else { ( WorkflowError::engine(terminated_message), - Some(WorkflowStatusReason::Terminated), + Some(StatusReason::Terminated), ) } } -fn should_reconcile_run_on_startup(status: WorkflowRunStatus) -> bool { +fn should_reconcile_run_on_startup(status: RunStatus) -> bool { matches!( status, - WorkflowRunStatus::Queued - | WorkflowRunStatus::Starting - | WorkflowRunStatus::Running - | WorkflowRunStatus::Blocked - | WorkflowRunStatus::Paused - | WorkflowRunStatus::Removing + RunStatus::Queued + | RunStatus::Starting + | RunStatus::Running + | RunStatus::Blocked + | RunStatus::Paused + | RunStatus::Removing ) } @@ -3562,7 +3554,7 @@ async fn persist_cancelled_run_status(state: &AppState, run_id: RunId) -> anyhow &workflow_event::Event::WorkflowRunFailed { error: WorkflowError::Cancelled, duration_ms: 0, - reason: Some(WorkflowStatusReason::Cancelled), + reason: Some(StatusReason::Cancelled), git_commit_sha: None, final_patch: None, }, @@ -3618,28 +3610,6 @@ fn managed_run( } } -fn api_status_from_workflow(status: WorkflowRunStatus) -> RunStatus { - match status { - WorkflowRunStatus::Submitted => RunStatus::Submitted, - WorkflowRunStatus::Queued => RunStatus::Queued, - WorkflowRunStatus::Starting => RunStatus::Starting, - WorkflowRunStatus::Running => RunStatus::Running, - WorkflowRunStatus::Blocked => RunStatus::Blocked, - WorkflowRunStatus::Paused => RunStatus::Paused, - WorkflowRunStatus::Removing => RunStatus::Removing, - WorkflowRunStatus::Succeeded => RunStatus::Succeeded, - WorkflowRunStatus::Failed => RunStatus::Failed, - WorkflowRunStatus::Dead => RunStatus::Dead, - WorkflowRunStatus::Archived => RunStatus::Archived, - } -} - -fn api_blocked_reason(reason: BlockedReason) -> ApiBlockedReason { - match reason { - BlockedReason::HumanInputRequired => ApiBlockedReason::HumanInputRequired, - } -} - fn worker_mode_arg(mode: RunExecutionMode) -> &'static str { match mode { RunExecutionMode::Start => "start", @@ -3647,43 +3617,19 @@ fn worker_mode_arg(mode: RunExecutionMode) -> &'static str { } } -fn api_status_reason(reason: WorkflowStatusReason) -> ApiStatusReason { - match reason { - WorkflowStatusReason::Completed => ApiStatusReason::Completed, - WorkflowStatusReason::PartialSuccess => ApiStatusReason::PartialSuccess, - WorkflowStatusReason::WorkflowError => ApiStatusReason::WorkflowError, - WorkflowStatusReason::Cancelled => ApiStatusReason::Cancelled, - WorkflowStatusReason::Terminated => ApiStatusReason::Terminated, - WorkflowStatusReason::TransientInfra => ApiStatusReason::TransientInfra, - WorkflowStatusReason::BudgetExhausted => ApiStatusReason::BudgetExhausted, - WorkflowStatusReason::LaunchFailed => ApiStatusReason::LaunchFailed, - WorkflowStatusReason::BootstrapFailed => ApiStatusReason::BootstrapFailed, - WorkflowStatusReason::SandboxInitFailed => ApiStatusReason::SandboxInitFailed, - WorkflowStatusReason::SandboxInitializing => ApiStatusReason::SandboxInitializing, - } -} - -fn api_pending_control(action: RunControlAction) -> ApiRunControlAction { - match action { - RunControlAction::Cancel => ApiRunControlAction::Cancel, - RunControlAction::Pause => ApiRunControlAction::Pause, - RunControlAction::Unpause => ApiRunControlAction::Unpause, - } -} - async fn load_run_status_metadata( state: &AppState, run_id: RunId, ) -> ( - Option, - Option, - Option, + Option, + Option, + Option, ) { match state.store.runs().find(&run_id).await { Ok(Some(summary)) => ( - summary.status_reason.map(api_status_reason), - summary.blocked_reason.map(api_blocked_reason), - summary.pending_control.map(api_pending_control), + summary.status_reason, + summary.blocked_reason, + summary.pending_control, ), _ => (None, None, None), } @@ -4365,7 +4311,7 @@ async fn start_run( } else if let Some(record) = run_state.status.as_ref() { if !matches!( record.status, - WorkflowRunStatus::Submitted | WorkflowRunStatus::Queued | WorkflowRunStatus::Starting + RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting ) { return ApiError::new( StatusCode::CONFLICT, @@ -4792,7 +4738,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { &workflow_event::Event::WorkflowRunFailed { error: WorkflowError::engine(err.to_string()), duration_ms: 0, - reason: Some(WorkflowStatusReason::LaunchFailed), + reason: Some(StatusReason::LaunchFailed), git_commit_sha: None, final_patch: None, }, @@ -4814,7 +4760,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { &workflow_event::Event::WorkflowRunFailed { error: WorkflowError::engine(message.clone()), duration_ms: 0, - reason: Some(WorkflowStatusReason::LaunchFailed), + reason: Some(StatusReason::LaunchFailed), git_commit_sha: None, final_patch: None, }, @@ -4844,7 +4790,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { &workflow_event::Event::WorkflowRunFailed { error: WorkflowError::engine(message.clone()), duration_ms: 0, - reason: Some(WorkflowStatusReason::LaunchFailed), + reason: Some(StatusReason::LaunchFailed), git_commit_sha: None, final_patch: None, }, @@ -4865,7 +4811,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { &workflow_event::Event::WorkflowRunFailed { error: WorkflowError::engine(message.clone()), duration_ms: 0, - reason: Some(WorkflowStatusReason::LaunchFailed), + reason: Some(StatusReason::LaunchFailed), git_commit_sha: None, final_patch: None, }, @@ -4898,7 +4844,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { &workflow_event::Event::WorkflowRunFailed { error: WorkflowError::engine(err.to_string()), duration_ms: 0, - reason: Some(WorkflowStatusReason::Terminated), + reason: Some(StatusReason::Terminated), git_commit_sha: None, final_patch: None, }, @@ -4984,7 +4930,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { let mut runs = state.runs.lock().expect("runs lock poisoned"); if let Some(managed_run) = runs.get_mut(&run_id) { if let Some(status) = final_state.status.as_ref() { - managed_run.status = api_status_from_workflow(status.status); + managed_run.status = status.status; } else if !wait_status.success() { managed_run.status = RunStatus::Failed; } @@ -6306,7 +6252,7 @@ async fn reject_if_archived(state: &AppState, run_id: &RunId) -> Option Response { ) .into_response(); }; - let status = api_status_from_workflow(record.status); - let status_reason = record.status_reason.map(api_status_reason); - let blocked_reason = record.blocked_reason.map(api_blocked_reason); + let status = record.status; + let status_reason = record.status_reason; + let blocked_reason = record.blocked_reason; ( StatusCode::OK, Json(RunStatusResponse { @@ -9988,8 +9934,8 @@ level = "debug" let run_store = state.store.open_run_reader(&run_id).await.unwrap(); let status = run_store.state().await.unwrap().status.unwrap(); - assert_eq!(status.status, WorkflowRunStatus::Failed); - assert_eq!(status.status_reason, Some(WorkflowStatusReason::Cancelled)); + assert_eq!(status.status, RunStatus::Failed); + assert_eq!(status.status_reason, Some(StatusReason::Cancelled)); } #[tokio::test] @@ -10145,7 +10091,7 @@ level = "debug" assert_eq!(body["pending_control"], serde_json::Value::Null); let summary = state.store.runs().find(&run_id).await.unwrap().unwrap(); - assert_eq!(summary.status, WorkflowRunStatus::Paused); + assert_eq!(summary.status, RunStatus::Paused); assert_eq!( summary.blocked_reason, Some(BlockedReason::HumanInputRequired) @@ -10231,7 +10177,7 @@ level = "debug" assert_eq!(body["pending_control"], serde_json::Value::Null); let summary = state.store.runs().find(&run_id).await.unwrap().unwrap(); - assert_eq!(summary.status, WorkflowRunStatus::Blocked); + assert_eq!(summary.status, RunStatus::Blocked); assert_eq!( summary.blocked_reason, Some(BlockedReason::HumanInputRequired) @@ -10282,7 +10228,7 @@ level = "debug" .state() .await .unwrap(); - assert_eq!(run_1.status.unwrap().status, WorkflowRunStatus::Submitted); + assert_eq!(run_1.status.unwrap().status, RunStatus::Submitted); let run_2 = state .store @@ -10293,11 +10239,8 @@ level = "debug" .await .unwrap(); let run_2_status = run_2.status.unwrap(); - assert_eq!(run_2_status.status, WorkflowRunStatus::Failed); - assert_eq!( - run_2_status.status_reason, - Some(WorkflowStatusReason::Terminated) - ); + assert_eq!(run_2_status.status, RunStatus::Failed); + assert_eq!(run_2_status.status_reason, Some(StatusReason::Terminated)); let run_3 = state .store @@ -10308,11 +10251,8 @@ level = "debug" .await .unwrap(); let run_3_status = run_3.status.unwrap(); - assert_eq!(run_3_status.status, WorkflowRunStatus::Failed); - assert_eq!( - run_3_status.status_reason, - Some(WorkflowStatusReason::Cancelled) - ); + assert_eq!(run_3_status.status, RunStatus::Failed); + assert_eq!(run_3_status.status_reason, Some(StatusReason::Cancelled)); assert_eq!(run_3.pending_control, None); } @@ -10383,11 +10323,8 @@ level = "debug" .await .unwrap(); let run_status = run_state.status.unwrap(); - assert_eq!(run_status.status, WorkflowRunStatus::Failed); - assert_eq!( - run_status.status_reason, - Some(WorkflowStatusReason::Terminated) - ); + assert_eq!(run_status.status, RunStatus::Failed); + assert_eq!(run_status.status_reason, Some(StatusReason::Terminated)); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -10473,8 +10410,8 @@ timeout = "30s" let mut status_record = None; for _ in 0..50 { if let Some(record) = run_store.state().await.unwrap().status { - if record.status == WorkflowRunStatus::Failed - && record.status_reason == Some(WorkflowStatusReason::Cancelled) + if record.status == RunStatus::Failed + && record.status_reason == Some(StatusReason::Cancelled) { status_record = Some(record); break; @@ -10484,11 +10421,8 @@ timeout = "30s" } let status_record = status_record.expect("status record should be persisted"); - assert_eq!(status_record.status, WorkflowRunStatus::Failed); - assert_eq!( - status_record.status_reason, - Some(WorkflowStatusReason::Cancelled) - ); + assert_eq!(status_record.status, RunStatus::Failed); + assert_eq!(status_record.status_reason, Some(StatusReason::Cancelled)); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)]