diff --git a/lib/apps/fabro-server/src/run_files.rs b/lib/apps/fabro-server/src/run_files.rs index 126408e43..920460f79 100644 --- a/lib/apps/fabro-server/src/run_files.rs +++ b/lib/apps/fabro-server/src/run_files.rs @@ -1217,7 +1217,7 @@ async fn reconnect_run_sandbox( .await .map_err(|err| ApiError::new(StatusCode::CONFLICT, err.to_string()))?; sandbox - .start() + .activate() .await .map_err(|err| ApiError::new(StatusCode::CONFLICT, err.display_with_causes()))?; Ok(sandbox) diff --git a/lib/apps/fabro-server/src/server/handler/sandbox.rs b/lib/apps/fabro-server/src/server/handler/sandbox.rs index c2027571e..53877c5c9 100644 --- a/lib/apps/fabro-server/src/server/handler/sandbox.rs +++ b/lib/apps/fabro-server/src/server/handler/sandbox.rs @@ -881,7 +881,7 @@ async fn reconnect_run_sandbox_instance( let detail = render_with_causes(&err.to_string(), &collect_causes(err.as_ref())); ApiError::new(StatusCode::CONFLICT, detail).into_response() })?; - sandbox.start().await.map_err(|err| { + sandbox.activate().await.map_err(|err| { ApiError::new(StatusCode::CONFLICT, err.display_with_causes()).into_response() })?; Ok(sandbox) @@ -927,7 +927,7 @@ async fn reconnect_daytona_sandbox_instance( .map_err(|err| { ApiError::new(StatusCode::CONFLICT, err.display_with_causes()).into_response() })?; - sandbox.start().await.map_err(|err| { + sandbox.activate().await.map_err(|err| { ApiError::new(StatusCode::CONFLICT, err.display_with_causes()).into_response() })?; Ok(sandbox) diff --git a/lib/apps/fabro-server/src/server/handler/sessions.rs b/lib/apps/fabro-server/src/server/handler/sessions.rs index c25d7ca27..d571078b1 100644 --- a/lib/apps/fabro-server/src/server/handler/sessions.rs +++ b/lib/apps/fabro-server/src/server/handler/sessions.rs @@ -730,6 +730,10 @@ async fn build_agent_session( let sandbox = reconnect_for_run(sandbox_instance, daytona_api_key, Some(run_id)) .await .map_err(AskFabroBuildError::SandboxUnavailable)?; + sandbox + .activate() + .await + .map_err(|err| AskFabroBuildError::SandboxUnavailable(anyhow::Error::new(err)))?; let sandbox: Arc = Arc::from(sandbox); // No optional web-tool dependencies: `AskFabroToolAccessPolicy` denies // `web_search` and `web_fetch`, and both `tools()` and the prompt are diff --git a/lib/components/fabro-sandbox/src/daytona/mod.rs b/lib/components/fabro-sandbox/src/daytona/mod.rs index 292cc3ebe..8de651f54 100644 --- a/lib/components/fabro-sandbox/src/daytona/mod.rs +++ b/lib/components/fabro-sandbox/src/daytona/mod.rs @@ -9,6 +9,7 @@ use anyhow::Context as _; use async_trait::async_trait; use daytona_api_client::apis::api_keys_api; use daytona_api_client::apis::configuration::Configuration; +use daytona_api_client::models::SandboxState; use daytona_api_client::models::api_key_list::Permissions; use daytona_sdk::api_types::SignedPortPreviewUrl; use daytona_sdk::toolbox_types::Command as SessionCommandResult; @@ -61,6 +62,7 @@ pub(crate) const DAYTONA_DASHBOARD_SANDBOXES_URL: &str = "https://app.daytona.io/dashboard/sandboxes"; const FABRO_SANDBOX_USER_AGENT: &str = concat!("fabro-sandbox/", env!("CARGO_PKG_VERSION")); const DAYTONA_PROBE_TIMEOUT: Duration = Duration::from_secs(20); +const DAYTONA_START_TIMEOUT: Duration = Duration::from_mins(1); /// Upper bound on explicit and Drop-triggered Daytona session deletion so a /// stalled REST call cannot block cancellation/timeout paths indefinitely. const DAYTONA_SESSION_CLOSE_TIMEOUT: Duration = Duration::from_secs(10); @@ -1367,6 +1369,25 @@ impl Sandbox for DaytonaSandbox { Ok(()) } + async fn activate(&self) -> crate::Result<()> { + let sandbox = self.sandbox()?; + let current = self.client.get(&sandbox.name).await.map_err(|e| { + crate::Error::context("Failed to inspect Daytona sandbox before activation", e) + })?; + if current.state == Some(SandboxState::Started) { + return Ok(()); + } + if current.state == Some(SandboxState::Starting) { + return current + .wait_for_start(Some(DAYTONA_START_TIMEOUT)) + .await + .map_err(|e| { + crate::Error::context("Failed to wait for Daytona sandbox activation", e) + }); + } + self.start().await + } + async fn stop(&self) -> crate::Result<()> { self.emit(SandboxEvent::StopStarted { provider: "daytona".into(), @@ -2522,10 +2543,12 @@ fn build_bash_session_command( #[cfg(test)] mod tests { + use std::sync::atomic::AtomicU32; + use daytona_api_client::models::api_key_list::Permissions; use fabro_util::error::collect_chain; - use httpmock::Method::GET; - use httpmock::MockServer; + use httpmock::Method::{GET, POST}; + use httpmock::{HttpMockResponse, MockServer}; use super::*; use crate::sandbox::BASH_PROBE_MARKER; @@ -2628,6 +2651,25 @@ mod tests { }) } + fn sandbox_body(name: &str, state: SandboxState) -> serde_json::Value { + serde_json::json!({ + "id": name, + "organizationId": "org-1", + "name": name, + "user": "daytona", + "env": {}, + "labels": {}, + "public": false, + "networkBlockAll": false, + "target": "us", + "cpu": 2.0, + "gpu": 0.0, + "memory": 4.0, + "disk": 20.0, + "state": state.to_string() + }) + } + #[test] fn daytona_config_defaults() { let config = DaytonaConfig::default(); @@ -2783,6 +2825,107 @@ mod tests { ); } + #[tokio::test] + async fn activate_skips_start_when_daytona_reports_started() { + let server = MockServer::start_async().await; + let get_sandbox = server + .mock_async(|when, then| { + when.method(GET) + .path("/sandbox/test-sandbox") + .header("authorization", "Bearer dtn_test"); + then.status(200) + .header("content-type", "application/json") + .json_body(sandbox_body("test-sandbox", SandboxState::Started)); + }) + .await; + let start_sandbox = server + .mock_async(|when, then| { + when.method(POST) + .path("/sandbox/test-sandbox/start") + .header("authorization", "Bearer dtn_test"); + then.status(200) + .header("content-type", "application/json") + .json_body(sandbox_body("test-sandbox", SandboxState::Started)); + }) + .await; + let sandbox = mock_daytona_sandbox(&server, "dtn_test", DaytonaConfig::default()).await; + let sdk_sandbox = sandbox + .client + .get("test-sandbox") + .await + .expect("test sandbox should load"); + sandbox + .sandbox + .set(sdk_sandbox) + .expect("test sandbox should initialize once"); + + let get_calls_before = get_sandbox.calls_async().await; + sandbox + .activate() + .await + .expect("an active sandbox should require no restart"); + + assert_eq!(get_sandbox.calls_async().await, get_calls_before + 1); + start_sandbox.assert_calls_async(0).await; + } + + #[tokio::test] + async fn activate_waits_for_a_daytona_start_already_in_progress() { + let server = MockServer::start_async().await; + let response_count = Arc::new(AtomicU32::new(0)); + let get_sandbox = server + .mock_async({ + let response_count = Arc::clone(&response_count); + move |when, then| { + when.method(GET) + .path("/sandbox/test-sandbox") + .header("authorization", "Bearer dtn_test"); + then.respond_with(move |_| { + let state = if response_count.fetch_add(1, Ordering::Relaxed) == 1 { + SandboxState::Starting + } else { + SandboxState::Started + }; + HttpMockResponse::builder() + .status(200) + .header("content-type", "application/json") + .body(sandbox_body("test-sandbox", state).to_string()) + .build() + }); + } + }) + .await; + let start_sandbox = server + .mock_async(|when, then| { + when.method(POST) + .path("/sandbox/test-sandbox/start") + .header("authorization", "Bearer dtn_test"); + then.status(200) + .header("content-type", "application/json") + .json_body(sandbox_body("test-sandbox", SandboxState::Started)); + }) + .await; + let sandbox = mock_daytona_sandbox(&server, "dtn_test", DaytonaConfig::default()).await; + let sdk_sandbox = sandbox + .client + .get("test-sandbox") + .await + .expect("test sandbox should load"); + sandbox + .sandbox + .set(sdk_sandbox) + .expect("test sandbox should initialize once"); + + let get_calls_before = get_sandbox.calls_async().await; + sandbox + .activate() + .await + .expect("an in-progress start should be awaited"); + + assert_eq!(get_sandbox.calls_async().await, get_calls_before + 2); + start_sandbox.assert_calls_async(0).await; + } + #[tokio::test] async fn base_params_merges_managed_daytona_labels() { let run_id: RunId = "01HY0000000000000000000000".parse().unwrap(); diff --git a/lib/components/fabro-sandbox/src/docker.rs b/lib/components/fabro-sandbox/src/docker.rs index fb9b7df9c..6236d425d 100644 --- a/lib/components/fabro-sandbox/src/docker.rs +++ b/lib/components/fabro-sandbox/src/docker.rs @@ -15,7 +15,7 @@ use bollard::container::{ use bollard::errors::Error as DockerError; use bollard::exec::{CreateExecOptions, StartExecOptions, StartExecResults}; use bollard::image::CreateImageOptions; -use bollard::models::HostConfig; +use bollard::models::{ContainerInspectResponse, HostConfig}; use fabro_github::GitHubCredentials; use fabro_types::{CommandOutputStream, CommandTermination, RunId}; use fabro_util::time::elapsed_ms; @@ -145,6 +145,13 @@ enum EnsureImageOutcome { Pulled, } +#[derive(Clone, Copy, Debug, PartialEq, Eq, strum::Display)] +#[strum(serialize_all = "lowercase")] +enum ContainerStartAction { + Start, + Unpause, +} + impl DockerSandbox { pub fn new( config: DockerSandboxOptions, @@ -154,7 +161,25 @@ impl DockerSandbox { clone_branch: Option, ) -> crate::Result { let docker = Docker::connect_with_local_defaults().map_err(crate::Error::docker_connect)?; - Ok(Self { + Ok(Self::with_docker_client( + docker, + config, + github_app, + run_id, + clone_origin_url, + clone_branch, + )) + } + + fn with_docker_client( + docker: Docker, + config: DockerSandboxOptions, + github_app: Option, + run_id: Option, + clone_origin_url: Option, + clone_branch: Option, + ) -> Self { + Self { docker, config, github_app, @@ -169,7 +194,7 @@ impl DockerSandbox { cached_os_version: std::sync::OnceLock::new(), rg_available: OnceCell::const_new(), event_callback: None, - }) + } } pub async fn reconnect( @@ -755,24 +780,26 @@ impl DockerSandbox { verify_managed_labels(container_id, &labels, self.run_id.as_ref()) } - async fn inspect_labels(&self, container_id: &str) -> crate::Result> { - let inspect = self - .docker + async fn inspect_container( + &self, + container_id: &str, + ) -> crate::Result { + self.docker .inspect_container(container_id, None::) .await - .map_err(|e| { - if docker_not_found(&e) { - crate::Error::message(format!("Docker container '{container_id}' is gone")) + .map_err(|source| { + let message = if docker_not_found(&source) { + format!("Docker container '{container_id}' is gone") } else { - crate::Error::message(format!( - "Failed to inspect Docker container '{container_id}': {e}" - )) - } - })?; - Ok(inspect - .config - .and_then(|config| config.labels) - .unwrap_or_default()) + format!("Failed to inspect Docker container '{container_id}'") + }; + crate::Error::context(message, source) + }) + } + + async fn inspect_labels(&self, container_id: &str) -> crate::Result> { + let inspect = self.inspect_container(container_id).await?; + Ok(container_labels(&inspect)) } async fn ensure_name_available(&self) -> crate::Result> { @@ -835,6 +862,67 @@ impl DockerSandbox { .map_err(|e| crate::Error::context("Failed to upload file to container", e)) } + fn begin_start(&self) -> Instant { + self.emit(SandboxEvent::StartStarted { + provider: "docker".into(), + }); + Instant::now() + } + + async fn set_container_running( + &self, + container_id: &str, + labels: &HashMap, + action: ContainerStartAction, + ) -> crate::Result<()> { + let result = match action { + ContainerStartAction::Start => { + self.docker + .start_container(container_id, None::>) + .await + } + ContainerStartAction::Unpause => self.docker.unpause_container(container_id).await, + }; + if let Err(source) = result { + if !docker_not_modified(&source) { + return Err(crate::Error::context( + format!( + "Failed to {action} Docker container '{container_id}' with labels {labels:?}" + ), + source, + )); + } + } + Ok(()) + } + + async fn complete_start( + &self, + started: Instant, + container_id: &str, + labels: &HashMap, + action: ContainerStartAction, + ) -> crate::Result<()> { + if let Err(error) = self + .set_container_running(container_id, labels, action) + .await + { + return self.start_error(error); + } + if let Err(error) = self.probe_bash(None).await { + return self.start_error(crate::Error::context( + format!("Docker container '{container_id}' health check"), + error, + )); + } + + self.emit(SandboxEvent::StartCompleted { + provider: "docker".into(), + duration_ms: elapsed_ms(started), + }); + Ok(()) + } + fn start_error(&self, error: crate::Error) -> crate::Result<()> { self.emit(SandboxEvent::StartFailed { provider: "docker".into(), @@ -1230,6 +1318,24 @@ fn verify_managed_labels( Ok(()) } +fn container_labels(inspect: &ContainerInspectResponse) -> HashMap { + inspect + .config + .as_ref() + .and_then(|config| config.labels.clone()) + .unwrap_or_default() +} + +fn activation_action(inspect: &ContainerInspectResponse) -> Option { + let Some(state) = inspect.state.as_ref() else { + return Some(ContainerStartAction::Start); + }; + if state.running != Some(true) { + return Some(ContainerStartAction::Start); + } + (state.paused == Some(true)).then_some(ContainerStartAction::Unpause) +} + fn docker_not_found(error: &DockerError) -> bool { matches!(error, DockerError::DockerResponseServerError { status_code: 404, @@ -1418,47 +1524,40 @@ impl Sandbox for DockerSandbox { } async fn start(&self) -> crate::Result<()> { - self.emit(SandboxEvent::StartStarted { - provider: "docker".into(), - }); - let start = Instant::now(); + let started = self.begin_start(); let container_id = self.container_id()?.to_string(); - let labels = match self.inspect_labels(&container_id).await { - Ok(labels) => labels, - Err(e) => return self.start_error(e), + let inspect = match self.inspect_container(&container_id).await { + Ok(inspect) => inspect, + Err(error) => return self.start_error(error), }; - if let Err(e) = verify_managed_labels(&container_id, &labels, self.run_id.as_ref()) { - return self.start_error(e); + let labels = container_labels(&inspect); + if let Err(error) = verify_managed_labels(&container_id, &labels, self.run_id.as_ref()) { + return self.start_error(error); } - - if let Err(e) = self - .docker - .start_container(&container_id, None::>) + let action = activation_action(&inspect).unwrap_or(ContainerStartAction::Start); + self.complete_start(started, &container_id, &labels, action) .await - { - if !docker_not_modified(&e) { - return self.start_error(crate::Error::context( - format!( - "Failed to start Docker container '{container_id}' with labels {labels:?}" - ), - e, - )); + } + + async fn activate(&self) -> crate::Result<()> { + let container_id = self.container_id()?.to_string(); + let inspect = self.inspect_container(&container_id).await?; + let labels = container_labels(&inspect); + verify_managed_labels(&container_id, &labels, self.run_id.as_ref())?; + let Some(action) = activation_action(&inspect) else { + return Ok(()); + }; + match action { + ContainerStartAction::Unpause => { + self.set_container_running(&container_id, &labels, action) + .await + } + ContainerStartAction::Start => { + let started = self.begin_start(); + self.complete_start(started, &container_id, &labels, action) + .await } } - - if let Err(e) = self.probe_bash(None).await { - return self.start_error(crate::Error::context( - format!("Docker container '{container_id}' health check"), - e, - )); - } - - let duration_ms = elapsed_ms(start); - self.emit(SandboxEvent::StartCompleted { - provider: "docker".into(), - duration_ms, - }); - Ok(()) } async fn stop(&self) -> crate::Result<()> { @@ -2023,6 +2122,9 @@ mod tests { use std::process::Stdio; use std::time::Duration; + use bollard::API_DEFAULT_VERSION; + use httpmock::Method::{GET, POST}; + use httpmock::MockServer; use tokio::io::AsyncWriteExt as _; use tokio::process::Command; @@ -2135,6 +2237,54 @@ mod tests { )); } + #[tokio::test] + async fn activate_unpauses_paused_running_container() { + let server = MockServer::start_async().await; + let inspect = server + .mock_async(|when, then| { + when.method(GET) + .path_suffix("/containers/test-container/json"); + then.status(200) + .header("content-type", "application/json") + .json_body(serde_json::json!({ + "Config": { + "Labels": managed_labels::for_run(None) + }, + "State": { + "Running": true, + "Paused": true + } + })); + }) + .await; + let unpause = server + .mock_async(|when, then| { + when.method(POST) + .path_suffix("/containers/test-container/unpause"); + then.status(204); + }) + .await; + let start = server + .mock_async(|when, then| { + when.method(POST) + .path_suffix("/containers/test-container/start"); + then.status(204); + }) + .await; + let docker = Docker::connect_with_http(&server.base_url(), 5, API_DEFAULT_VERSION) + .expect("mock Docker client should connect"); + let sandbox = test_docker_sandbox(docker, "test-container"); + + sandbox + .activate() + .await + .expect("a paused running container should be unpaused"); + + inspect.assert_calls_async(1).await; + unpause.assert_calls_async(1).await; + start.assert_calls_async(0).await; + } + #[test] fn default_options_are_clone_based() { let options = DockerSandboxOptions::default(); @@ -2447,4 +2597,20 @@ mod tests { entry.read_to_string(&mut content).unwrap(); assert_eq!(content, "hello"); } + + fn test_docker_sandbox(docker: Docker, container_id: &str) -> DockerSandbox { + let sandbox = DockerSandbox::with_docker_client( + docker, + DockerSandboxOptions::default(), + None, + None, + None, + None, + ); + sandbox + .container_id + .set(container_id.to_string()) + .expect("test container should initialize once"); + sandbox + } } diff --git a/lib/components/fabro-sandbox/src/local.rs b/lib/components/fabro-sandbox/src/local.rs index 2da5a788e..937abdc64 100644 --- a/lib/components/fabro-sandbox/src/local.rs +++ b/lib/components/fabro-sandbox/src/local.rs @@ -835,6 +835,13 @@ impl Sandbox for LocalSandbox { result } + async fn activate(&self) -> crate::Result<()> { + // Local sandboxes have no provider resource that can stop or pause. + // Resume paths still call `start()` to recreate the directory and + // verify Bash. + Ok(()) + } + async fn git_push_ref(&self, refspec: &str) -> crate::Result<()> { let has_origin = match self .exec_command("git remote get-url origin", 10_000, None, None, None) diff --git a/lib/components/fabro-sandbox/src/sandbox.rs b/lib/components/fabro-sandbox/src/sandbox.rs index fee0c6c02..b29ccea39 100644 --- a/lib/components/fabro-sandbox/src/sandbox.rs +++ b/lib/components/fabro-sandbox/src/sandbox.rs @@ -250,6 +250,10 @@ macro_rules! delegate_sandbox { self.$field.initialize().await } + async fn activate(&self) -> $crate::Result<()> { + self.$field.activate().await + } + async fn start(&self) -> $crate::Result<()> { self.$field.start().await } @@ -1130,6 +1134,15 @@ pub trait Sandbox: Send + Sync { remote_path: &str, ) -> crate::Result<()>; async fn initialize(&self) -> crate::Result<()>; + /// Ensure the provider resource is running and not paused before access. + /// + /// This access-time operation must be idempotent. Providers that can stop + /// independently should avoid restarting an already-active sandbox. This + /// lightweight check does not require the full health verification done by + /// [`Sandbox::start`], and it does not keep a sandbox active between calls. + async fn activate(&self) -> crate::Result<()> { + self.start().await + } async fn start(&self) -> crate::Result<()> { Ok(()) } diff --git a/lib/components/fabro-sandbox/src/terminal.rs b/lib/components/fabro-sandbox/src/terminal.rs index 44081f9bb..f8df8aa2a 100644 --- a/lib/components/fabro-sandbox/src/terminal.rs +++ b/lib/components/fabro-sandbox/src/terminal.rs @@ -64,7 +64,7 @@ pub async fn open_terminal_for_run( runtime.clone_branch.clone(), ) .await?; - sandbox.start().await?; + sandbox.activate().await?; let api_key = resolve_daytona_api_key(daytona_api_key)?; let organization_id = resolve_daytona_organization_id(daytona_organization_id); let session = DaytonaTerminalSession::open( @@ -95,7 +95,7 @@ pub async fn open_terminal_for_run( run_id, ) .await?; - sandbox.start().await?; + sandbox.activate().await?; let session = DockerTerminalSession::open(&sandbox, size).await?; Ok(Box::new(session)) } diff --git a/lib/components/fabro-sandbox/src/test_support.rs b/lib/components/fabro-sandbox/src/test_support.rs index 4b62ac20a..65ff4af25 100644 --- a/lib/components/fabro-sandbox/src/test_support.rs +++ b/lib/components/fabro-sandbox/src/test_support.rs @@ -1,5 +1,6 @@ use std::collections::HashMap; use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use async_trait::async_trait; @@ -38,6 +39,9 @@ pub struct MockSandbox { pub captured_working_dirs: Mutex>>, /// Captures the `env_vars` argument from `exec_command` calls. pub captured_env_vars: Mutex>>, + pub active: AtomicBool, + pub activate_error: Option, + pub activate_calls: Mutex, pub start_calls: Mutex, pub stop_calls: Mutex, pub delete_calls: Mutex, @@ -51,6 +55,8 @@ pub struct MockSandbox { /// filtering. pub walk_files: Vec, pub walk_files_error: Option, + pub walk_files_called: AtomicBool, + pub walked_while_inactive: AtomicBool, /// Reported by `exec_command_streaming`. Set to `false` to model a /// provider that cannot separate stdout from stderr. pub streams_separated: bool, @@ -70,10 +76,25 @@ impl MockSandbox { *self.start_calls.lock().expect("start_calls lock poisoned") } + pub fn activate_count(&self) -> u32 { + *self + .activate_calls + .lock() + .expect("activate_calls lock poisoned") + } + pub fn stop_count(&self) -> u32 { *self.stop_calls.lock().expect("stop_calls lock poisoned") } + pub fn walk_files_was_called(&self) -> bool { + self.walk_files_called.load(Ordering::Relaxed) + } + + pub fn walked_while_inactive(&self) -> bool { + self.walked_while_inactive.load(Ordering::Relaxed) + } + pub fn delete_count(&self) -> u32 { *self .delete_calls @@ -99,6 +120,12 @@ impl MockSandbox { self.walk_files_error = Some(error.into()); self } + + #[must_use] + pub fn with_activate_error(mut self, error: impl Into) -> Self { + self.activate_error = Some(error.into()); + self + } } impl MockSandbox { @@ -132,6 +159,9 @@ impl Default for MockSandbox { captured_commands: Mutex::new(Vec::new()), captured_working_dirs: Mutex::new(Vec::new()), captured_env_vars: Mutex::new(None), + active: AtomicBool::new(true), + activate_error: None, + activate_calls: Mutex::new(0), start_calls: Mutex::new(0), stop_calls: Mutex::new(0), delete_calls: Mutex::new(0), @@ -141,6 +171,8 @@ impl Default for MockSandbox { exec_error: None, walk_files: Vec::new(), walk_files_error: None, + walk_files_called: AtomicBool::new(false), + walked_while_inactive: AtomicBool::new(false), streams_separated: true, } } @@ -367,6 +399,11 @@ impl Sandbox for MockSandbox { relative_start: &str, options: &WalkOptions, ) -> crate::Result> { + self.walk_files_called.store(true, Ordering::Relaxed); + if !self.active.load(Ordering::Relaxed) { + self.walked_while_inactive.store(true, Ordering::Relaxed); + return Err(crate::Error::message("Sandbox is stopped")); + } if let Some(error) = &self.walk_files_error { return Err(crate::Error::message(error.clone())); } @@ -430,6 +467,7 @@ impl Sandbox for MockSandbox { } async fn initialize(&self) -> crate::Result<()> { + self.active.store(true, Ordering::Relaxed); self.emit(SandboxEvent::Initializing { provider: "mock".into(), }); @@ -444,13 +482,29 @@ impl Sandbox for MockSandbox { Ok(()) } + async fn activate(&self) -> crate::Result<()> { + *self + .activate_calls + .lock() + .expect("activate_calls lock poisoned") += 1; + if let Some(error) = &self.activate_error { + return Err(crate::Error::context( + "Mock sandbox activation failed", + std::io::Error::other(error.clone()), + )); + } + self.start().await + } + async fn start(&self) -> crate::Result<()> { *self.start_calls.lock().expect("start_calls lock poisoned") += 1; + self.active.store(true, Ordering::Relaxed); Ok(()) } async fn stop(&self) -> crate::Result<()> { *self.stop_calls.lock().expect("stop_calls lock poisoned") += 1; + self.active.store(false, Ordering::Relaxed); Ok(()) } diff --git a/lib/components/fabro-workflow/src/lifecycle/mod.rs b/lib/components/fabro-workflow/src/lifecycle/mod.rs index 751c20842..64a6323f2 100644 --- a/lib/components/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/components/fabro-workflow/src/lifecycle/mod.rs @@ -13,7 +13,7 @@ use std::sync::{Arc, Mutex}; use std::time::Instant; use async_trait::async_trait; -use fabro_core::error::Result as CoreResult; +use fabro_core::error::{Error as CoreError, Result as CoreResult}; use fabro_core::graph::NodeSpec; use fabro_core::lifecycle::{ AttemptContext, AttemptResultContext, EdgeContext, EdgeDecision, NodeDecision, RunLifecycle, @@ -60,6 +60,7 @@ pub(crate) struct WorkflowLifecycle { circuit_breaker: Arc, git: GitLifecycle, artifact: ArtifactLifecycle, + sandbox: Arc, on_node: crate::OnNodeCallback, emitter: Arc, run_control: Option>, @@ -190,6 +191,7 @@ impl WorkflowLifecycle { circuit_breaker, git, artifact, + sandbox: Arc::clone(sandbox), on_node, emitter: Arc::clone(emitter), run_control, @@ -280,6 +282,13 @@ impl RunLifecycle for WorkflowLifecycle { if let Some(run_control) = &self.run_control { run_control.wait_if_paused(self.emitter.as_ref()).await; } + // A provider may auto-stop while the run is paused between nodes. + self.sandbox.activate().await.map_err(|err| { + CoreError::context( + format!("failed to activate sandbox before node {}", node.id()), + err, + ) + })?; if let Some(on_node) = &self.on_node { on_node(node.id()); } @@ -330,6 +339,17 @@ impl RunLifecycle for WorkflowLifecycle { if let Some(run_control) = &self.run_control { run_control.wait_if_paused(self.emitter.as_ref()).await; } + // Human, wait, and paused stages can return after a long period with + // no sandbox traffic. Reactivate before artifact and checkpoint work. + self.sandbox.activate().await.map_err(|err| { + CoreError::context( + format!( + "failed to activate sandbox after node attempt {}", + ctx.node.id() + ), + err, + ) + })?; self.artifact.after_attempt(ctx, state).await?; self.event.after_attempt(ctx, state).await?; Ok(()) diff --git a/lib/components/fabro-workflow/src/pipeline/execute.rs b/lib/components/fabro-workflow/src/pipeline/execute.rs index e89006ab0..a82fd7c8a 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute.rs @@ -276,6 +276,13 @@ pub async fn execute(init: Initialized) -> Executed { Err(fabro_core::Error::Blocked { message }) => { (Err(Error::engine(message)), initial_context) } + Err(error @ fabro_core::Error::Context { .. }) => ( + Err(Error::engine_with_source( + "Pipeline lifecycle operation failed", + error, + )), + initial_context, + ), Err(e) => (Err(Error::engine(e.to_string())), initial_context), }; diff --git a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs index 2322e6b71..5faa9c1ac 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs @@ -16,6 +16,7 @@ use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_hooks::HookSettings; use fabro_interview::AutoApproveInterviewer; use fabro_sandbox::SandboxSpec; +use fabro_sandbox::test_support::MockSandbox; use fabro_store::Database; use fabro_types::settings::run::RunModelControls; use fabro_types::{ @@ -647,6 +648,28 @@ impl HandlerTrait for SlowHandler { } } +struct StopsSandboxHandler { + sandbox: Arc, +} + +#[async_trait] +impl HandlerTrait for StopsSandboxHandler { + async fn execute( + &self, + _node: &Node, + _context: &Context, + _graph: &Graph, + _run_dir: &Path, + _services: &crate::handler::EngineServices, + ) -> std::result::Result { + self.sandbox + .stop() + .await + .map_err(|err| Error::handler_with_source("failed to stop test sandbox", err))?; + Ok(Outcome::success()) + } +} + struct PanickingHandler; #[async_trait] @@ -802,6 +825,65 @@ async fn execute_runs_simple_workflow() { assert_eq!(outcome.status, StageOutcome::Succeeded); } +#[tokio::test] +async fn execute_preserves_sandbox_activation_error_chain() { + let dir = tempfile::tempdir().unwrap(); + let sandbox: Arc = + Arc::new(MockSandbox::linux().with_activate_error("provider unavailable")); + + let error = run_graph( + make_registry(), + test_emitter_arc("test-run"), + sandbox, + &simple_graph(), + &test_run_options(dir.path(), "test-run"), + ) + .await + .expect_err("sandbox activation should fail"); + + assert_eq!(error.causes(), vec![ + "failed to activate sandbox before node start", + "Mock sandbox activation failed", + "provider unavailable", + ]); +} + +#[tokio::test] +async fn execute_reactivates_sandbox_after_a_stage_can_leave_it_stopped() { + let dir = tempfile::tempdir().unwrap(); + let sandbox = Arc::new(MockSandbox::linux()); + let mut registry = make_registry(); + registry.register( + "start", + Box::new(StopsSandboxHandler { + sandbox: Arc::clone(&sandbox), + }), + ); + let sandbox_for_run: Arc = sandbox.clone(); + let mut run_options = test_run_options(dir.path(), "test-run"); + run_options + .settings + .run + .artifacts + .include + .push("**/*".to_string()); + + let outcome = run_graph( + registry, + test_emitter_arc("test-run"), + sandbox_for_run, + &simple_graph(), + &run_options, + ) + .await + .unwrap(); + + assert_eq!(outcome.status, StageOutcome::Succeeded); + assert_eq!(sandbox.stop_count(), 1); + assert!(sandbox.walk_files_was_called()); + assert!(!sandbox.walked_while_inactive()); +} + #[tokio::test] async fn execute_saves_checkpoint() { let dir = tempfile::tempdir().unwrap(); diff --git a/lib/components/fabro-workflow/src/pipeline/initialize.rs b/lib/components/fabro-workflow/src/pipeline/initialize.rs index 149a9ee59..0f02db0c2 100644 --- a/lib/components/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/components/fabro-workflow/src/pipeline/initialize.rs @@ -392,6 +392,8 @@ pub async fn initialize( }); if attach_existing { + // Resume needs the full provider health check. `activate()` is the + // lighter access-time operation used after a run is already active. sandbox .start() .await diff --git a/lib/foundation/fabro-core/src/error.rs b/lib/foundation/fabro-core/src/error.rs index ea99e87d9..fc4bfbdeb 100644 --- a/lib/foundation/fabro-core/src/error.rs +++ b/lib/foundation/fabro-core/src/error.rs @@ -55,6 +55,12 @@ pub enum Error { StallTimeout { node_id: String }, #[error("{detail}")] Handler { detail: Box }, + #[error("{message}")] + Context { + message: String, + #[source] + source: Box, + }, #[error("{0}")] Other(String), } @@ -72,6 +78,16 @@ impl Error { } } + pub fn context( + message: impl Into, + source: impl std::error::Error + Send + Sync + 'static, + ) -> Self { + Self::Context { + message: message.into(), + source: Box::new(source), + } + } + pub fn is_retryable(&self) -> bool { matches!(self, Self::Handler { detail } if detail.retryable) } @@ -94,6 +110,8 @@ pub type Result = std::result::Result; #[cfg(test)] mod tests { + use std::error::Error as _; + use super::*; use crate::outcome::FailureCategory; @@ -153,6 +171,20 @@ mod tests { assert!(!not_retryable.is_retryable()); } + #[test] + fn core_error_context_preserves_source() { + let error = Error::context( + "failed to activate sandbox", + std::io::Error::other("provider unavailable"), + ); + + assert_eq!(error.to_string(), "failed to activate sandbox"); + assert_eq!( + error.source().map(ToString::to_string).as_deref(), + Some("provider unavailable") + ); + } + #[test] fn core_error_handler_to_fail_outcome() { let err = Error::handler(HandlerErrorDetail {