use agent_contracts::backend::{
BackendEndpoint, BackendId, BackendInstance, BackendInstanceId, BackendLifecycleState,
BackendPath, BackendProviderKind, BackendResourceAllocation, BackendRuntimeCapabilities,
OperationBackend, OperationError,
};
use serde::Deserialize;
use serde_json::{json, Map, Value};
use std::collections::BTreeMap;
use std::future::Future;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio_util::io::ReaderStream;
use super::backend::{
envd_host, join_url, normalize_backend_path, E2bBackendState, E2bLifecycle,
E2bOperationBackend, DEFAULT_API_BASE, DEFAULT_DOMAIN, DEFAULT_ENVD_PORT, DEFAULT_HOME_DIR,
DEFAULT_SHELL, DEFAULT_TEMPLATE_ID, DEFAULT_TEMP_ROOT, DEFAULT_TIMEOUT_SECS,
DEFAULT_WORKSPACE_ROOT, E2B_PROVIDER_KIND,
};
use super::error::E2bFailure;
use super::exec::E2bExec;
use crate::backend::BackendError;
const E2B_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
const WORKSPACE_INIT_MAX_ATTEMPTS: usize = 6;
const WORKSPACE_INIT_BASE_DELAY_MS: u64 = 200;
const WORKSPACE_INIT_MAX_DELAY_MS: u64 = 2_000;
pub(crate) struct E2bCreateBackendInput {
pub(crate) backend_id: BackendId,
pub(crate) session_id_for_instance: String,
pub(crate) workspace_root_text: String,
pub(crate) provider_options: Value,
pub(crate) resource_limits: agent_contracts::backend::BackendResourceLimits,
pub(crate) metadata: Value,
pub(crate) bootstrap: Option<Arc<super::bootstrap::E2bBootstrapArchive>>,
}
pub(crate) struct E2bCreatedBackend {
pub(crate) instance: BackendInstance,
pub(crate) backend: Arc<dyn OperationBackend>,
}
pub(crate) struct E2bSnapshotInput {
pub(crate) provider_options: Value,
pub(crate) sandbox_id: String,
pub(crate) name: Option<String>,
}
pub(crate) struct E2bDeleteSnapshotInput {
pub(crate) provider_options: Value,
pub(crate) snapshot_id: String,
}
pub(crate) struct E2bDeleteSandboxInput {
pub(crate) provider_options: Value,
pub(crate) sandbox_id: String,
}
#[derive(Debug, Clone)]
pub(crate) struct E2bSnapshotResult {
pub(crate) snapshot_id: String,
pub(crate) names: Vec<String>,
}
#[derive(Debug, Clone, Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
struct E2bProviderOptions {
api_key: Option<String>,
api_key_env: Option<String>,
#[serde(alias = "apiBase", alias = "api_url", alias = "apiUrl")]
api_base: Option<String>,
#[serde(alias = "sandbox_domain", alias = "sandboxDomain")]
domain: Option<String>,
#[serde(alias = "templateID", alias = "template")]
template_id: Option<String>,
#[serde(alias = "timeout")]
timeout_secs: Option<u64>,
secure: Option<bool>,
#[serde(alias = "allowInternetAccess")]
allow_internet_access: Option<bool>,
#[serde(alias = "autoPause")]
auto_pause: Option<bool>,
#[serde(alias = "autoResume")]
auto_resume: Option<bool>,
#[serde(alias = "envdPort")]
envd_port: Option<u16>,
#[serde(alias = "envdScheme")]
envd_scheme: Option<String>,
#[serde(alias = "workspaceRoot", alias = "remoteWorkspaceRoot")]
workspace_root: Option<String>,
#[serde(alias = "homeDir")]
home_dir: Option<String>,
#[serde(alias = "tempRoot")]
temp_root: Option<String>,
#[serde(alias = "defaultShell")]
default_shell: Option<String>,
#[serde(alias = "envdFileUpload")]
envd_file_upload: Option<String>,
username: Option<String>,
metadata: Option<BTreeMap<String, String>>,
#[serde(alias = "envVars")]
env_vars: Option<BTreeMap<String, String>>,
network: Option<Value>,
mcp: Option<Value>,
#[serde(alias = "volumeMounts")]
volume_mounts: Option<Value>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct CreateSandboxResponse {
#[serde(rename = "sandboxID")]
sandbox_id: String,
#[serde(rename = "templateID")]
template_id: String,
#[serde(rename = "envdAccessToken")]
envd_access_token: Option<String>,
#[serde(rename = "trafficAccessToken")]
traffic_access_token: Option<String>,
}
#[derive(Debug, Deserialize)]
struct CreateSnapshotResponse {
#[serde(rename = "snapshotID")]
snapshot_id: String,
#[serde(default)]
names: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct E2bConnectionOptions {
api_base: String,
sandbox_domain: String,
}
pub(crate) async fn create_backend(
input: E2bCreateBackendInput,
) -> Result<E2bCreatedBackend, BackendError> {
let bootstrap = input.bootstrap.clone();
let options = parse_options(&input.provider_options)?;
let api_key = resolve_api_key(&options)?;
let http = new_e2b_http_client()?;
let connection = resolve_connection_options(&options)?;
let api_base = connection.api_base;
let sandbox_domain = connection.sandbox_domain;
let template_id = options
.template_id
.as_deref()
.filter(|value| !value.trim().is_empty())
.unwrap_or(DEFAULT_TEMPLATE_ID)
.to_string();
let workspace_root = backend_path(
options
.workspace_root
.as_deref()
.unwrap_or(DEFAULT_WORKSPACE_ROOT),
)?;
let home_dir = options
.home_dir
.as_deref()
.map(backend_path)
.transpose()?
.or_else(|| Some(BackendPath(DEFAULT_HOME_DIR.to_string())));
let temp_root = backend_path(options.temp_root.as_deref().unwrap_or(DEFAULT_TEMP_ROOT))?;
let envd_port = options.envd_port.unwrap_or(DEFAULT_ENVD_PORT);
let envd_scheme = options
.envd_scheme
.clone()
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| "https".to_string());
let created = create_e2b_sandbox(
&http,
api_base.as_str(),
api_key.as_str(),
&template_id,
&options,
&input,
)
.await?;
let now = current_time_ms();
let backend_id = input.backend_id;
let endpoint = provider_handle(
&created,
envd_port,
envd_scheme.as_str(),
sandbox_domain.as_str(),
);
let instance = BackendInstance {
backend_id: backend_id.clone(),
provider: BackendProviderKind(E2B_PROVIDER_KIND.to_string()),
instance_id: BackendInstanceId(created.sandbox_id.clone()),
session_id: input.session_id_for_instance,
state: BackendLifecycleState::Active,
workspace_root: workspace_root.clone(),
endpoint: Some(endpoint),
snapshot: None,
capabilities: BackendRuntimeCapabilities {
supports_exec: true,
supports_file_read: true,
supports_file_write: true,
supports_search: true,
supports_export_file: true,
supports_lsp: false,
supports_pause: false,
supports_snapshot: true,
supports_delete: true,
},
resources: BackendResourceAllocation {
vcpu_count: input.resource_limits.vcpu_count,
memory_mb: input.resource_limits.memory_mb,
disk_mb: input.resource_limits.disk_mb,
},
metadata: metadata_for_instance(
input.metadata,
&input.provider_options,
&created,
&options,
),
created_at_ms: now,
updated_at_ms: now,
};
let state = Arc::new(E2bBackendState {
backend_id: backend_id.0,
api_base,
api_key,
sandbox_id: created.sandbox_id,
sandbox_domain,
envd_access_token: created.envd_access_token,
envd_port,
envd_scheme,
workspace_root,
home_dir,
temp_root,
default_shell: Some(
options
.default_shell
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| DEFAULT_SHELL.to_string()),
),
username: options.username.filter(|value| !value.trim().is_empty()),
envd_file_upload_multipart: options
.envd_file_upload
.as_deref()
.is_some_and(|value| value.trim().eq_ignore_ascii_case("multipart")),
http,
lifecycle: Mutex::new(E2bLifecycle::Active),
});
let backend = Arc::new(E2bOperationBackend::new(Arc::clone(&state)));
let initialization = async {
ensure_remote_roots(&state).await?;
if let Some(bootstrap) = bootstrap.as_deref() {
install_bootstrap_archive(&state, bootstrap).await?;
}
Ok::<(), OperationError>(())
}
.await;
if let Err(error) = initialization {
if let Err(cleanup_error) = state.delete_sandbox().await {
tracing::warn!(
operation = "cleanup_failed_workspace_initialization",
sandbox_id = %state.sandbox_id,
error = %cleanup_error,
"Failed to delete E2B sandbox after workspace initialization failure"
);
}
return Err(BackendError::BuildFailed {
message: format!("e2b sandbox created but bootstrap initialization failed: {error}"),
});
}
Ok(E2bCreatedBackend { instance, backend })
}
pub(crate) async fn create_snapshot(
input: E2bSnapshotInput,
) -> Result<E2bSnapshotResult, BackendError> {
let options = parse_options(&input.provider_options)?;
let api_key = resolve_api_key(&options)?;
let api_base = resolve_connection_options(&options)?.api_base;
let http = new_e2b_http_client()?;
let mut body = Map::new();
if let Some(name) = input
.name
.as_deref()
.map(str::trim)
.filter(|name| !name.is_empty())
{
body.insert("name".to_string(), Value::String(name.to_string()));
}
let response = http
.post(join_url(
api_base.as_str(),
format!("/sandboxes/{}/snapshots", input.sandbox_id).as_str(),
))
.header("X-API-Key", api_key)
.json(&Value::Object(body))
.send()
.await
.map_err(|error| BackendError::BuildFailed {
message: format!("failed to create e2b snapshot: {error}"),
})?;
if response.status() != reqwest::StatusCode::CREATED {
let status = response.status();
let text = response.text().await.unwrap_or_default();
let message = super::backend::parse_error_message(text.as_str()).unwrap_or(text);
return Err(BackendError::BuildFailed {
message: format!("e2b create snapshot failed with HTTP {status}: {message}"),
});
}
let parsed = response
.json::<CreateSnapshotResponse>()
.await
.map_err(|error| BackendError::BuildFailed {
message: format!("failed to decode e2b create snapshot response: {error}"),
})?;
Ok(E2bSnapshotResult {
snapshot_id: parsed.snapshot_id,
names: parsed.names,
})
}
pub(crate) async fn delete_snapshot(input: E2bDeleteSnapshotInput) -> Result<bool, BackendError> {
let snapshot_id = input.snapshot_id.trim();
if snapshot_id.is_empty() {
return Err(BackendError::InvalidRequest {
message: "e2b snapshot id cannot be empty".to_string(),
});
}
let options = parse_options(&input.provider_options)?;
let api_key = resolve_api_key(&options)?;
let api_base = resolve_connection_options(&options)?.api_base;
let http = new_e2b_http_client()?;
let response = http
.delete(join_url(
api_base.as_str(),
format!("/templates/{}", encode_path_segment(snapshot_id)).as_str(),
))
.header("X-API-Key", api_key)
.send()
.await
.map_err(|error| BackendError::BuildFailed {
message: format!("failed to delete e2b snapshot template: {error}"),
})?;
if response.status() == reqwest::StatusCode::NO_CONTENT {
return Ok(true);
}
if response.status() == reqwest::StatusCode::NOT_FOUND {
return Ok(false);
}
let status = response.status();
let text = response.text().await.unwrap_or_default();
let message = super::backend::parse_error_message(text.as_str()).unwrap_or(text);
Err(BackendError::BuildFailed {
message: format!("e2b delete snapshot template failed with HTTP {status}: {message}"),
})
}
pub(crate) async fn delete_sandbox_by_id(input: E2bDeleteSandboxInput) -> Result<(), BackendError> {
let sandbox_id = input.sandbox_id.trim();
if sandbox_id.is_empty() {
return Err(BackendError::InvalidRequest {
message: "e2b sandbox id cannot be empty".to_string(),
});
}
let options = parse_options(&input.provider_options)?;
let api_key = resolve_api_key(&options)?;
let api_base = resolve_connection_options(&options)?.api_base;
let http = new_e2b_http_client()?;
let response = http
.delete(join_url(
api_base.as_str(),
format!("/sandboxes/{}", encode_path_segment(sandbox_id)).as_str(),
))
.header("X-API-Key", api_key)
.send()
.await
.map_err(|error| BackendError::BuildFailed {
message: format!("failed to delete e2b sandbox: {error}"),
})?;
if response.status() == reqwest::StatusCode::NO_CONTENT
|| response.status() == reqwest::StatusCode::NOT_FOUND
{
return Ok(());
}
let status = response.status();
let text = response.text().await.unwrap_or_default();
let message = super::backend::parse_error_message(text.as_str()).unwrap_or(text);
Err(BackendError::BuildFailed {
message: format!("e2b delete sandbox failed with HTTP {status}: {message}"),
})
}
fn parse_options(value: &Value) -> Result<E2bProviderOptions, BackendError> {
let value = if value.is_null() {
Value::Object(Map::new())
} else {
value.clone()
};
serde_json::from_value(value).map_err(|error| BackendError::InvalidRequest {
message: format!("invalid e2b backend options: {error}"),
})
}
fn resolve_connection_options(
options: &E2bProviderOptions,
) -> Result<E2bConnectionOptions, BackendError> {
let api_url_env = std::env::var("E2B_API_URL").ok();
let domain_env = std::env::var("E2B_DOMAIN").ok();
resolve_connection_options_from_values(options, api_url_env.as_deref(), domain_env.as_deref())
}
fn resolve_connection_options_from_values(
options: &E2bProviderOptions,
api_url_env: Option<&str>,
domain_env: Option<&str>,
) -> Result<E2bConnectionOptions, BackendError> {
let raw_domain = non_empty(options.domain.as_deref())
.or_else(|| non_empty(domain_env))
.unwrap_or(DEFAULT_DOMAIN);
let sandbox_domain = normalize_sandbox_domain(raw_domain)?;
let api_base = non_empty(options.api_base.as_deref())
.or_else(|| non_empty(api_url_env))
.map(str::to_string)
.unwrap_or_else(|| {
if sandbox_domain == DEFAULT_DOMAIN {
DEFAULT_API_BASE.to_string()
} else {
format!("https://api.{sandbox_domain}")
}
});
Ok(E2bConnectionOptions {
api_base,
sandbox_domain,
})
}
fn non_empty(value: Option<&str>) -> Option<&str> {
value.map(str::trim).filter(|value| !value.is_empty())
}
fn normalize_sandbox_domain(value: &str) -> Result<String, BackendError> {
let domain = value.trim().trim_end_matches('.');
if domain.is_empty()
|| domain.contains("://")
|| domain.contains('/')
|| domain.contains('?')
|| domain.contains('#')
{
return Err(BackendError::InvalidRequest {
message: format!(
"invalid e2b domain {value:?}; expected a hostname without scheme or path"
),
});
}
Ok(domain.to_string())
}
fn resolve_api_key(options: &E2bProviderOptions) -> Result<String, BackendError> {
if let Some(api_key) = options
.api_key
.as_deref()
.filter(|value| !value.trim().is_empty())
{
return Ok(api_key.to_string());
}
let env_name = options
.api_key_env
.as_deref()
.filter(|value| !value.trim().is_empty())
.unwrap_or("E2B_API_KEY");
std::env::var(env_name)
.ok()
.filter(|value| !value.trim().is_empty())
.ok_or_else(|| BackendError::InvalidRequest {
message: format!("e2b backend requires api_key or non-empty env var {env_name}"),
})
}
pub(crate) fn resolve_api_key_from_options(options: &Value) -> Option<String> {
let parsed = parse_options(options).ok()?;
resolve_api_key(&parsed).ok()
}
fn backend_path(value: &str) -> Result<BackendPath, BackendError> {
normalize_backend_path(std::path::Path::new(value)).map_err(|error| {
BackendError::InvalidRequest {
message: error.to_string(),
}
})
}
async fn create_e2b_sandbox(
http: &reqwest::Client,
api_base: &str,
api_key: &str,
template_id: &str,
options: &E2bProviderOptions,
input: &E2bCreateBackendInput,
) -> Result<CreateSandboxResponse, BackendError> {
let timeout_secs = options
.timeout_secs
.or_else(|| input.resource_limits.timeout_ms.map(|ms| ms / 1000))
.unwrap_or(DEFAULT_TIMEOUT_SECS);
let mut body = Map::new();
body.insert(
"templateID".to_string(),
Value::String(template_id.to_string()),
);
body.insert("timeout".to_string(), json!(timeout_secs));
body.insert("secure".to_string(), json!(options.secure.unwrap_or(true)));
if let Some(value) = options.allow_internet_access {
body.insert("allow_internet_access".to_string(), json!(value));
}
if let Some(value) = options.auto_pause {
body.insert("autoPause".to_string(), json!(value));
}
if let Some(value) = options.auto_resume {
body.insert("autoResume".to_string(), json!({ "enabled": value }));
}
let metadata = platform_metadata(input, options);
if !metadata.is_empty() {
body.insert(
"metadata".to_string(),
serde_json::to_value(metadata).unwrap(),
);
}
if let Some(env_vars) = options
.env_vars
.as_ref()
.filter(|values| !values.is_empty())
{
body.insert(
"envVars".to_string(),
serde_json::to_value(env_vars).unwrap(),
);
}
if let Some(network) = options.network.clone() {
body.insert("network".to_string(), network);
}
if let Some(mcp) = options.mcp.clone() {
body.insert("mcp".to_string(), mcp);
}
if let Some(volume_mounts) = options.volume_mounts.clone() {
body.insert("volumeMounts".to_string(), volume_mounts);
}
let started_at = Instant::now();
let response = http
.post(join_url(api_base, "/sandboxes"))
.header("X-API-Key", api_key)
.json(&Value::Object(body))
.send()
.await
.map_err(|error| {
let failure = E2bFailure::from_reqwest("failed to create e2b sandbox", &error);
failure.log(
"create_sandbox",
None,
None,
started_at.elapsed().as_millis(),
);
BackendError::BuildFailed {
message: failure.message,
}
})?;
if response.status() != reqwest::StatusCode::CREATED {
let status = response.status();
let text = response.text().await.unwrap_or_default();
let message = super::backend::parse_error_message(text.as_str()).unwrap_or(text);
let failure = E2bFailure::from_status("e2b create sandbox", status, message.clone());
failure.log(
"create_sandbox",
None,
None,
started_at.elapsed().as_millis(),
);
return Err(BackendError::BuildFailed {
message: failure.message,
});
}
response
.json::<CreateSandboxResponse>()
.await
.map_err(|error| BackendError::BuildFailed {
message: format!("failed to decode e2b create sandbox response: {error}"),
})
}
fn platform_metadata(
input: &E2bCreateBackendInput,
options: &E2bProviderOptions,
) -> BTreeMap<String, String> {
let mut metadata = BTreeMap::new();
if let Some(values) = &options.metadata {
metadata.extend(values.clone());
}
if let Some(object) = input.metadata.as_object() {
for (key, value) in object {
if let Some(value) = value.as_str() {
metadata.insert(key.clone(), value.to_string());
}
}
}
metadata.insert("xiaoo_backend_id".to_string(), input.backend_id.0.clone());
metadata.insert(
"xiaoo_session_id".to_string(),
input.session_id_for_instance.clone(),
);
metadata.insert(
"xiaoo_host_workspace".to_string(),
input.workspace_root_text.clone(),
);
metadata
}
fn metadata_for_instance(
metadata: Value,
provider_options: &Value,
sandbox: &CreateSandboxResponse,
options: &E2bProviderOptions,
) -> Value {
let mut object = match metadata {
Value::Object(object) => object,
Value::Null => Map::new(),
other => {
let mut object = Map::new();
object.insert("user_metadata".to_string(), other);
object
}
};
object.insert("provider".to_string(), Value::String("e2b".to_string()));
object.insert(
"sandbox_id".to_string(),
Value::String(sandbox.sandbox_id.clone()),
);
object.insert(
"template_id".to_string(),
Value::String(sandbox.template_id.clone()),
);
if sandbox.traffic_access_token.is_some() {
object.insert(
"traffic_access_token_present".to_string(),
Value::Bool(true),
);
}
object.insert(
"provider_options".to_string(),
redacted_provider_options(provider_options, options),
);
Value::Object(object)
}
fn redacted_provider_options(provider_options: &Value, options: &E2bProviderOptions) -> Value {
let mut object = provider_options.as_object().cloned().unwrap_or_default();
object.remove("api_key");
object.remove("env_vars");
object.remove("envVars");
if options.api_key.is_some() {
object.insert("api_key_configured".to_string(), Value::Bool(true));
}
Value::Object(object)
}
fn provider_handle(
sandbox: &CreateSandboxResponse,
envd_port: u16,
envd_scheme: &str,
sandbox_domain: &str,
) -> BackendEndpoint {
BackendEndpoint::ProviderHandle {
value: json!({
"provider": "e2b",
"sandbox_id": sandbox.sandbox_id.clone(),
"envd_host": envd_host(envd_port, sandbox.sandbox_id.as_str(), sandbox_domain),
"envd_port": envd_port,
"envd_scheme": envd_scheme,
}),
}
}
async fn ensure_remote_roots(state: &Arc<E2bBackendState>) -> Result<(), OperationError> {
let exec = E2bExec::new(Arc::clone(state));
let script = format!(
"mkdir -p {} {} {}",
super::backend::shell_quote(state.workspace_root.0.as_str()),
super::backend::shell_quote(state.temp_root.0.as_str()),
super::backend::shell_quote("/home/user/.xiaoo")
);
let output = retry_workspace_initialization(
state.sandbox_id.as_str(),
|| exec.run_shell_script_detailed(script.as_str(), None),
tokio::time::sleep,
)
.await
.map_err(super::exec::E2bExecFailure::into_operation_error)?;
if output.exit_code == Some(0) {
return Ok(());
}
Err(OperationError::ExecutionFailed {
message: String::from_utf8_lossy(output.stderr.as_slice()).to_string(),
})
}
async fn install_bootstrap_archive(
state: &Arc<E2bBackendState>,
archive: &super::bootstrap::E2bBootstrapArchive,
) -> Result<(), OperationError> {
use reqwest::header::ACCEPT;
use reqwest::multipart::{Form, Part};
use reqwest::Method;
let nonce = uuid::Uuid::new_v4().simple().to_string();
let remote_archive = format!("/tmp/xiaoo-bootstrap-{nonce}.tar");
let remote_manifest = format!("/home/user/.xiaoo/.bootstrap-manifest-{nonce}.json");
let file = tokio::fs::File::open(archive.path())
.await
.map_err(|error| OperationError::Transport {
message: format!("failed to open E2B bootstrap archive: {error}"),
})?;
let stream = ReaderStream::new(file);
let archive_part =
Part::stream_with_length(reqwest::Body::wrap_stream(stream), archive.size_bytes())
.file_name(remote_archive.clone());
let response = state
.envd_request(Method::POST, "/files")
.query(&[("path", remote_archive.as_str())])
.header(ACCEPT, "application/json")
.multipart(Form::new().part("file", archive_part))
.send()
.await
.map_err(|error| OperationError::Transport {
message: format!("failed to stream E2B bootstrap archive: {error}"),
})?;
if !response.status().is_success() {
return Err(super::backend::http_error("upload E2B bootstrap archive", response).await);
}
let manifest = archive
.manifest_json()
.map_err(|error| OperationError::Transport {
message: format!("failed to encode E2B bootstrap manifest: {error}"),
})?;
let manifest_part = Part::bytes(manifest).file_name(remote_manifest.clone());
let response = state
.envd_request(Method::POST, "/files")
.query(&[("path", remote_manifest.as_str())])
.header(ACCEPT, "application/json")
.multipart(Form::new().part("file", manifest_part))
.send()
.await
.map_err(|error| OperationError::Transport {
message: format!("failed to upload E2B bootstrap manifest: {error}"),
})?;
if !response.status().is_success() {
return Err(super::backend::http_error("upload E2B bootstrap manifest", response).await);
}
let q = super::backend::shell_quote;
let stage_pattern = format!("/home/user/.xiaoo/bootstrap-stage-{nonce}.XXXXXX");
let workspace_target = "/home/user/workspace";
let skills_target = "/home/user/.xiaoo/skills";
let workspace_old = format!("/home/user/.xiaoo/workspace-old-{nonce}");
let skills_old = format!("/home/user/.xiaoo/skills-old-{nonce}");
let manifest_target = "/home/user/.xiaoo/bootstrap/manifest.json";
let script = format!(
r#"set -eu
archive={archive}
manifest={manifest}
expected={expected}
actual=$(sha256sum "$archive" | awk '{{print $1}}')
[ "$actual" = "$expected" ] || {{ echo "bootstrap archive SHA-256 mismatch" >&2; exit 41; }}
stage=$(mktemp -d {stage_pattern})
tar --no-same-owner --same-permissions -xf "$archive" -C "$stage"
[ -d "$stage/workspace" ] || {{ echo "bootstrap workspace payload missing" >&2; exit 42; }}
[ -d "$stage/skills" ] || {{ echo "bootstrap skills payload missing" >&2; exit 43; }}
workspace_target={workspace_target}
skills_target={skills_target}
workspace_old={workspace_old}
skills_old={skills_old}
had_workspace=0
had_skills=0
workspace_installed=0
skills_installed=0
rollback() {{
set +e
[ "$skills_installed" = 0 ] || rm -rf "$skills_target"
[ "$had_skills" = 0 ] || mv "$skills_old" "$skills_target"
[ "$workspace_installed" = 0 ] || rm -rf "$workspace_target"
[ "$had_workspace" = 0 ] || mv "$workspace_old" "$workspace_target"
rm -rf "$stage" "$archive" "$manifest"
exit 44
}}
rm -rf "$workspace_old" "$skills_old"
if [ -e "$workspace_target" ]; then mv "$workspace_target" "$workspace_old"; had_workspace=1; fi
mv "$stage/workspace" "$workspace_target" || rollback
workspace_installed=1
if [ -e "$skills_target" ]; then mv "$skills_target" "$skills_old"; had_skills=1; fi
mv "$stage/skills" "$skills_target" || rollback
skills_installed=1
mkdir -p $(dirname {manifest_target}) || rollback
mv "$manifest" {manifest_target} || rollback
rm -rf "$workspace_old" "$skills_old" "$stage" "$archive" || true
"#,
archive = q(&remote_archive),
manifest = q(&remote_manifest),
expected = q(archive.sha256()),
stage_pattern = q(&stage_pattern),
workspace_target = q(workspace_target),
skills_target = q(skills_target),
workspace_old = q(&workspace_old),
skills_old = q(&skills_old),
manifest_target = q(manifest_target),
);
let exec = E2bExec::new(Arc::clone(state));
let output = exec.run_shell_script(script.as_str(), None).await?;
if output.exit_code == Some(0) {
return Ok(());
}
Err(OperationError::ExecutionFailed {
message: format!(
"E2B bootstrap install failed (exit {:?}): {}",
output.exit_code,
String::from_utf8_lossy(&output.stderr)
),
})
}
async fn retry_workspace_initialization<Attempt, AttemptFuture, Sleep, SleepFuture>(
sandbox_id: &str,
mut operation: Attempt,
mut sleep: Sleep,
) -> Result<super::exec::E2bExecOutput, super::exec::E2bExecFailure>
where
Attempt: FnMut() -> AttemptFuture,
AttemptFuture: Future<Output = Result<super::exec::E2bExecOutput, super::exec::E2bExecFailure>>,
Sleep: FnMut(Duration) -> SleepFuture,
SleepFuture: Future<Output = ()>,
{
for attempt in 1..=WORKSPACE_INIT_MAX_ATTEMPTS {
match operation().await {
Ok(output) => return Ok(output),
Err(error) if error.retryable() && attempt < WORKSPACE_INIT_MAX_ATTEMPTS => {
let delay = workspace_init_backoff(attempt);
tracing::warn!(
operation = "initialize_workspace",
sandbox_id,
attempt,
max_attempts = WORKSPACE_INIT_MAX_ATTEMPTS,
retry_delay_ms = delay.as_millis(),
error = %error.message(),
"E2B workspace initialization failed; retrying"
);
sleep(delay).await;
}
Err(error) => return Err(error),
}
}
unreachable!("workspace initialization retry loop always returns")
}
fn workspace_init_backoff(attempt: usize) -> Duration {
let exponent = u32::try_from(attempt.saturating_sub(1)).unwrap_or(u32::MAX);
let base = WORKSPACE_INIT_BASE_DELAY_MS
.saturating_mul(2u64.saturating_pow(exponent))
.min(WORKSPACE_INIT_MAX_DELAY_MS);
let spread = base / 5;
let jitter = if spread == 0 {
0
} else {
rand::random::<u64>() % (spread.saturating_mul(2).saturating_add(1))
};
Duration::from_millis(base.saturating_sub(spread).saturating_add(jitter))
}
fn new_e2b_http_client() -> Result<reqwest::Client, BackendError> {
reqwest::Client::builder()
.connect_timeout(E2B_CONNECT_TIMEOUT)
.build()
.map_err(|error| BackendError::BuildFailed {
message: format!("failed to build e2b HTTP client: {error}"),
})
}
fn current_time_ms() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|duration| duration.as_millis() as u64)
.unwrap_or(0)
}
fn encode_path_segment(value: &str) -> String {
let mut encoded = String::new();
for byte in value.as_bytes() {
match *byte {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
encoded.push(*byte as char)
}
_ => encoded.push_str(format!("%{byte:02X}").as_str()),
}
}
encoded
}
#[cfg(test)]
mod tests {
use super::*;
use agent_contracts::backend::capability::exec::ExecRequest;
use agent_contracts::backend::capability::filesystem::{WriteBytesRequest, WriteMode};
use std::cell::Cell;
use std::future::ready;
#[tokio::test]
#[ignore = "requires E2B_API_KEY and creates a real E2B sandbox"]
async fn live_structured_grep_executes_command_with_args() {
assert!(
std::env::var_os("E2B_API_KEY").is_some(),
"E2B_API_KEY must be set"
);
let suffix = uuid::Uuid::new_v4().simple().to_string();
let created = create_backend(E2bCreateBackendInput {
backend_id: BackendId(format!("e2b-live-grep:{suffix}")),
session_id_for_instance: format!("e2b-live-grep-session:{suffix}"),
workspace_root_text: DEFAULT_WORKSPACE_ROOT.to_string(),
provider_options: json!({
"api_key_env": "E2B_API_KEY",
"template_id": "base",
"timeout_secs": 300,
"default_shell": "/bin/sh"
}),
resource_limits: Default::default(),
metadata: json!({"purpose": "structured grep live smoke"}),
bootstrap: None,
})
.await
.expect("create live E2B backend");
let backend = created.backend;
let smoke_path = BackendPath(format!("{DEFAULT_WORKSPACE_ROOT}/grep-smoke.py"));
let smoke_result: Result<String, String> = async {
backend
.files()
.write_bytes(WriteBytesRequest {
path: smoke_path,
content: b"watt = watts = W = Quantity(\"watt\")\n".to_vec(),
mode: WriteMode::Overwrite,
})
.await
.map_err(|error| format!("write smoke fixture: {error}"))?;
let output = backend
.exec()
.exec(ExecRequest {
command: "grep".to_string(),
args: vec![
"-P".to_string(),
"-n".to_string(),
"-e".to_string(),
r"W\s*=".to_string(),
"grep-smoke.py".to_string(),
],
cwd: Some(BackendPath(DEFAULT_WORKSPACE_ROOT.to_string())),
timeout_ms: Some(30_000),
..Default::default()
})
.await
.map_err(|error| format!("execute structured grep: {error}"))?;
let stdout = String::from_utf8_lossy(&output.stdout).to_string();
let stderr = String::from_utf8_lossy(&output.stderr);
if output.exit_code != Some(0) {
return Err(format!(
"grep exited with {:?}; stderr: {stderr}",
output.exit_code
));
}
if !stdout.contains("1:watt = watts = W = Quantity") {
return Err(format!("unexpected grep stdout: {stdout:?}"));
}
Ok(stdout)
}
.await;
let cleanup_result = backend.shutdown().await;
cleanup_result.expect("delete live E2B sandbox");
let stdout = smoke_result.expect("structured grep smoke");
eprintln!("live E2B structured grep passed: {}", stdout.trim());
}
#[tokio::test]
async fn workspace_initialization_retries_transient_failures() {
let attempts = Cell::new(0usize);
let sleeps = Cell::new(0usize);
let output = retry_workspace_initialization(
"sandbox-test",
|| {
let attempt = attempts.get() + 1;
attempts.set(attempt);
ready(if attempt < 3 {
Err(super::super::exec::E2bExecFailure::retryable_for_test(
"temporary reset",
))
} else {
Ok(super::super::exec::E2bExecOutput {
stdout: Vec::new(),
stderr: Vec::new(),
exit_code: Some(0),
timed_out: false,
})
})
},
|_| {
sleeps.set(sleeps.get() + 1);
ready(())
},
)
.await
.expect("third attempt should succeed");
assert_eq!(output.exit_code, Some(0));
assert_eq!(attempts.get(), 3);
assert_eq!(sleeps.get(), 2);
}
#[test]
fn workspace_backoff_is_bounded_and_jittered() {
for attempt in 1..=WORKSPACE_INIT_MAX_ATTEMPTS {
let delay_ms = workspace_init_backoff(attempt).as_millis() as u64;
let exponent = u32::try_from(attempt.saturating_sub(1)).unwrap_or(u32::MAX);
let base = WORKSPACE_INIT_BASE_DELAY_MS
.saturating_mul(2u64.saturating_pow(exponent))
.min(WORKSPACE_INIT_MAX_DELAY_MS);
assert!(delay_ms >= base - base / 5);
assert!(delay_ms <= base + base / 5);
}
}
#[test]
fn redacts_direct_api_key_from_metadata() {
let options = parse_options(&json!({
"api_key": "secret",
"template_id": "base",
"envVars": {"TOKEN": "secret"}
}))
.expect("options");
let redacted = redacted_provider_options(
&json!({
"api_key": "secret",
"template_id": "base",
"envVars": {"TOKEN": "secret"}
}),
&options,
);
let object = redacted.as_object().expect("object");
assert!(!object.contains_key("api_key"));
assert!(!object.contains_key("envVars"));
assert_eq!(object.get("api_key_configured"), Some(&Value::Bool(true)));
}
#[test]
fn default_template_is_base() {
let options = parse_options(&json!({})).expect("options");
assert_eq!(
options
.template_id
.as_deref()
.unwrap_or(DEFAULT_TEMPLATE_ID),
"base"
);
}
#[test]
fn connection_options_default_to_e2b_cloud() {
let options = parse_options(&json!({})).expect("options");
let connection = resolve_connection_options_from_values(&options, None, None)
.expect("connection options");
assert_eq!(connection.api_base, DEFAULT_API_BASE);
assert_eq!(connection.sandbox_domain, DEFAULT_DOMAIN);
}
#[test]
fn connection_options_derive_api_url_from_self_hosted_domain() {
let options = parse_options(&json!({})).expect("options");
let connection = resolve_connection_options_from_values(
&options,
None,
Some(" self-hosted.example.com. "),
)
.expect("connection options");
assert_eq!(connection.api_base, "https://api.self-hosted.example.com");
assert_eq!(connection.sandbox_domain, "self-hosted.example.com");
}
#[test]
fn connection_options_accept_api_url_and_domain_environment_values() {
let options = parse_options(&json!({})).expect("options");
let connection = resolve_connection_options_from_values(
&options,
Some(" https://control.self-hosted.example.com/ "),
Some("self-hosted.example.com"),
)
.expect("connection options");
assert_eq!(
connection.api_base,
"https://control.self-hosted.example.com/"
);
assert_eq!(connection.sandbox_domain, "self-hosted.example.com");
}
#[test]
fn explicit_connection_options_override_environment_values() {
let options = parse_options(&json!({
"apiUrl": "https://control.internal.example.com/",
"sandboxDomain": "sandboxes.internal.example.com"
}))
.expect("options");
let connection = resolve_connection_options_from_values(
&options,
Some("https://api.from-env.example.com"),
Some("from-env.example.com"),
)
.expect("connection options");
assert_eq!(connection.api_base, "https://control.internal.example.com/");
assert_eq!(connection.sandbox_domain, "sandboxes.internal.example.com");
}
#[test]
fn rejects_domain_with_scheme() {
let options = parse_options(&json!({
"domain": "https://self-hosted.example.com"
}))
.expect("options");
let error = resolve_connection_options_from_values(&options, None, None)
.expect_err("domain with a scheme must be rejected");
assert!(error
.to_string()
.contains("hostname without scheme or path"));
}
#[test]
fn provider_handle_uses_configured_sandbox_domain() {
let sandbox = CreateSandboxResponse {
sandbox_id: "sandbox-test".to_string(),
template_id: "base".to_string(),
envd_access_token: Some("access-token".to_string()),
traffic_access_token: None,
};
let endpoint = provider_handle(&sandbox, 49_983, "https", "self-hosted.example.com");
let BackendEndpoint::ProviderHandle { value } = endpoint else {
panic!("expected provider handle");
};
assert_eq!(
value["envd_host"],
"49983-sandbox-test.self-hosted.example.com"
);
}
#[test]
fn encodes_template_id_as_single_path_segment() {
assert_eq!(
encode_path_segment("team/fork-test:default"),
"team%2Ffork-test%3Adefault"
);
}
}