Skip to main content

agent_workspace_adapter_local/
local_environment.rs

1use agent_workspace_contract::*;
2use anyhow::{Context, Result};
3use async_trait::async_trait;
4use std::collections::BTreeSet;
5use std::path::{Component, Path, PathBuf};
6use std::process::Stdio;
7use std::sync::Arc;
8use tokio::io::AsyncReadExt;
9
10/// Development-only environment provider. Production composition never
11/// constructs this adapter.
12#[derive(Debug, Clone)]
13pub struct LocalEnvironmentProvider {
14    root: Arc<PathBuf>,
15}
16
17impl LocalEnvironmentProvider {
18    pub fn for_local_development(root: impl Into<PathBuf>) -> Result<Self> {
19        let root = root.into();
20        std::fs::create_dir_all(&root)?;
21        let root = std::fs::canonicalize(root)?;
22        for name in ["computers", "sandboxes", "snapshots"] {
23            std::fs::create_dir_all(root.join(name))?;
24        }
25        Ok(Self {
26            root: Arc::new(root),
27        })
28    }
29
30    fn resource(&self, kind: &str, id: &str) -> Result<PathBuf> {
31        anyhow::ensure!(
32            !id.is_empty()
33                && id
34                    .bytes()
35                    .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')),
36            "invalid local environment resource id"
37        );
38        Ok(self.root.join(kind).join(id))
39    }
40
41    fn reference(&self, kind: &str) -> ProviderResourceRef {
42        ProviderResourceRef {
43            provider: "local".into(),
44            external_id: format!("{kind}-{}", uuid::Uuid::new_v4().simple()),
45        }
46    }
47
48    fn computer_root(&self, external_id: &str) -> Result<PathBuf> {
49        self.resource("computers", external_id)
50    }
51
52    fn space_root(&self, computer_external_id: &str, space_directory: &str) -> Result<PathBuf> {
53        let relative = Path::new(space_directory);
54        anyhow::ensure!(
55            !relative.is_absolute()
56                && relative
57                    .components()
58                    .all(|component| matches!(component, Component::Normal(_))),
59            "space directory escapes the computer root"
60        );
61        Ok(self.computer_root(computer_external_id)?.join(relative))
62    }
63
64    fn file_in_space(
65        &self,
66        computer_external_id: &str,
67        space_directory: &str,
68        path: &str,
69    ) -> Result<PathBuf> {
70        let relative = Path::new(path);
71        anyhow::ensure!(
72            !relative.is_absolute()
73                && relative
74                    .components()
75                    .all(|component| matches!(component, Component::Normal(_))),
76            "local Space path escapes its root"
77        );
78        Ok(self
79            .space_root(computer_external_id, space_directory)?
80            .join(relative))
81    }
82}
83
84#[async_trait]
85impl ComputerProvider for LocalEnvironmentProvider {
86    fn kind(&self) -> &str {
87        "local"
88    }
89    fn capabilities(&self) -> ProviderCapabilities {
90        capabilities([
91            "computer.lifecycle",
92            "computer.snapshot",
93            "computer.files",
94            "computer.exec",
95            "space.change_set",
96            "space.restore",
97            "space.snapshot",
98        ])
99    }
100    async fn create_computer(
101        &self,
102        _: &ResourceScope,
103        _: &ProviderComputerRequest,
104        _: &str,
105    ) -> Result<ProviderResourceRef> {
106        let reference = self.reference("computer");
107        tokio::fs::create_dir_all(self.resource("computers", &reference.external_id)?).await?;
108        Ok(reference)
109    }
110    async fn delete_computer(&self, external_id: &str, _: &str) -> Result<()> {
111        remove_if_present(self.resource("computers", external_id)?).await
112    }
113    async fn create_space_directory(
114        &self,
115        computer_external_id: &str,
116        space_directory: &str,
117        request: &ProviderSpaceRequest,
118        _: &str,
119    ) -> Result<()> {
120        let destination = self.space_root(computer_external_id, space_directory)?;
121        if let Some(snapshot) = request.source_snapshot_external_id.as_deref() {
122            let source = self.resource("snapshots", snapshot)?;
123            tokio::task::spawn_blocking(move || copy_tree(&source, &destination)).await??;
124        } else {
125            tokio::fs::create_dir_all(destination).await?;
126        }
127        Ok(())
128    }
129    async fn delete_space_directory(
130        &self,
131        computer_external_id: &str,
132        space_directory: &str,
133        _: &str,
134    ) -> Result<()> {
135        remove_if_present(self.space_root(computer_external_id, space_directory)?).await
136    }
137    async fn read_file(
138        &self,
139        computer_external_id: &str,
140        space_directory: &str,
141        path: &str,
142    ) -> Result<ProviderFileContent> {
143        Ok(ProviderFileContent {
144            bytes: tokio::fs::read(self.file_in_space(
145                computer_external_id,
146                space_directory,
147                path,
148            )?)
149            .await?,
150            provider_revision: None,
151        })
152    }
153    async fn write_file(
154        &self,
155        computer_external_id: &str,
156        space_directory: &str,
157        path: &str,
158        content: &[u8],
159        _: Option<&str>,
160        _: &str,
161    ) -> Result<ProviderWriteResult> {
162        let file = self.file_in_space(computer_external_id, space_directory, path)?;
163        if let Some(parent) = file.parent() {
164            tokio::fs::create_dir_all(parent).await?;
165        }
166        tokio::fs::write(file, content).await?;
167        Ok(ProviderWriteResult {
168            provider_revision: None,
169        })
170    }
171    async fn create_change_set(
172        &self,
173        computer_external_id: &str,
174        source_space_directory: &str,
175        _: &str,
176        _: &str,
177    ) -> Result<String> {
178        Ok(format!("{computer_external_id}:{source_space_directory}"))
179    }
180    async fn apply_change_set(
181        &self,
182        computer_external_id: &str,
183        source_space_directory: &str,
184        target_space_directory: &str,
185        provider_change_ref: &str,
186        _: &str,
187    ) -> Result<ProviderWriteResult> {
188        anyhow::ensure!(
189            provider_change_ref == format!("{computer_external_id}:{source_space_directory}"),
190            "local ChangeSet source reference mismatch"
191        );
192        let source = self.space_root(computer_external_id, source_space_directory)?;
193        let target = self.space_root(computer_external_id, target_space_directory)?;
194        tokio::task::spawn_blocking(move || {
195            if target.exists() {
196                std::fs::remove_dir_all(&target)?;
197            }
198            copy_tree(&source, &target)
199        })
200        .await??;
201        Ok(ProviderWriteResult {
202            provider_revision: None,
203        })
204    }
205    async fn exec_command(
206        &self,
207        computer_external_id: &str,
208        space_directory: &str,
209        command: &CommandSpec,
210    ) -> Result<ProviderCommandOutput> {
211        anyhow::ensure!(!command.argv.is_empty(), "argv must not be empty");
212        let cwd_path = match command.cwd.as_deref() {
213            None | Some("") | Some(".") => {
214                self.space_root(computer_external_id, space_directory)?
215            }
216            Some(cwd) => self.file_in_space(computer_external_id, space_directory, cwd)?,
217        };
218        let mut process = if command.shell {
219            let mut value = tokio::process::Command::new("sh");
220            value.arg("-lc").arg(command.argv.join(" "));
221            value
222        } else {
223            let mut value = tokio::process::Command::new(&command.argv[0]);
224            value.args(&command.argv[1..]);
225            value
226        };
227        process
228            .current_dir(cwd_path)
229            .stdin(Stdio::null())
230            .stdout(Stdio::piped())
231            .stderr(Stdio::piped())
232            .kill_on_drop(true);
233        let mut child = process.spawn()?;
234        let stdout = child.stdout.take();
235        let stderr = child.stderr.take();
236        let status = tokio::time::timeout(
237            std::time::Duration::from_millis(command.timeout_ms),
238            child.wait(),
239        )
240        .await
241        .map_err(|_| anyhow::anyhow!("command timed out"))??;
242        let mut stdout_bytes = Vec::new();
243        let mut stderr_bytes = Vec::new();
244        if let Some(mut stream) = stdout {
245            stream.read_to_end(&mut stdout_bytes).await?;
246        }
247        if let Some(mut stream) = stderr {
248            stream.read_to_end(&mut stderr_bytes).await?;
249        }
250        let truncated = stdout_bytes.len() > command.max_output_bytes
251            || stderr_bytes.len() > command.max_output_bytes;
252        if stdout_bytes.len() > command.max_output_bytes {
253            stdout_bytes.truncate(command.max_output_bytes);
254        }
255        if stderr_bytes.len() > command.max_output_bytes {
256            stderr_bytes.truncate(command.max_output_bytes);
257        }
258        Ok(ProviderCommandOutput {
259            exit_code: status.code(),
260            stdout: String::from_utf8_lossy(&stdout_bytes).into_owned(),
261            stderr: String::from_utf8_lossy(&stderr_bytes).into_owned(),
262            truncated,
263        })
264    }
265}
266
267#[async_trait]
268impl SandboxProvider for LocalEnvironmentProvider {
269    fn kind(&self) -> &str {
270        "local"
271    }
272    fn capabilities(&self) -> ProviderCapabilities {
273        capabilities([
274            "sandbox.lifecycle",
275            "sandbox.snapshot",
276            "sandbox.snapshot.restore",
277        ])
278    }
279    fn validate_create(&self, request: &ProviderSandboxRequest) -> Result<()> {
280        if !request.mounts.is_empty() {
281            return Err(EnvironmentDomainError::new(
282                EnvironmentErrorCode::CapabilityUnsupported,
283                "local environment provider does not emulate Sandbox mounts",
284            )
285            .into());
286        }
287        Ok(())
288    }
289    async fn create_sandbox(
290        &self,
291        _: &ResourceScope,
292        request: &ProviderSandboxRequest,
293        _: &str,
294    ) -> Result<ProviderResourceRef> {
295        let reference = self.reference("sandbox");
296        anyhow::ensure!(
297            request.mounts.is_empty(),
298            "local environment provider does not emulate Sandbox mounts"
299        );
300        let destination = self.resource("sandboxes", &reference.external_id)?;
301        if let Some(snapshot) = request.source_snapshot_external_id.as_deref() {
302            let source = self.resource("snapshots", snapshot)?;
303            tokio::task::spawn_blocking(move || copy_tree(&source, &destination)).await??;
304        } else {
305            tokio::fs::create_dir_all(destination).await?;
306        }
307        Ok(reference)
308    }
309    async fn stop_sandbox(&self, external_id: &str, _: &str) -> Result<()> {
310        remove_if_present(self.resource("sandboxes", external_id)?).await
311    }
312}
313
314#[async_trait]
315impl SpaceProvider for LocalEnvironmentProvider {
316    fn kind(&self) -> &str {
317        "local"
318    }
319    fn capabilities(&self) -> ProviderCapabilities {
320        capabilities([
321            "space.lifecycle",
322            "space.files",
323            "space.mount",
324            "space.snapshot",
325            "space.restore",
326            "space.change_set",
327        ])
328    }
329    async fn create_space(
330        &self,
331        _: &ResourceScope,
332        _request: &ProviderSpaceRequest,
333        _: &str,
334    ) -> Result<ProviderResourceRef> {
335        Err(EnvironmentDomainError::new(
336            EnvironmentErrorCode::CapabilityUnsupported,
337            "local environment no longer exposes standalone Space volumes",
338        )
339        .into())
340    }
341    async fn read_file(&self, _: &str, _: &str) -> Result<ProviderFileContent> {
342        Err(EnvironmentDomainError::new(
343            EnvironmentErrorCode::CapabilityUnsupported,
344            "use Computer file APIs for local Spaces",
345        )
346        .into())
347    }
348    async fn write_file(
349        &self,
350        _: &str,
351        _: &str,
352        _: &[u8],
353        _: Option<&str>,
354        _: &str,
355    ) -> Result<ProviderWriteResult> {
356        Err(EnvironmentDomainError::new(
357            EnvironmentErrorCode::CapabilityUnsupported,
358            "use Computer file APIs for local Spaces",
359        )
360        .into())
361    }
362    async fn create_change_set(&self, _: &str, _: &str, _: &str) -> Result<String> {
363        Err(EnvironmentDomainError::new(
364            EnvironmentErrorCode::CapabilityUnsupported,
365            "use Computer change set APIs for local Spaces",
366        )
367        .into())
368    }
369    async fn apply_change_set(
370        &self,
371        _: &str,
372        _: &str,
373        _: &str,
374        _: &str,
375    ) -> Result<ProviderWriteResult> {
376        Err(EnvironmentDomainError::new(
377            EnvironmentErrorCode::CapabilityUnsupported,
378            "use Computer change set APIs for local Spaces",
379        )
380        .into())
381    }
382    async fn delete_space(&self, _: &str, _: &str) -> Result<()> {
383        Err(EnvironmentDomainError::new(
384            EnvironmentErrorCode::CapabilityUnsupported,
385            "use Computer directory APIs for local Spaces",
386        )
387        .into())
388    }
389}
390
391#[async_trait]
392impl SnapshotProviderPort for LocalEnvironmentProvider {
393    fn kind(&self) -> &str {
394        "local"
395    }
396    async fn capture_snapshot(
397        &self,
398        source: &ProviderSnapshotSource,
399        _: &str,
400    ) -> Result<ProviderResourceRef> {
401        let reference = self.reference("snapshot");
402        let source = match source {
403            ProviderSnapshotSource::Computer(external_id) => {
404                self.resource("computers", external_id)?
405            }
406            ProviderSnapshotSource::Sandbox(external_id) => {
407                self.resource("sandboxes", external_id)?
408            }
409            ProviderSnapshotSource::Space(path) => {
410                let (computer_external_id, space_directory) = path
411                    .split_once('/')
412                    .ok_or_else(|| anyhow::anyhow!("invalid local space snapshot source"))?;
413                self.space_root(computer_external_id, space_directory)?
414            }
415        };
416        let destination = self.resource("snapshots", &reference.external_id)?;
417        tokio::task::spawn_blocking(move || copy_tree(&source, &destination)).await??;
418        Ok(reference)
419    }
420    async fn delete_snapshot(&self, external_id: &str, _: &str) -> Result<()> {
421        remove_if_present(self.resource("snapshots", external_id)?).await
422    }
423}
424
425#[async_trait]
426impl TemplateProviderPort for LocalEnvironmentProvider {
427    fn kind(&self) -> &str {
428        "local"
429    }
430    async fn inspect_template(&self, _: &str) -> Result<ProviderCapabilities> {
431        Ok(capabilities(["template.computer", "template.sandbox"]))
432    }
433}
434
435impl EnvironmentProviderRegistry for LocalEnvironmentProvider {
436    fn computer(&self, provider: &str) -> Result<Arc<dyn ComputerProvider>> {
437        ensure_local(provider)?;
438        Ok(Arc::new(self.clone()))
439    }
440    fn sandbox(&self, provider: &str) -> Result<Arc<dyn SandboxProvider>> {
441        ensure_local(provider)?;
442        Ok(Arc::new(self.clone()))
443    }
444    fn space(&self, provider: &str) -> Result<Arc<dyn SpaceProvider>> {
445        ensure_local(provider)?;
446        Ok(Arc::new(self.clone()))
447    }
448    fn snapshot(&self, provider: &str) -> Result<Arc<dyn SnapshotProviderPort>> {
449        ensure_local(provider)?;
450        Ok(Arc::new(self.clone()))
451    }
452    fn template(&self, provider: &str) -> Result<Arc<dyn TemplateProviderPort>> {
453        ensure_local(provider)?;
454        Ok(Arc::new(self.clone()))
455    }
456}
457
458fn capabilities<const N: usize>(values: [&str; N]) -> ProviderCapabilities {
459    ProviderCapabilities {
460        capabilities: values
461            .into_iter()
462            .map(str::to_string)
463            .collect::<BTreeSet<_>>(),
464    }
465}
466
467fn ensure_local(provider: &str) -> Result<()> {
468    anyhow::ensure!(
469        provider == "local",
470        "local adapter cannot serve provider {provider:?}"
471    );
472    Ok(())
473}
474
475async fn remove_if_present(path: PathBuf) -> Result<()> {
476    match tokio::fs::remove_dir_all(path).await {
477        Ok(()) => Ok(()),
478        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
479        Err(error) => Err(error.into()),
480    }
481}
482
483fn copy_tree(source: &Path, destination: &Path) -> Result<()> {
484    anyhow::ensure!(source.is_dir(), "local snapshot source does not exist");
485    std::fs::create_dir_all(destination)?;
486    for entry in std::fs::read_dir(source)? {
487        let entry = entry?;
488        let file_type = entry.file_type()?;
489        let target = destination.join(entry.file_name());
490        if file_type.is_dir() {
491            copy_tree(&entry.path(), &target)?;
492        } else if file_type.is_file() {
493            std::fs::copy(entry.path(), target).context("copying local snapshot file")?;
494        } else {
495            anyhow::bail!("local snapshots reject symlinks and special files");
496        }
497    }
498    Ok(())
499}
500
501#[cfg(test)]
502mod computer_tests {
503    use super::*;
504    use agent_workspace_contract::{CommandNetworkPolicy, CommandSpec, SpacePersistence};
505    use std::collections::BTreeMap;
506
507    fn scope() -> ResourceScope {
508        ResourceScope {
509            tenant_id: "tenant".into(),
510            project_id: "project".into(),
511        }
512    }
513
514    fn space_request() -> ProviderSpaceRequest {
515        ProviderSpaceRequest {
516            persistence: SpacePersistence::Persistent,
517            source_snapshot_external_id: None,
518        }
519    }
520
521    fn command(argv: Vec<&str>, cwd: Option<&str>) -> CommandSpec {
522        CommandSpec {
523            argv: argv.into_iter().map(str::to_string).collect(),
524            cwd: cwd.map(str::to_string),
525            env: BTreeMap::new(),
526            shell: false,
527            timeout_ms: 5_000,
528            memory_bytes: 64 * 1024 * 1024,
529            cpu_millis: 1_000,
530            max_processes: 8,
531            network: CommandNetworkPolicy::Deny,
532            disk_bytes: 64 * 1024 * 1024,
533            max_output_bytes: 64 * 1024,
534        }
535    }
536
537    async fn computer() -> (
538        tempfile::TempDir,
539        Arc<dyn ComputerProvider>,
540        ProviderResourceRef,
541    ) {
542        let root = tempfile::tempdir().unwrap();
543        let registry = LocalEnvironmentProvider::for_local_development(root.path()).unwrap();
544        let provider = registry.computer("local").unwrap();
545        let reference = provider
546            .create_computer(
547                &scope(),
548                &ProviderComputerRequest {
549                    template_external_id: None,
550                    profile: BTreeMap::new(),
551                },
552                "computer-key",
553            )
554            .await
555            .unwrap();
556        (root, provider, reference)
557    }
558
559    #[tokio::test]
560    async fn local_computer_keeps_space_files_and_survives_exec() {
561        let (_root, provider, computer) = computer().await;
562        let space = "spaces/spc_local";
563        provider
564            .create_space_directory(&computer.external_id, space, &space_request(), "space-key")
565            .await
566            .unwrap();
567        provider
568            .write_file(
569                &computer.external_id,
570                space,
571                "hello.txt",
572                b"from-local",
573                None,
574                "write-key",
575            )
576            .await
577            .unwrap();
578        let content = provider
579            .read_file(&computer.external_id, space, "hello.txt")
580            .await
581            .unwrap();
582        assert_eq!(content.bytes, b"from-local");
583
584        let output = provider
585            .exec_command(
586                &computer.external_id,
587                space,
588                &command(vec!["cat", "hello.txt"], None),
589            )
590            .await
591            .unwrap();
592        assert_eq!(output.exit_code, Some(0));
593        assert_eq!(output.stdout.trim(), "from-local");
594
595        let still_there = provider
596            .read_file(&computer.external_id, space, "hello.txt")
597            .await
598            .unwrap();
599        assert_eq!(
600            still_there.bytes, b"from-local",
601            "exec must not destroy the long-lived local Computer"
602        );
603        provider
604            .delete_computer(&computer.external_id, "delete-key")
605            .await
606            .unwrap();
607    }
608
609    #[tokio::test]
610    async fn local_computer_rejects_cwd_that_escapes_the_space() {
611        let (_root, provider, computer) = computer().await;
612        let space = "spaces/spc_local";
613        provider
614            .create_space_directory(&computer.external_id, space, &space_request(), "space-key")
615            .await
616            .unwrap();
617        let error = provider
618            .exec_command(
619                &computer.external_id,
620                space,
621                &command(vec!["pwd"], Some("../..")),
622            )
623            .await
624            .expect_err("cwd must stay inside the Space directory");
625        assert!(
626            error.to_string().contains("escapes"),
627            "unexpected error: {error}"
628        );
629    }
630}