scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-09 03:20:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Merge pull request #670 from fabro-sh/fix/sandbox-activate-before-access

Browse source 
Reactivate stopped sandboxes at workflow boundaries
This commit is contained in:
Bryan Helmkamp 2026-07-28 15:54:57 -04:00 • committed by GitHub
commit 450a1701b2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 591 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/apps/fabro-server/src/run_files.rs
View file

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

4
lib/apps/fabro-server/src/server/handler/sandbox.rs
View file

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

4
lib/apps/fabro-server/src/server/handler/sessions.rs
View file

@ -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<dyn fabro_agent::Sandbox> = Arc::from(sandbox);
// No optional web-tool dependencies: `AskFabroToolAccessPolicy` denies
// `web_search` and `web_fetch`, and both `tools()` and the prompt are

147
lib/components/fabro-sandbox/src/daytona/mod.rs
View file

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

272
lib/components/fabro-sandbox/src/docker.rs
View file

@ -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<String>,
) -> crate::Result<Self> {
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<GitHubCredentials>,
run_id: Option<RunId>,
clone_origin_url: Option<String>,
clone_branch: Option<String>,
) -> 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<HashMap<String, String>> {
let inspect = self
.docker
async fn inspect_container(
&self,
container_id: &str,
) -> crate::Result<ContainerInspectResponse> {
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"))
.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<HashMap<String, String>> {
let inspect = self.inspect_container(container_id).await?;
Ok(container_labels(&inspect))
}
async fn ensure_name_available(&self) -> crate::Result<Option<String>> {
@ -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<String, String>,
action: ContainerStartAction,
) -> crate::Result<()> {
let result = match action {
ContainerStartAction::Start => {
self.docker
.start_container(container_id, None::<StartContainerOptions<String>>)
.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<String, String>,
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<String, String> {
inspect
.config
.as_ref()
.and_then(|config| config.labels.clone())
.unwrap_or_default()
}
fn activation_action(inspect: &ContainerInspectResponse) -> Option<ContainerStartAction> {
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::<StartContainerOptions<String>>)
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
}
}

7
lib/components/fabro-sandbox/src/local.rs
View file

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

13
lib/components/fabro-sandbox/src/sandbox.rs
View file

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

4
lib/components/fabro-sandbox/src/terminal.rs
View file

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

54
lib/components/fabro-sandbox/src/test_support.rs
View file

@ -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<Vec<Option<String>>>,
/// Captures the `env_vars` argument from `exec_command` calls.
pub captured_env_vars: Mutex<Option<HashMap<String, String>>>,
pub active: AtomicBool,
pub activate_error: Option<String>,
pub activate_calls: Mutex<u32>,
pub start_calls: Mutex<u32>,
pub stop_calls: Mutex<u32>,
pub delete_calls: Mutex<u32>,
@ -51,6 +55,8 @@ pub struct MockSandbox {
/// filtering.
pub walk_files: Vec<SandboxFile>,
pub walk_files_error: Option<String>,
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<String>) -> 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<Vec<SandboxFile>> {
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(())
}

22
lib/components/fabro-workflow/src/lifecycle/mod.rs
View file

@ -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,13 @@ 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::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<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::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(())

7
lib/components/fabro-workflow/src/pipeline/execute.rs
View file

@ -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),
};

82
lib/components/fabro-workflow/src/pipeline/execute/tests.rs
View file

@ -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<MockSandbox>,
}
#[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.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<dyn Sandbox> =
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<dyn Sandbox> = 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();

2
lib/components/fabro-workflow/src/pipeline/initialize.rs
View file

@ -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

32
lib/foundation/fabro-core/src/error.rs
View file

@ -55,6 +55,12 @@ pub enum Error {
StallTimeout { node_id: String },
#[error("{detail}")]
Handler { detail: Box<HandlerErrorDetail> },
#[error("{message}")]
Context {
message: String,
#[source]
source: Box<dyn std::error::Error + Send + Sync + 'static>,
},
#[error("{0}")]
Other(String),
}
@ -72,6 +78,16 @@ impl Error {
}
}
pub fn context(
message: impl Into<String>,
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<T> = std::result::Result<T, Error>;
#[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 {