fix(sandbox): reactivate stopped run sandboxes

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)

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)

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();

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::<InspectContainerOptions>)
+            .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(),

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)

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(())
     }

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))
         }

lib/components/fabro-sandbox/src/test_support.rs

@@ -38,6 +38,7 @@ pub struct MockSandbox {
     pub captured_working_dirs: Mutex<Vec<Option<String>>>,
     /// Captures the `env_vars` argument from `exec_command` calls.
     pub captured_env_vars:     Mutex<Option<HashMap<String, String>>>,
+    pub activate_calls:        Mutex<u32>,
     pub start_calls:           Mutex<u32>,
     pub stop_calls:            Mutex<u32>,
     pub delete_calls:          Mutex<u32>,
@@ -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(())

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<CircuitBreakerLifecycle>,
     git:                   GitLifecycle,
     artifact:              ArtifactLifecycle,
+    sandbox:               Arc<dyn Sandbox>,
     on_node:               crate::OnNodeCallback,
     emitter:               Arc<Emitter>,
     run_control:           Option<Arc<RunControlState>>,
@@ -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<WorkflowGraph> 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<WorkflowGraph> 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(())

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<MockSandbox>,
+    observed_activation_count: Arc<AtomicU32>,
+}
+
+#[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<Outcome, Error> {
+        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<dyn Sandbox> = 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();