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}" + ); + } } }