diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index bb69cc765..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,18 +3598,40 @@ 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 pending_control = match run_store.state().await { + Ok(run_state) => run_state.pending_control, + Err(state_err) => { + tracing::warn!( + run_id = %run_id, + error = %state_err, + "Failed to load run state after worker launch failure" + ); + None + } + }; + 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, + ) + }); + 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( - &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 c70c92486..9cea921ec 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 { @@ -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,102 @@ 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; + wait_until( + &self.forced_notify, + || self.forced_refs().contains(worker_ref), + "worker should be force-stopped after the cancellation grace period", + ) + .await; + } +} + +#[derive(Clone, Copy)] +enum PreStartWorkerOutcome { + LaunchFailure, + 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, + start_entered: Notify, + release_start: Option, +} + +impl PreStartWorkerRuntime { + fn new(outcome: PreStartWorkerOutcome) -> Self { + Self { + outcome, + starts: AtomicUsize::new(0), + start_entered: Notify::new(), + release_start: None, + } + } + + fn held(outcome: PreStartWorkerOutcome) -> Self { + Self { + release_start: Some(Notify::new()), + ..Self::new(outcome) + } + } + + fn start_count(&self) -> usize { + self.starts.load(Ordering::Relaxed) + } + + async fn wait_for_start(&self) { + 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 + .as_ref() + .expect("runtime should have been built with a held 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 let Some(release_start) = &self.release_start { + release_start.notified().await; + } + + match self.outcome { + PreStartWorkerOutcome::LaunchFailure => { + anyhow::bail!("test worker launch failed") } - }) - .await - .expect("worker should be force-stopped after the cancellation grace period"); + 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 } } @@ -5795,6 +5904,203 @@ 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(); + 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, +) { + 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 = 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 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( + 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) + .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); + assert_run_failed_before_start(&state, run_id, FailureReason::Cancelled).await; +} + 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 1c1857666..663f72bc2 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -599,16 +599,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" } }, @@ -622,6 +622,21 @@ 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 { + assert!(!FailureReason::SandboxInitFailed.can_occur_before_start()); + failure_payload( + label, + FailureReason::SandboxInitFailed, + "sandbox initialization failed", + ) + } + async fn append_completed(run: &RunDatabase, label: &str, created_at: DateTime) { append_running(run, label, created_at).await; run.append_event(&event_payload( @@ -980,7 +995,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(); @@ -992,7 +1007,7 @@ mod tests { Error::InvalidTransition(fabro_types::InvalidTransition { from: RunStatus::Runnable, to: RunStatus::Failed { - reason: FailureReason::WorkflowError, + reason: FailureReason::SandboxInitFailed, }, }) )); @@ -1012,7 +1027,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..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) @@ -158,12 +164,9 @@ impl RunStatus { ) | (Self::Pending { .. }, Self::Runnable) | (Self::Runnable, Self::Starting) - | ( - Self::Submitted | Self::Pending { .. } | Self::Runnable, - Self::Failed { - reason: FailureReason::Cancelled, - } - ) + | (Self::Submitted | Self::Pending { .. }, Self::Failed { + reason: FailureReason::Cancelled, + }) | (Self::Pending { .. }, Self::Failed { reason: FailureReason::ApprovalDenied, }) @@ -284,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")] @@ -301,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 { @@ -349,6 +388,8 @@ pub enum RunControlAction { mod tests { use std::str::FromStr; + use strum::VariantArray; + use super::{ BlockedReason, FailureReason, InvalidTransition, PendingReason, RunStatus, SuccessReason, }; @@ -417,9 +458,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 +466,45 @@ mod tests { })); } + #[test] + 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_eq!( + RunStatus::Runnable.can_transition_to(failed), + 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}" + ); + } + } + } + #[test] fn success_and_failure_reasons_parse_and_round_trip() { let success = SuccessReason::from_str("completed").expect("completed should parse");