From 7906af3f3fcafae5f9523adfdcf0b86ffbaa16b5 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 28 Jul 2026 12:50:20 -0400 Subject: [PATCH] fix(sandbox): reactivate stopped run sandboxes --- lib/apps/fabro-server/src/run_files.rs | 2 +- .../src/server/handler/sandbox.rs | 4 +- .../fabro-sandbox/src/daytona/mod.rs | 63 +++++++++++++++++++ lib/components/fabro-sandbox/src/docker.rs | 29 +++++++++ lib/components/fabro-sandbox/src/local.rs | 4 ++ lib/components/fabro-sandbox/src/sandbox.rs | 12 ++++ lib/components/fabro-sandbox/src/terminal.rs | 4 +- .../fabro-sandbox/src/test_support.rs | 17 +++++ .../fabro-workflow/src/lifecycle/mod.rs | 21 ++++++- .../src/pipeline/execute/tests.rs | 58 +++++++++++++++++ 10 files changed, 208 insertions(+), 6 deletions(-) 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/components/fabro-sandbox/src/daytona/mod.rs b/lib/components/fabro-sandbox/src/daytona/mod.rs index 292cc3ebe..c4e01e6ff 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; @@ -1367,6 +1368,17 @@ 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(()); + } + self.start().await + } + async fn stop(&self) -> crate::Result<()> { self.emit(SandboxEvent::StopStarted { provider: "daytona".into(), @@ -2628,6 +2640,25 @@ mod tests { }) } + fn sandbox_body(name: &str, state: &str) -> 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 + }) + } + #[test] fn daytona_config_defaults() { let config = DaytonaConfig::default(); @@ -2783,6 +2814,38 @@ 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", "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"); + + sandbox + .activate() + .await + .expect("an active sandbox should require no restart"); + + get_sandbox.assert_calls_async(2).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..9a9bdaa08 100644 --- a/lib/components/fabro-sandbox/src/docker.rs +++ b/lib/components/fabro-sandbox/src/docker.rs @@ -1461,6 +1461,35 @@ impl Sandbox for DockerSandbox { Ok(()) } + async fn activate(&self) -> crate::Result<()> { + let container_id = self.container_id()?.to_string(); + let inspect = 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")) + } else { + crate::Error::message(format!( + "Failed to inspect Docker container '{container_id}': {e}" + )) + } + })?; + let labels = inspect + .config + .and_then(|config| config.labels) + .unwrap_or_default(); + verify_managed_labels(&container_id, &labels, self.run_id.as_ref())?; + let active = inspect + .state + .is_some_and(|state| state.running == Some(true) && state.paused != Some(true)); + if active { + return Ok(()); + } + self.start().await + } + async fn stop(&self) -> crate::Result<()> { self.emit(SandboxEvent::StopStarted { provider: "docker".into(), diff --git a/lib/components/fabro-sandbox/src/local.rs b/lib/components/fabro-sandbox/src/local.rs index 2da5a788e..6af6c4e8d 100644 --- a/lib/components/fabro-sandbox/src/local.rs +++ b/lib/components/fabro-sandbox/src/local.rs @@ -835,6 +835,10 @@ impl Sandbox for LocalSandbox { result } + async fn activate(&self) -> crate::Result<()> { + 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..90fdd9bcb 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,14 @@ pub trait Sandbox: Send + Sync { remote_path: &str, ) -> crate::Result<()>; async fn initialize(&self) -> crate::Result<()>; + /// Ensure the sandbox is active and ready for ordinary operations. + /// + /// This access-time operation must be idempotent. Providers that can stop + /// independently should avoid restarting an already-active sandbox. This + /// method 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..a1f9e102b 100644 --- a/lib/components/fabro-sandbox/src/test_support.rs +++ b/lib/components/fabro-sandbox/src/test_support.rs @@ -38,6 +38,7 @@ pub struct MockSandbox { pub captured_working_dirs: Mutex>>, /// Captures the `env_vars` argument from `exec_command` calls. pub captured_env_vars: Mutex>>, + pub activate_calls: Mutex, pub start_calls: Mutex, pub stop_calls: Mutex, pub delete_calls: Mutex, @@ -70,6 +71,13 @@ 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") } @@ -132,6 +140,7 @@ impl Default for MockSandbox { captured_commands: Mutex::new(Vec::new()), captured_working_dirs: Mutex::new(Vec::new()), captured_env_vars: Mutex::new(None), + activate_calls: Mutex::new(0), start_calls: Mutex::new(0), stop_calls: Mutex::new(0), delete_calls: Mutex::new(0), @@ -444,6 +453,14 @@ impl Sandbox for MockSandbox { Ok(()) } + async fn activate(&self) -> crate::Result<()> { + *self + .activate_calls + .lock() + .expect("activate_calls lock poisoned") += 1; + self.start().await + } + async fn start(&self) -> crate::Result<()> { *self.start_calls.lock().expect("start_calls lock poisoned") += 1; Ok(()) diff --git a/lib/components/fabro-workflow/src/lifecycle/mod.rs b/lib/components/fabro-workflow/src/lifecycle/mod.rs index 751c20842..cf442f497 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,14 @@ 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::Other(format!( + "failed to activate sandbox before node {:?}: {}", + node.id(), + err.display_with_causes() + )) + })?; if let Some(on_node) = &self.on_node { on_node(node.id()); } @@ -330,6 +340,15 @@ 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::Other(format!( + "failed to activate sandbox after node attempt {:?}: {}", + ctx.node.id(), + err.display_with_causes() + )) + })?; 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/tests.rs b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs index 2322e6b71..6807277e8 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,31 @@ impl HandlerTrait for SlowHandler { } } +struct StopsSandboxHandler { + sandbox: Arc, + observed_activation_count: 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.observed_activation_count + .store(self.sandbox.activate_count(), Ordering::Relaxed); + 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 +828,38 @@ async fn execute_runs_simple_workflow() { assert_eq!(outcome.status, StageOutcome::Succeeded); } +#[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 observed_activation_count = Arc::new(AtomicU32::new(0)); + let mut registry = make_registry(); + registry.register( + "start", + Box::new(StopsSandboxHandler { + sandbox: Arc::clone(&sandbox), + observed_activation_count: Arc::clone(&observed_activation_count), + }), + ); + let sandbox_for_run: Arc = sandbox.clone(); + + let outcome = run_graph( + registry, + test_emitter_arc("test-run"), + sandbox_for_run, + &simple_graph(), + &test_run_options(dir.path(), "test-run"), + ) + .await + .unwrap(); + + assert_eq!(outcome.status, StageOutcome::Succeeded); + assert_eq!(observed_activation_count.load(Ordering::Relaxed), 1); + assert_eq!(sandbox.stop_count(), 1); + assert_eq!(sandbox.activate_count(), 2); + assert_eq!(sandbox.start_count(), 2); +} + #[tokio::test] async fn execute_saves_checkpoint() { let dir = tempfile::tempdir().unwrap();