use agent_workspace_contract::*;
use anyhow::{Context, Result};
use async_trait::async_trait;
use std::collections::BTreeSet;
use std::path::{Component, Path, PathBuf};
use std::process::Stdio;
use std::sync::Arc;
use tokio::io::AsyncReadExt;
#[derive(Debug, Clone)]
pub struct LocalEnvironmentProvider {
root: Arc<PathBuf>,
}
impl LocalEnvironmentProvider {
pub fn for_local_development(root: impl Into<PathBuf>) -> Result<Self> {
let root = root.into();
std::fs::create_dir_all(&root)?;
let root = std::fs::canonicalize(root)?;
for name in ["computers", "sandboxes", "snapshots"] {
std::fs::create_dir_all(root.join(name))?;
}
Ok(Self {
root: Arc::new(root),
})
}
fn resource(&self, kind: &str, id: &str) -> Result<PathBuf> {
anyhow::ensure!(
!id.is_empty()
&& id
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')),
"invalid local environment resource id"
);
Ok(self.root.join(kind).join(id))
}
fn reference(&self, kind: &str) -> ProviderResourceRef {
ProviderResourceRef {
provider: "local".into(),
external_id: format!("{kind}-{}", uuid::Uuid::new_v4().simple()),
}
}
fn computer_root(&self, external_id: &str) -> Result<PathBuf> {
self.resource("computers", external_id)
}
fn space_root(&self, computer_external_id: &str, space_directory: &str) -> Result<PathBuf> {
let relative = Path::new(space_directory);
anyhow::ensure!(
!relative.is_absolute()
&& relative
.components()
.all(|component| matches!(component, Component::Normal(_))),
"space directory escapes the computer root"
);
Ok(self.computer_root(computer_external_id)?.join(relative))
}
fn file_in_space(
&self,
computer_external_id: &str,
space_directory: &str,
path: &str,
) -> Result<PathBuf> {
let relative = Path::new(path);
anyhow::ensure!(
!relative.is_absolute()
&& relative
.components()
.all(|component| matches!(component, Component::Normal(_))),
"local Space path escapes its root"
);
Ok(self
.space_root(computer_external_id, space_directory)?
.join(relative))
}
}
#[async_trait]
impl ComputerProvider for LocalEnvironmentProvider {
fn kind(&self) -> &str {
"local"
}
fn capabilities(&self) -> ProviderCapabilities {
capabilities([
"computer.lifecycle",
"computer.snapshot",
"computer.files",
"computer.exec",
"space.change_set",
"space.restore",
"space.snapshot",
])
}
async fn create_computer(
&self,
_: &ResourceScope,
_: &ProviderComputerRequest,
_: &str,
) -> Result<ProviderResourceRef> {
let reference = self.reference("computer");
tokio::fs::create_dir_all(self.resource("computers", &reference.external_id)?).await?;
Ok(reference)
}
async fn delete_computer(&self, external_id: &str, _: &str) -> Result<()> {
remove_if_present(self.resource("computers", external_id)?).await
}
async fn create_space_directory(
&self,
computer_external_id: &str,
space_directory: &str,
request: &ProviderSpaceRequest,
_: &str,
) -> Result<()> {
let destination = self.space_root(computer_external_id, space_directory)?;
if let Some(snapshot) = request.source_snapshot_external_id.as_deref() {
let source = self.resource("snapshots", snapshot)?;
tokio::task::spawn_blocking(move || copy_tree(&source, &destination)).await??;
} else {
tokio::fs::create_dir_all(destination).await?;
}
Ok(())
}
async fn delete_space_directory(
&self,
computer_external_id: &str,
space_directory: &str,
_: &str,
) -> Result<()> {
remove_if_present(self.space_root(computer_external_id, space_directory)?).await
}
async fn read_file(
&self,
computer_external_id: &str,
space_directory: &str,
path: &str,
) -> Result<ProviderFileContent> {
Ok(ProviderFileContent {
bytes: tokio::fs::read(self.file_in_space(
computer_external_id,
space_directory,
path,
)?)
.await?,
provider_revision: None,
})
}
async fn write_file(
&self,
computer_external_id: &str,
space_directory: &str,
path: &str,
content: &[u8],
_: Option<&str>,
_: &str,
) -> Result<ProviderWriteResult> {
let file = self.file_in_space(computer_external_id, space_directory, path)?;
if let Some(parent) = file.parent() {
tokio::fs::create_dir_all(parent).await?;
}
tokio::fs::write(file, content).await?;
Ok(ProviderWriteResult {
provider_revision: None,
})
}
async fn create_change_set(
&self,
computer_external_id: &str,
source_space_directory: &str,
_: &str,
_: &str,
) -> Result<String> {
Ok(format!("{computer_external_id}:{source_space_directory}"))
}
async fn apply_change_set(
&self,
computer_external_id: &str,
source_space_directory: &str,
target_space_directory: &str,
provider_change_ref: &str,
_: &str,
) -> Result<ProviderWriteResult> {
anyhow::ensure!(
provider_change_ref == format!("{computer_external_id}:{source_space_directory}"),
"local ChangeSet source reference mismatch"
);
let source = self.space_root(computer_external_id, source_space_directory)?;
let target = self.space_root(computer_external_id, target_space_directory)?;
tokio::task::spawn_blocking(move || {
if target.exists() {
std::fs::remove_dir_all(&target)?;
}
copy_tree(&source, &target)
})
.await??;
Ok(ProviderWriteResult {
provider_revision: None,
})
}
async fn exec_command(
&self,
computer_external_id: &str,
space_directory: &str,
command: &CommandSpec,
) -> Result<ProviderCommandOutput> {
anyhow::ensure!(!command.argv.is_empty(), "argv must not be empty");
let cwd_path = match command.cwd.as_deref() {
None | Some("") | Some(".") => {
self.space_root(computer_external_id, space_directory)?
}
Some(cwd) => self.file_in_space(computer_external_id, space_directory, cwd)?,
};
let mut process = if command.shell {
let mut value = tokio::process::Command::new("sh");
value.arg("-lc").arg(command.argv.join(" "));
value
} else {
let mut value = tokio::process::Command::new(&command.argv[0]);
value.args(&command.argv[1..]);
value
};
process
.current_dir(cwd_path)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true);
let mut child = process.spawn()?;
let stdout = child.stdout.take();
let stderr = child.stderr.take();
let status = tokio::time::timeout(
std::time::Duration::from_millis(command.timeout_ms),
child.wait(),
)
.await
.map_err(|_| anyhow::anyhow!("command timed out"))??;
let mut stdout_bytes = Vec::new();
let mut stderr_bytes = Vec::new();
if let Some(mut stream) = stdout {
stream.read_to_end(&mut stdout_bytes).await?;
}
if let Some(mut stream) = stderr {
stream.read_to_end(&mut stderr_bytes).await?;
}
let truncated = stdout_bytes.len() > command.max_output_bytes
|| stderr_bytes.len() > command.max_output_bytes;
if stdout_bytes.len() > command.max_output_bytes {
stdout_bytes.truncate(command.max_output_bytes);
}
if stderr_bytes.len() > command.max_output_bytes {
stderr_bytes.truncate(command.max_output_bytes);
}
Ok(ProviderCommandOutput {
exit_code: status.code(),
stdout: String::from_utf8_lossy(&stdout_bytes).into_owned(),
stderr: String::from_utf8_lossy(&stderr_bytes).into_owned(),
truncated,
})
}
}
#[async_trait]
impl SandboxProvider for LocalEnvironmentProvider {
fn kind(&self) -> &str {
"local"
}
fn capabilities(&self) -> ProviderCapabilities {
capabilities([
"sandbox.lifecycle",
"sandbox.snapshot",
"sandbox.snapshot.restore",
])
}
fn validate_create(&self, request: &ProviderSandboxRequest) -> Result<()> {
if !request.mounts.is_empty() {
return Err(EnvironmentDomainError::new(
EnvironmentErrorCode::CapabilityUnsupported,
"local environment provider does not emulate Sandbox mounts",
)
.into());
}
Ok(())
}
async fn create_sandbox(
&self,
_: &ResourceScope,
request: &ProviderSandboxRequest,
_: &str,
) -> Result<ProviderResourceRef> {
let reference = self.reference("sandbox");
anyhow::ensure!(
request.mounts.is_empty(),
"local environment provider does not emulate Sandbox mounts"
);
let destination = self.resource("sandboxes", &reference.external_id)?;
if let Some(snapshot) = request.source_snapshot_external_id.as_deref() {
let source = self.resource("snapshots", snapshot)?;
tokio::task::spawn_blocking(move || copy_tree(&source, &destination)).await??;
} else {
tokio::fs::create_dir_all(destination).await?;
}
Ok(reference)
}
async fn stop_sandbox(&self, external_id: &str, _: &str) -> Result<()> {
remove_if_present(self.resource("sandboxes", external_id)?).await
}
}
#[async_trait]
impl SpaceProvider for LocalEnvironmentProvider {
fn kind(&self) -> &str {
"local"
}
fn capabilities(&self) -> ProviderCapabilities {
capabilities([
"space.lifecycle",
"space.files",
"space.mount",
"space.snapshot",
"space.restore",
"space.change_set",
])
}
async fn create_space(
&self,
_: &ResourceScope,
_request: &ProviderSpaceRequest,
_: &str,
) -> Result<ProviderResourceRef> {
Err(EnvironmentDomainError::new(
EnvironmentErrorCode::CapabilityUnsupported,
"local environment no longer exposes standalone Space volumes",
)
.into())
}
async fn read_file(&self, _: &str, _: &str) -> Result<ProviderFileContent> {
Err(EnvironmentDomainError::new(
EnvironmentErrorCode::CapabilityUnsupported,
"use Computer file APIs for local Spaces",
)
.into())
}
async fn write_file(
&self,
_: &str,
_: &str,
_: &[u8],
_: Option<&str>,
_: &str,
) -> Result<ProviderWriteResult> {
Err(EnvironmentDomainError::new(
EnvironmentErrorCode::CapabilityUnsupported,
"use Computer file APIs for local Spaces",
)
.into())
}
async fn create_change_set(&self, _: &str, _: &str, _: &str) -> Result<String> {
Err(EnvironmentDomainError::new(
EnvironmentErrorCode::CapabilityUnsupported,
"use Computer change set APIs for local Spaces",
)
.into())
}
async fn apply_change_set(
&self,
_: &str,
_: &str,
_: &str,
_: &str,
) -> Result<ProviderWriteResult> {
Err(EnvironmentDomainError::new(
EnvironmentErrorCode::CapabilityUnsupported,
"use Computer change set APIs for local Spaces",
)
.into())
}
async fn delete_space(&self, _: &str, _: &str) -> Result<()> {
Err(EnvironmentDomainError::new(
EnvironmentErrorCode::CapabilityUnsupported,
"use Computer directory APIs for local Spaces",
)
.into())
}
}
#[async_trait]
impl SnapshotProviderPort for LocalEnvironmentProvider {
fn kind(&self) -> &str {
"local"
}
async fn capture_snapshot(
&self,
source: &ProviderSnapshotSource,
_: &str,
) -> Result<ProviderResourceRef> {
let reference = self.reference("snapshot");
let source = match source {
ProviderSnapshotSource::Computer(external_id) => {
self.resource("computers", external_id)?
}
ProviderSnapshotSource::Sandbox(external_id) => {
self.resource("sandboxes", external_id)?
}
ProviderSnapshotSource::Space(path) => {
let (computer_external_id, space_directory) = path
.split_once('/')
.ok_or_else(|| anyhow::anyhow!("invalid local space snapshot source"))?;
self.space_root(computer_external_id, space_directory)?
}
};
let destination = self.resource("snapshots", &reference.external_id)?;
tokio::task::spawn_blocking(move || copy_tree(&source, &destination)).await??;
Ok(reference)
}
async fn delete_snapshot(&self, external_id: &str, _: &str) -> Result<()> {
remove_if_present(self.resource("snapshots", external_id)?).await
}
}
#[async_trait]
impl TemplateProviderPort for LocalEnvironmentProvider {
fn kind(&self) -> &str {
"local"
}
async fn inspect_template(&self, _: &str) -> Result<ProviderCapabilities> {
Ok(capabilities(["template.computer", "template.sandbox"]))
}
}
impl EnvironmentProviderRegistry for LocalEnvironmentProvider {
fn computer(&self, provider: &str) -> Result<Arc<dyn ComputerProvider>> {
ensure_local(provider)?;
Ok(Arc::new(self.clone()))
}
fn sandbox(&self, provider: &str) -> Result<Arc<dyn SandboxProvider>> {
ensure_local(provider)?;
Ok(Arc::new(self.clone()))
}
fn space(&self, provider: &str) -> Result<Arc<dyn SpaceProvider>> {
ensure_local(provider)?;
Ok(Arc::new(self.clone()))
}
fn snapshot(&self, provider: &str) -> Result<Arc<dyn SnapshotProviderPort>> {
ensure_local(provider)?;
Ok(Arc::new(self.clone()))
}
fn template(&self, provider: &str) -> Result<Arc<dyn TemplateProviderPort>> {
ensure_local(provider)?;
Ok(Arc::new(self.clone()))
}
}
fn capabilities<const N: usize>(values: [&str; N]) -> ProviderCapabilities {
ProviderCapabilities {
capabilities: values
.into_iter()
.map(str::to_string)
.collect::<BTreeSet<_>>(),
}
}
fn ensure_local(provider: &str) -> Result<()> {
anyhow::ensure!(
provider == "local",
"local adapter cannot serve provider {provider:?}"
);
Ok(())
}
async fn remove_if_present(path: PathBuf) -> Result<()> {
match tokio::fs::remove_dir_all(path).await {
Ok(()) => Ok(()),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(error.into()),
}
}
fn copy_tree(source: &Path, destination: &Path) -> Result<()> {
anyhow::ensure!(source.is_dir(), "local snapshot source does not exist");
std::fs::create_dir_all(destination)?;
for entry in std::fs::read_dir(source)? {
let entry = entry?;
let file_type = entry.file_type()?;
let target = destination.join(entry.file_name());
if file_type.is_dir() {
copy_tree(&entry.path(), &target)?;
} else if file_type.is_file() {
std::fs::copy(entry.path(), target).context("copying local snapshot file")?;
} else {
anyhow::bail!("local snapshots reject symlinks and special files");
}
}
Ok(())
}
#[cfg(test)]
mod computer_tests {
use super::*;
use agent_workspace_contract::{CommandNetworkPolicy, CommandSpec, SpacePersistence};
use std::collections::BTreeMap;
fn scope() -> ResourceScope {
ResourceScope {
tenant_id: "tenant".into(),
project_id: "project".into(),
}
}
fn space_request() -> ProviderSpaceRequest {
ProviderSpaceRequest {
persistence: SpacePersistence::Persistent,
source_snapshot_external_id: None,
}
}
fn command(argv: Vec<&str>, cwd: Option<&str>) -> CommandSpec {
CommandSpec {
argv: argv.into_iter().map(str::to_string).collect(),
cwd: cwd.map(str::to_string),
env: BTreeMap::new(),
shell: false,
timeout_ms: 5_000,
memory_bytes: 64 * 1024 * 1024,
cpu_millis: 1_000,
max_processes: 8,
network: CommandNetworkPolicy::Deny,
disk_bytes: 64 * 1024 * 1024,
max_output_bytes: 64 * 1024,
}
}
async fn computer() -> (
tempfile::TempDir,
Arc<dyn ComputerProvider>,
ProviderResourceRef,
) {
let root = tempfile::tempdir().unwrap();
let registry = LocalEnvironmentProvider::for_local_development(root.path()).unwrap();
let provider = registry.computer("local").unwrap();
let reference = provider
.create_computer(
&scope(),
&ProviderComputerRequest {
template_external_id: None,
profile: BTreeMap::new(),
},
"computer-key",
)
.await
.unwrap();
(root, provider, reference)
}
#[tokio::test]
async fn local_computer_keeps_space_files_and_survives_exec() {
let (_root, provider, computer) = computer().await;
let space = "spaces/spc_local";
provider
.create_space_directory(&computer.external_id, space, &space_request(), "space-key")
.await
.unwrap();
provider
.write_file(
&computer.external_id,
space,
"hello.txt",
b"from-local",
None,
"write-key",
)
.await
.unwrap();
let content = provider
.read_file(&computer.external_id, space, "hello.txt")
.await
.unwrap();
assert_eq!(content.bytes, b"from-local");
let output = provider
.exec_command(
&computer.external_id,
space,
&command(vec!["cat", "hello.txt"], None),
)
.await
.unwrap();
assert_eq!(output.exit_code, Some(0));
assert_eq!(output.stdout.trim(), "from-local");
let still_there = provider
.read_file(&computer.external_id, space, "hello.txt")
.await
.unwrap();
assert_eq!(
still_there.bytes, b"from-local",
"exec must not destroy the long-lived local Computer"
);
provider
.delete_computer(&computer.external_id, "delete-key")
.await
.unwrap();
}
#[tokio::test]
async fn local_computer_rejects_cwd_that_escapes_the_space() {
let (_root, provider, computer) = computer().await;
let space = "spaces/spc_local";
provider
.create_space_directory(&computer.external_id, space, &space_request(), "space-key")
.await
.unwrap();
let error = provider
.exec_command(
&computer.external_id,
space,
&command(vec!["pwd"], Some("../..")),
)
.await
.expect_err("cwd must stay inside the Space directory");
assert!(
error.to_string().contains("escapes"),
"unexpected error: {error}"
);
}
}