From 54666632c8dc752a04381eeeec299b3d1fb5639d Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Wed, 2 Sep 2026 17:29:41 -0400 Subject: [PATCH 1/2] Persist pre-start worker failures --- lib/apps/fabro-server/src/server.rs | 35 ++- lib/apps/fabro-server/src/server/tests.rs | 320 +++++++++++++++++++- lib/components/fabro-store/src/slate/mod.rs | 29 +- lib/foundation/fabro-types/src/status.rs | 59 +++- 4 files changed, 430 insertions(+), 13 deletions(-) diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index bb69cc765..9108be90c 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -3585,18 +3585,45 @@ async fn fail_worker_launch( err: anyhow::Error, ) { tracing::error!(run_id = %run_id, error = %err, "Failed to spawn worker"); - let message = format!("Failed to spawn worker: {err}"); + let cancellation_pending = match run_store.state().await { + Ok(run_state) => run_state.pending_control == Some(RunControlAction::Cancel), + Err(state_err) => { + tracing::warn!( + run_id = %run_id, + error = %render_compact_with_causes( + &state_err.to_string(), + &collect_causes(&state_err), + ), + "Failed to load run state while recording worker launch failure" + ); + false + } + }; + let (error, reason, message) = if cancellation_pending { + ( + WorkflowError::Cancelled, + FailureReason::Cancelled, + "Run cancelled before worker launch completed".to_string(), + ) + } else { + let message = format!("Failed to spawn worker: {err}"); + ( + WorkflowError::engine_with_anyhow("Failed to spawn worker", err), + FailureReason::LaunchFailed, + message, + ) + }; let failure_event = workflow_event::Event::workflow_run_failed_from_error( - &WorkflowError::engine_with_anyhow("Failed to spawn worker", err), + &error, fabro_types::RunTiming::default(), - FailureReason::LaunchFailed, + reason, None, None, None, None, ); let _ = workflow_event::append_event(run_store, &run_id, &failure_event).await; - fail_managed_run(state, run_id, FailureReason::LaunchFailed, message); + fail_managed_run(state, run_id, reason, message); state.scheduler_notify.notify_one(); } diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 7f23d1989..7f939857e 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -4,7 +4,7 @@ use std::os::unix::fs::PermissionsExt; use std::path::{Path, PathBuf}; #[cfg(unix)] use std::process::Stdio; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc as StdArc, Mutex as StdMutex}; use async_zip::base::read::mem::ZipFileReader; @@ -58,7 +58,7 @@ use crate::worker_control::{ LocalWorkerControlBus, WorkerControlBus, WorkerControlCursor, WorkerControlReceiver, }; use crate::worker_runtime::{ - LocalWorkerRuntime, StartedWorker, WorkerLaunchSpec, WorkerRef, WorkerRuntime, + LocalWorkerRuntime, StartedWorker, WorkerExit, WorkerLaunchSpec, WorkerRef, WorkerRuntime, }; const MINIMAL_DOT: &str = r#"digraph Test { @@ -2982,6 +2982,96 @@ impl RecordingWorkerRuntime { } } +#[derive(Clone, Copy)] +enum PreStartWorkerOutcome { + LaunchFailure, + EarlyExit, +} + +struct PreStartWorkerRuntime { + outcome: PreStartWorkerOutcome, + starts: AtomicUsize, + hold_launch_failure: bool, + start_entered: Notify, + release_start: Notify, +} + +impl PreStartWorkerRuntime { + fn new(outcome: PreStartWorkerOutcome) -> Self { + Self { + outcome, + starts: AtomicUsize::new(0), + hold_launch_failure: false, + start_entered: Notify::new(), + release_start: Notify::new(), + } + } + + fn held_launch_failure() -> Self { + Self { + hold_launch_failure: true, + ..Self::new(PreStartWorkerOutcome::LaunchFailure) + } + } + + fn start_count(&self) -> usize { + self.starts.load(Ordering::Relaxed) + } + + async fn wait_for_start(&self) { + tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + let notified = self.start_entered.notified(); + if self.start_count() > 0 { + return; + } + notified.await; + } + }) + .await + .expect("test worker runtime should receive one start request"); + } + + fn release_held_start(&self) { + self.release_start.notify_one(); + } +} + +#[async_trait::async_trait] +impl WorkerRuntime for PreStartWorkerRuntime { + async fn start(&self, _spec: WorkerLaunchSpec) -> anyhow::Result { + self.starts.fetch_add(1, Ordering::Relaxed); + self.start_entered.notify_waiters(); + if self.hold_launch_failure { + self.release_start.notified().await; + } + + match self.outcome { + PreStartWorkerOutcome::LaunchFailure => { + anyhow::bail!("test worker launch failed") + } + PreStartWorkerOutcome::EarlyExit => Ok(StartedWorker { + worker_ref: test_worker_ref(u32::MAX), + stderr: Box::pin(tokio::io::empty()), + wait: Box::pin(async { + Ok(WorkerExit { + success: false, + detail: "test worker exited before starting".to_string(), + }) + }), + }), + } + } + + async fn request_stop(&self, _worker_ref: &WorkerRef) {} + + async fn force_stop(&self, _worker_ref: &WorkerRef) {} + + async fn is_alive(&self, _worker_ref: &WorkerRef) -> bool { + false + } +} + #[async_trait::async_trait] impl WorkerRuntime for RecordingWorkerRuntime { async fn start(&self, _spec: WorkerLaunchSpec) -> anyhow::Result { @@ -5700,6 +5790,232 @@ async fn create_and_start_run(app: &Router, dot_source: &str) -> String { run_id } +fn subprocess_pre_start_failure_state(runtime: StdArc) -> Arc { + let state = TestAppStateBuilder::new() + .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")]) + .worker_runtime(runtime) + .build(); + let runtime_directory = Storage::new(state.server_storage_dir()).runtime_directory(); + ServerDaemon::new( + std::process::id(), + Bind::Tcp("127.0.0.1:32276".parse().expect("test bind should parse")), + runtime_directory.log_path(), + ) + .write(&runtime_directory) + .expect("test server record should be written"); + state +} + +async fn assert_subprocess_pre_start_failure( + outcome: PreStartWorkerOutcome, + expected_reason: FailureReason, +) { + let runtime = StdArc::new(PreStartWorkerRuntime::new(outcome)); + let state = subprocess_pre_start_failure_state(StdArc::clone(&runtime)); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = create_and_start_run(&app, MINIMAL_DOT) + .await + .parse::() + .expect("created run id should parse"); + let run_store = state + .stores + .runs + .open_run_reader(&run_id) + .await + .expect("created run should remain readable"); + + assert_eq!( + run_store + .state() + .await + .expect("runnable run state should load") + .status, + RunStatus::Runnable + ); + + execute_run(Arc::clone(&state), run_id).await; + + assert_eq!(runtime.start_count(), 1); + let events = run_store + .list_events() + .await + .expect("failed run events should remain readable"); + let lifecycle_events = events + .iter() + .map(|envelope| envelope.event.event_name()) + .filter(|name| matches!(*name, "run.runnable" | "run.starting" | "run.failed")) + .collect::>(); + assert_eq!(lifecycle_events, vec!["run.runnable", "run.failed"]); + let failure_reasons = events + .iter() + .filter_map(|envelope| match &envelope.event.body { + EventBody::RunFailed(props) => Some(props.failure.reason), + _ => None, + }) + .collect::>(); + assert_eq!(failure_reasons, vec![expected_reason]); + + let expected_status = RunStatus::Failed { + reason: expected_reason, + }; + assert_eq!( + run_store + .state() + .await + .expect("failed run state should load") + .status, + expected_status + ); + assert_eq!( + state + .runs + .lock() + .expect("runs lock poisoned") + .get(&run_id) + .expect("managed run should remain present") + .status, + expected_status + ); + + let response = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}"))) + .body(Body::empty()) + .expect("run request should build"), + ) + .await + .expect("run request should complete"); + let body = response_json!(response, StatusCode::OK).await; + assert_eq!(run_json_status(&body)["kind"], "failed"); + assert_eq!( + run_json_status(&body)["reason"], + expected_reason.to_string() + ); + + let response = app + .oneshot( + Request::builder() + .method("GET") + .uri(api("/runs")) + .body(Body::empty()) + .expect("run list request should build"), + ) + .await + .expect("run list request should complete"); + let body = response_json!(response, StatusCode::OK).await; + let run_id_string = run_id.to_string(); + let listed = body["data"] + .as_array() + .expect("run list data should be an array") + .iter() + .find(|run| run_json_id(run) == Some(run_id_string.as_str())) + .expect("failed run should remain listed"); + assert_eq!(run_json_status(listed)["kind"], "failed"); + assert_eq!( + run_json_status(listed)["reason"], + expected_reason.to_string() + ); +} + +#[tokio::test] +async fn subprocess_pre_start_failure_persists_launch_failure_from_runnable() { + assert_subprocess_pre_start_failure( + PreStartWorkerOutcome::LaunchFailure, + FailureReason::LaunchFailed, + ) + .await; +} + +#[tokio::test] +async fn subprocess_pre_start_failure_persists_early_worker_exit_from_runnable() { + assert_subprocess_pre_start_failure( + PreStartWorkerOutcome::EarlyExit, + FailureReason::Terminated, + ) + .await; +} + +#[tokio::test] +async fn subprocess_pre_start_failure_preserves_pending_cancellation() { + let runtime = StdArc::new(PreStartWorkerRuntime::held_launch_failure()); + let state = subprocess_pre_start_failure_state(StdArc::clone(&runtime)); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = create_and_start_run(&app, MINIMAL_DOT) + .await + .parse::() + .expect("created run id should parse"); + + let execution = tokio::spawn(execute_run(Arc::clone(&state), run_id)); + runtime.wait_for_start().await; + + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/cancel"))) + .body(Body::empty()) + .expect("cancel request should build"), + ) + .await + .expect("cancel request should complete"); + assert_status!(response, StatusCode::ACCEPTED).await; + + runtime.release_held_start(); + execution.await.expect("run execution task should complete"); + + assert_eq!(runtime.start_count(), 1); + let run_store = state + .stores + .runs + .open_run_reader(&run_id) + .await + .expect("cancelled run should remain readable"); + let events = run_store + .list_events() + .await + .expect("cancelled run events should remain readable"); + let failure_reasons = events + .iter() + .filter_map(|envelope| match &envelope.event.body { + EventBody::RunFailed(props) => Some(props.failure.reason), + _ => None, + }) + .collect::>(); + assert_eq!(failure_reasons, vec![FailureReason::Cancelled]); + assert!(!events.iter().any(|envelope| { + matches!( + &envelope.event.body, + EventBody::RunFailed(props) + if props.failure.reason == FailureReason::LaunchFailed + ) + })); + + let expected_status = RunStatus::Failed { + reason: FailureReason::Cancelled, + }; + assert_eq!( + run_store + .state() + .await + .expect("cancelled run state should load") + .status, + expected_status + ); + assert_eq!( + state + .runs + .lock() + .expect("runs lock poisoned") + .get(&run_id) + .expect("managed run should remain present") + .status, + expected_status + ); +} + async fn create_durable_run_with_events( state: &Arc, run_id: RunId, diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 5678ee954..ebd6a5465 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -634,6 +634,29 @@ mod tests { ) } + fn sandbox_init_failure_payload(label: &str) -> EventPayload { + event_payload( + label, + "2026-03-27T12:00:04Z", + "run.failed", + &serde_json::json!({ + "failure": { + "reason": "sandbox_init_failed", + "detail": { + "message": "sandbox initialization failed", + "category": "deterministic" + } + }, + "timing": { + "wall_time_ms": 1, + "inference_time_ms": 0, + "tool_time_ms": 0, + "active_time_ms": 0 + }, + }), + ) + } + async fn append_completed(run: &RunDatabase, label: &str, created_at: DateTime) { append_running(run, label, created_at).await; run.append_event(&event_payload( @@ -977,7 +1000,7 @@ mod tests { let events_before = run.list_events().await.unwrap(); let err = run - .append_event(&workflow_failure_payload("run-1")) + .append_event(&sandbox_init_failure_payload("run-1")) .await .unwrap_err(); @@ -989,7 +1012,7 @@ mod tests { Error::InvalidTransition(fabro_types::InvalidTransition { from: RunStatus::Runnable, to: RunStatus::Failed { - reason: FailureReason::WorkflowError, + reason: FailureReason::SandboxInitFailed, }, }) )); @@ -1009,7 +1032,7 @@ mod tests { append_runnable(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; let err = run - .append_event(&workflow_failure_payload("run-1")) + .append_event(&sandbox_init_failure_payload("run-1")) .await .unwrap_err(); assert!(matches!(err, Error::EventRejected { .. })); diff --git a/lib/foundation/fabro-types/src/status.rs b/lib/foundation/fabro-types/src/status.rs index 2224d0708..d423078b3 100644 --- a/lib/foundation/fabro-types/src/status.rs +++ b/lib/foundation/fabro-types/src/status.rs @@ -157,7 +157,18 @@ impl RunStatus { Self::Submitted ) | (Self::Pending { .. }, Self::Runnable) - | (Self::Runnable, Self::Starting) + // A worker can fail before it appends RunStarting, while the durable run is still + // Runnable. Keep this set aligned with the audited pre-Starting failure callers. + | ( + Self::Runnable, + Self::Starting + | Self::Failed { + reason: FailureReason::LaunchFailed + | FailureReason::WorkflowError + | FailureReason::Terminated + | FailureReason::BootstrapFailed, + } + ) | ( Self::Submitted | Self::Pending { .. } | Self::Runnable, Self::Failed { @@ -417,9 +428,6 @@ mod tests { assert!(runnable.can_transition_to(RunStatus::Failed { reason: FailureReason::Cancelled, })); - assert!(!runnable.can_transition_to(RunStatus::Failed { - reason: FailureReason::Terminated, - })); assert!(running.can_transition_to(blocked)); assert!(blocked.can_transition_to(running)); assert!(blocked.can_transition_to(paused)); @@ -428,6 +436,49 @@ mod tests { })); } + #[test] + fn runnable_pre_start_failures_allow_only_audited_reasons() { + for reason in [ + FailureReason::LaunchFailed, + FailureReason::WorkflowError, + FailureReason::Terminated, + FailureReason::BootstrapFailed, + ] { + let failed = RunStatus::Failed { reason }; + assert!( + RunStatus::Runnable.can_transition_to(failed), + "Runnable should accept {reason} before Starting" + ); + assert!( + !RunStatus::Submitted.can_transition_to(failed), + "Submitted should reject {reason}" + ); + assert!( + !RunStatus::Pending { + reason: PendingReason::ApprovalRequired, + } + .can_transition_to(failed), + "Pending should reject {reason}" + ); + } + + assert!(RunStatus::Runnable.can_transition_to(RunStatus::Failed { + reason: FailureReason::Cancelled, + })); + for reason in [ + FailureReason::SandboxInitFailed, + FailureReason::PublishFailed, + FailureReason::TransientInfra, + FailureReason::BudgetExhausted, + FailureReason::ApprovalDenied, + ] { + assert!( + !RunStatus::Runnable.can_transition_to(RunStatus::Failed { reason }), + "Runnable should reject unrelated failure {reason}" + ); + } + } + #[test] fn success_and_failure_reasons_parse_and_round_trip() { let success = SuccessReason::from_str("completed").expect("completed should parse"); From c437dc012d65433431744eb09c5a15b606e0c566 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Thu, 3 Sep 2026 11:45:47 -0400 Subject: [PATCH 2/2] Simplify pre-start worker failure handling Move the "which failures can happen before Starting" classification onto FailureReason as an exhaustive predicate and use it for every Runnable -> Failed transition, replacing the hand-maintained allowlist. Give the pending-cancel precedence rule a single owner shared by the worker launch and worker exit paths. Test cleanups: share the Notify wait loop, server record fixture, and post-failure assertions; simplify the pre-start test runtime's hold flag; and parameterize the slate run.failed payload helper. Co-Authored-By: Claude Fable 5.1 --- lib/apps/fabro-server/src/server.rs | 48 ++-- lib/apps/fabro-server/src/server/tests.rs | 256 ++++++++++---------- lib/components/fabro-store/src/slate/mod.rs | 34 +-- lib/foundation/fabro-types/src/status.rs | 128 ++++++---- 4 files changed, 241 insertions(+), 225 deletions(-) diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 9108be90c..11f02947d 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -3060,18 +3060,31 @@ struct LiveWorkerProcess { worker_ref: WorkerRef, } -fn failure_for_incomplete_run( +/// Pick the terminal failure for a run that never produced its own terminal +/// event. A pending cancel wins over whatever failure the caller observed, so a +/// run that was cancelled while its worker was launching or dying is recorded +/// as cancelled rather than as broken. +fn failure_honoring_pending_cancel( pending_control: Option, - terminated_message: String, + otherwise: impl FnOnce() -> (WorkflowError, FailureReason), ) -> (WorkflowError, FailureReason) { if pending_control == Some(RunControlAction::Cancel) { (WorkflowError::Cancelled, FailureReason::Cancelled) } else { + otherwise() + } +} + +fn failure_for_incomplete_run( + pending_control: Option, + terminated_message: String, +) -> (WorkflowError, FailureReason) { + failure_honoring_pending_cancel(pending_control, || { ( WorkflowError::engine(terminated_message), FailureReason::Terminated, ) - } + }) } pub(crate) async fn reconcile_incomplete_runs_on_startup( @@ -3585,33 +3598,28 @@ async fn fail_worker_launch( err: anyhow::Error, ) { tracing::error!(run_id = %run_id, error = %err, "Failed to spawn worker"); - let cancellation_pending = match run_store.state().await { - Ok(run_state) => run_state.pending_control == Some(RunControlAction::Cancel), + let pending_control = match run_store.state().await { + Ok(run_state) => run_state.pending_control, Err(state_err) => { tracing::warn!( run_id = %run_id, - error = %render_compact_with_causes( - &state_err.to_string(), - &collect_causes(&state_err), - ), - "Failed to load run state while recording worker launch failure" + error = %state_err, + "Failed to load run state after worker launch failure" ); - false + None } }; - let (error, reason, message) = if cancellation_pending { - ( - WorkflowError::Cancelled, - FailureReason::Cancelled, - "Run cancelled before worker launch completed".to_string(), - ) - } else { - let message = format!("Failed to spawn worker: {err}"); + let launch_message = format!("Failed to spawn worker: {err}"); + let (error, reason) = failure_honoring_pending_cancel(pending_control, || { ( WorkflowError::engine_with_anyhow("Failed to spawn worker", err), FailureReason::LaunchFailed, - message, ) + }); + let message = if reason == FailureReason::Cancelled { + "Run cancelled before worker launch completed".to_string() + } else { + launch_message }; let failure_event = workflow_event::Event::workflow_run_failed_from_error( &error, diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 7f939857e..e5d213898 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -2859,14 +2859,7 @@ allowed_usernames = ["octocat"] .collect::>() .join(", ") ); - let runtime_directory = Storage::new(storage_dir).runtime_directory(); - ServerDaemon::new( - std::process::id(), - Bind::Tcp("127.0.0.1:32276".parse::().unwrap()), - runtime_directory.log_path(), - ) - .write(&runtime_directory) - .unwrap(); + write_test_server_record(storage_dir); let mut server_secret_env: HashMap = dev_token .map(|token| HashMap::from([("FABRO_DEV_TOKEN".to_string(), token)])) @@ -2943,6 +2936,37 @@ fn worker_token_claims(cmd: &Command, state: &AppState) -> crate::worker_token:: .claims } +fn write_test_server_record(storage_dir: &Path) { + let runtime_directory = Storage::new(storage_dir).runtime_directory(); + ServerDaemon::new( + std::process::id(), + Bind::Tcp( + "127.0.0.1:32276" + .parse::() + .expect("test bind should parse"), + ), + runtime_directory.log_path(), + ) + .write(&runtime_directory) + .expect("test server record should be written"); +} + +/// Waits up to one second for `condition` to hold, re-checking whenever +/// `notify` fires. +async fn wait_until(notify: &Notify, condition: impl Fn() -> bool, expectation: &str) { + tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + let notified = notify.notified(); + if condition() { + return; + } + notified.await; + } + }) + .await + .expect(expectation); +} + #[derive(Default)] struct RecordingWorkerRuntime { requested: StdMutex>, @@ -2968,17 +2992,12 @@ impl RecordingWorkerRuntime { } async fn wait_for_forced_ref(&self, worker_ref: &WorkerRef) { - tokio::time::timeout(std::time::Duration::from_secs(1), async { - loop { - let notified = self.forced_notify.notified(); - if self.forced_refs().contains(worker_ref) { - return; - } - notified.await; - } - }) - .await - .expect("worker should be force-stopped after the cancellation grace period"); + wait_until( + &self.forced_notify, + || self.forced_refs().contains(worker_ref), + "worker should be force-stopped after the cancellation grace period", + ) + .await; } } @@ -2988,12 +3007,15 @@ enum PreStartWorkerOutcome { EarlyExit, } +/// Test worker runtime whose `start` fails before the worker reaches +/// `Starting`, either by refusing to launch or by exiting immediately. When +/// built with `held`, `start` blocks until `release_held_start` so a test can +/// act while the launch is in flight. struct PreStartWorkerRuntime { - outcome: PreStartWorkerOutcome, - starts: AtomicUsize, - hold_launch_failure: bool, - start_entered: Notify, - release_start: Notify, + outcome: PreStartWorkerOutcome, + starts: AtomicUsize, + start_entered: Notify, + release_start: Option, } impl PreStartWorkerRuntime { @@ -3001,16 +3023,15 @@ impl PreStartWorkerRuntime { Self { outcome, starts: AtomicUsize::new(0), - hold_launch_failure: false, start_entered: Notify::new(), - release_start: Notify::new(), + release_start: None, } } - fn held_launch_failure() -> Self { + fn held(outcome: PreStartWorkerOutcome) -> Self { Self { - hold_launch_failure: true, - ..Self::new(PreStartWorkerOutcome::LaunchFailure) + release_start: Some(Notify::new()), + ..Self::new(outcome) } } @@ -3019,21 +3040,19 @@ impl PreStartWorkerRuntime { } async fn wait_for_start(&self) { - tokio::time::timeout(std::time::Duration::from_secs(1), async { - loop { - let notified = self.start_entered.notified(); - if self.start_count() > 0 { - return; - } - notified.await; - } - }) - .await - .expect("test worker runtime should receive one start request"); + wait_until( + &self.start_entered, + || self.start_count() > 0, + "test worker runtime should receive one start request", + ) + .await; } fn release_held_start(&self) { - self.release_start.notify_one(); + self.release_start + .as_ref() + .expect("runtime should have been built with a held start") + .notify_one(); } } @@ -3042,8 +3061,8 @@ impl WorkerRuntime for PreStartWorkerRuntime { async fn start(&self, _spec: WorkerLaunchSpec) -> anyhow::Result { self.starts.fetch_add(1, Ordering::Relaxed); self.start_entered.notify_waiters(); - if self.hold_launch_failure { - self.release_start.notified().await; + if let Some(release_start) = &self.release_start { + release_start.notified().await; } match self.outcome { @@ -5795,17 +5814,65 @@ fn subprocess_pre_start_failure_state(runtime: StdArc) -> .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")]) .worker_runtime(runtime) .build(); - let runtime_directory = Storage::new(state.server_storage_dir()).runtime_directory(); - ServerDaemon::new( - std::process::id(), - Bind::Tcp("127.0.0.1:32276".parse().expect("test bind should parse")), - runtime_directory.log_path(), - ) - .write(&runtime_directory) - .expect("test server record should be written"); + write_test_server_record(&state.server_storage_dir()); state } +fn run_failed_reasons(events: &[EventEnvelope]) -> Vec { + events + .iter() + .filter_map(|envelope| match &envelope.event.body { + EventBody::RunFailed(props) => Some(props.failure.reason), + _ => None, + }) + .collect() +} + +/// Asserts that a run which failed before its worker reached `Starting` +/// recorded exactly one `run.failed` event with `expected_reason` and that the +/// durable and in-memory statuses agree. Returns the run's events for further +/// inspection. +async fn assert_run_failed_before_start( + state: &Arc, + run_id: RunId, + expected_reason: FailureReason, +) -> Vec { + let run_store = state + .stores + .runs + .open_run_reader(&run_id) + .await + .expect("failed run should remain readable"); + let events = run_store + .list_events() + .await + .expect("failed run events should remain readable"); + assert_eq!(run_failed_reasons(&events), vec![expected_reason]); + + let expected_status = RunStatus::Failed { + reason: expected_reason, + }; + assert_eq!( + run_store + .state() + .await + .expect("failed run state should load") + .status, + expected_status + ); + assert_eq!( + state + .runs + .lock() + .expect("runs lock poisoned") + .get(&run_id) + .expect("managed run should remain present") + .status, + expected_status + ); + events +} + async fn assert_subprocess_pre_start_failure( outcome: PreStartWorkerOutcome, expected_reason: FailureReason, @@ -5836,46 +5903,13 @@ async fn assert_subprocess_pre_start_failure( execute_run(Arc::clone(&state), run_id).await; assert_eq!(runtime.start_count(), 1); - let events = run_store - .list_events() - .await - .expect("failed run events should remain readable"); + let events = assert_run_failed_before_start(&state, run_id, expected_reason).await; let lifecycle_events = events .iter() .map(|envelope| envelope.event.event_name()) .filter(|name| matches!(*name, "run.runnable" | "run.starting" | "run.failed")) .collect::>(); assert_eq!(lifecycle_events, vec!["run.runnable", "run.failed"]); - let failure_reasons = events - .iter() - .filter_map(|envelope| match &envelope.event.body { - EventBody::RunFailed(props) => Some(props.failure.reason), - _ => None, - }) - .collect::>(); - assert_eq!(failure_reasons, vec![expected_reason]); - - let expected_status = RunStatus::Failed { - reason: expected_reason, - }; - assert_eq!( - run_store - .state() - .await - .expect("failed run state should load") - .status, - expected_status - ); - assert_eq!( - state - .runs - .lock() - .expect("runs lock poisoned") - .get(&run_id) - .expect("managed run should remain present") - .status, - expected_status - ); let response = app .clone() @@ -5940,7 +5974,9 @@ async fn subprocess_pre_start_failure_persists_early_worker_exit_from_runnable() #[tokio::test] async fn subprocess_pre_start_failure_preserves_pending_cancellation() { - let runtime = StdArc::new(PreStartWorkerRuntime::held_launch_failure()); + let runtime = StdArc::new(PreStartWorkerRuntime::held( + PreStartWorkerOutcome::LaunchFailure, + )); let state = subprocess_pre_start_failure_state(StdArc::clone(&runtime)); let app = crate::test_support::build_test_router(Arc::clone(&state)); let run_id = create_and_start_run(&app, MINIMAL_DOT) @@ -5967,53 +6003,7 @@ async fn subprocess_pre_start_failure_preserves_pending_cancellation() { execution.await.expect("run execution task should complete"); assert_eq!(runtime.start_count(), 1); - let run_store = state - .stores - .runs - .open_run_reader(&run_id) - .await - .expect("cancelled run should remain readable"); - let events = run_store - .list_events() - .await - .expect("cancelled run events should remain readable"); - let failure_reasons = events - .iter() - .filter_map(|envelope| match &envelope.event.body { - EventBody::RunFailed(props) => Some(props.failure.reason), - _ => None, - }) - .collect::>(); - assert_eq!(failure_reasons, vec![FailureReason::Cancelled]); - assert!(!events.iter().any(|envelope| { - matches!( - &envelope.event.body, - EventBody::RunFailed(props) - if props.failure.reason == FailureReason::LaunchFailed - ) - })); - - let expected_status = RunStatus::Failed { - reason: FailureReason::Cancelled, - }; - assert_eq!( - run_store - .state() - .await - .expect("cancelled run state should load") - .status, - expected_status - ); - assert_eq!( - state - .runs - .lock() - .expect("runs lock poisoned") - .get(&run_id) - .expect("managed run should remain present") - .status, - expected_status - ); + assert_run_failed_before_start(&state, run_id, FailureReason::Cancelled).await; } async fn create_durable_run_with_events( diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index ebd6a5465..d8df285ba 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -611,16 +611,16 @@ mod tests { .unwrap(); } - fn workflow_failure_payload(label: &str) -> EventPayload { + fn failure_payload(label: &str, reason: FailureReason, message: &str) -> EventPayload { event_payload( label, "2026-03-27T12:00:04Z", "run.failed", &serde_json::json!({ "failure": { - "reason": "workflow_error", + "reason": reason.to_string(), "detail": { - "message": "workflow failed", + "message": message, "category": "deterministic" } }, @@ -634,26 +634,18 @@ mod tests { ) } + fn workflow_failure_payload(label: &str) -> EventPayload { + failure_payload(label, FailureReason::WorkflowError, "workflow failed") + } + + /// A failure that can only occur after `Starting`, so a `Runnable` run must + /// reject it. fn sandbox_init_failure_payload(label: &str) -> EventPayload { - event_payload( + assert!(!FailureReason::SandboxInitFailed.can_occur_before_start()); + failure_payload( label, - "2026-03-27T12:00:04Z", - "run.failed", - &serde_json::json!({ - "failure": { - "reason": "sandbox_init_failed", - "detail": { - "message": "sandbox initialization failed", - "category": "deterministic" - } - }, - "timing": { - "wall_time_ms": 1, - "inference_time_ms": 0, - "tool_time_ms": 0, - "active_time_ms": 0 - }, - }), + FailureReason::SandboxInitFailed, + "sandbox initialization failed", ) } diff --git a/lib/foundation/fabro-types/src/status.rs b/lib/foundation/fabro-types/src/status.rs index d423078b3..31eb62d86 100644 --- a/lib/foundation/fabro-types/src/status.rs +++ b/lib/foundation/fabro-types/src/status.rs @@ -142,6 +142,12 @@ impl RunStatus { if self.is_immutable() { return false; } + // A worker can fail before it appends `RunStarting`, while the durable run is + // still Runnable. This guards live appends only: replay synthesizes the + // missing intermediate statuses instead of rejecting the event. + if let (Self::Runnable, Self::Failed { reason }) = (self, to) { + return reason.can_occur_before_start(); + } matches!( (self, to), (Self::Submitted, Self::Pending { .. } | Self::Runnable) @@ -157,24 +163,10 @@ impl RunStatus { Self::Submitted ) | (Self::Pending { .. }, Self::Runnable) - // A worker can fail before it appends RunStarting, while the durable run is still - // Runnable. Keep this set aligned with the audited pre-Starting failure callers. - | ( - Self::Runnable, - Self::Starting - | Self::Failed { - reason: FailureReason::LaunchFailed - | FailureReason::WorkflowError - | FailureReason::Terminated - | FailureReason::BootstrapFailed, - } - ) - | ( - Self::Submitted | Self::Pending { .. } | Self::Runnable, - Self::Failed { - reason: FailureReason::Cancelled, - } - ) + | (Self::Runnable, Self::Starting) + | (Self::Submitted | Self::Pending { .. }, Self::Failed { + reason: FailureReason::Cancelled, + }) | (Self::Pending { .. }, Self::Failed { reason: FailureReason::ApprovalDenied, }) @@ -295,7 +287,17 @@ pub enum SuccessReason { } #[derive( - Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Display, EnumString, IntoStaticStr, + Debug, + Clone, + Copy, + PartialEq, + Eq, + Serialize, + Deserialize, + Display, + EnumString, + IntoStaticStr, + VariantArray, )] #[serde(rename_all = "snake_case")] #[strum(serialize_all = "snake_case")] @@ -312,6 +314,32 @@ pub enum FailureReason { SandboxInitFailed, } +impl FailureReason { + /// Whether a run can fail for this reason before its worker reaches + /// `Starting`. + /// + /// Launch, bootstrap, and engine failures, a worker dying early, and + /// cancellation all happen before the worker appends `RunStarting`. + /// Every other reason implies the run already progressed past + /// `Starting` (sandbox init, publish, budget, transient infra) or + /// belongs to the approval flow. + #[must_use] + pub fn can_occur_before_start(self) -> bool { + match self { + Self::Cancelled + | Self::LaunchFailed + | Self::WorkflowError + | Self::Terminated + | Self::BootstrapFailed => true, + Self::PublishFailed + | Self::ApprovalDenied + | Self::TransientInfra + | Self::BudgetExhausted + | Self::SandboxInitFailed => false, + } + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum TerminalStatus { @@ -360,6 +388,8 @@ pub enum RunControlAction { mod tests { use std::str::FromStr; + use strum::VariantArray; + use super::{ BlockedReason, FailureReason, InvalidTransition, PendingReason, RunStatus, SuccessReason, }; @@ -437,45 +467,41 @@ mod tests { } #[test] - fn runnable_pre_start_failures_allow_only_audited_reasons() { - for reason in [ + fn runnable_accepts_only_failures_that_can_occur_before_start() { + let pre_start = [ + FailureReason::Cancelled, FailureReason::LaunchFailed, FailureReason::WorkflowError, FailureReason::Terminated, FailureReason::BootstrapFailed, - ] { + ]; + for reason in FailureReason::VARIANTS.iter().copied() { + let expected = pre_start.contains(&reason); + assert_eq!( + reason.can_occur_before_start(), + expected, + "{reason} pre-start classification" + ); let failed = RunStatus::Failed { reason }; - assert!( + assert_eq!( RunStatus::Runnable.can_transition_to(failed), - "Runnable should accept {reason} before Starting" - ); - assert!( - !RunStatus::Submitted.can_transition_to(failed), - "Submitted should reject {reason}" - ); - assert!( - !RunStatus::Pending { - reason: PendingReason::ApprovalRequired, - } - .can_transition_to(failed), - "Pending should reject {reason}" - ); - } - - assert!(RunStatus::Runnable.can_transition_to(RunStatus::Failed { - reason: FailureReason::Cancelled, - })); - for reason in [ - FailureReason::SandboxInitFailed, - FailureReason::PublishFailed, - FailureReason::TransientInfra, - FailureReason::BudgetExhausted, - FailureReason::ApprovalDenied, - ] { - assert!( - !RunStatus::Runnable.can_transition_to(RunStatus::Failed { reason }), - "Runnable should reject unrelated failure {reason}" + expected, + "Runnable -> {reason}" ); + if reason != FailureReason::Cancelled { + assert!( + !RunStatus::Submitted.can_transition_to(failed), + "Submitted should reject {reason}" + ); + assert_eq!( + RunStatus::Pending { + reason: PendingReason::ApprovalRequired, + } + .can_transition_to(failed), + reason == FailureReason::ApprovalDenied, + "Pending -> {reason}" + ); + } } }