agent_workspace_adapter_local/
executor.rs

1//! Explicit local/test executor with durable, cross-process bulkheads.
2//!
3//! Production composition must inject a sandbox provider implementing
4//! `WorkspaceCommandExecutor`; this adapter never silently becomes the
5//! production default.
6
7use super::read_bounded;
8use agent_workspace_contract::{
9    CmdOutput, CommandPermitRepository, CommandPermitRequest, CommandSpec, SecretResolver,
10    WorkspaceCommandExecutor, WORKSPACE_MAX_COMMAND_OUTPUT_BYTES,
11};
12use anyhow::{Context, Result};
13use async_trait::async_trait;
14use std::fmt;
15use std::process::Stdio;
16use std::sync::Arc;
17use std::time::Duration;
18use tokio::io::AsyncRead;
19
20const MAX_COMMAND_TIMEOUT_MS: u64 = 60_000;
21
22#[derive(Clone)]
23pub struct LocalHostExecutor {
24    permits: Arc<dyn CommandPermitRepository>,
25    secrets: Arc<dyn SecretResolver>,
26    owner: String,
27    root: std::path::PathBuf,
28    global_limit: usize,
29    tenant_limit: usize,
30    allow_shell: bool,
31}
32
33#[derive(Debug)]
34pub struct EnvironmentSecretResolver;
35
36#[async_trait]
37impl SecretResolver for EnvironmentSecretResolver {
38    async fn resolve(
39        &self,
40        _tenant_id: &str,
41        secret: &agent_workspace_contract::SecretRef,
42    ) -> Result<Vec<u8>> {
43        anyhow::ensure!(
44            secret.provider == "env",
45            "local executor only accepts env SecretRef"
46        );
47        anyhow::ensure!(
48            valid_env_name(&secret.key),
49            "invalid secret environment key"
50        );
51        std::env::var_os(&secret.key)
52            .map(|value| value.to_string_lossy().as_bytes().to_vec())
53            .context("secret environment variable is unavailable")
54    }
55}
56
57impl fmt::Debug for LocalHostExecutor {
58    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
59        formatter
60            .debug_struct("LocalHostExecutor")
61            .field("owner", &self.owner)
62            .field("root", &self.root)
63            .field("global_limit", &self.global_limit)
64            .field("tenant_limit", &self.tenant_limit)
65            .field("allow_shell", &self.allow_shell)
66            .finish_non_exhaustive()
67    }
68}
69
70impl LocalHostExecutor {
71    /// This constructor is deliberately explicit and intended only for local
72    /// development and tests. Server production startup rejects this adapter.
73    pub fn for_local_development(
74        permits: Arc<dyn CommandPermitRepository>,
75        secrets: Arc<dyn SecretResolver>,
76        owner: impl Into<String>,
77        root: std::path::PathBuf,
78        global_limit: usize,
79        tenant_limit: usize,
80        allow_shell: bool,
81    ) -> Result<Self> {
82        anyhow::ensure!(global_limit > 0, "global command limit must be positive");
83        anyhow::ensure!(
84            tenant_limit > 0 && tenant_limit <= global_limit,
85            "tenant command limit must be in 1..=global limit"
86        );
87        Ok(Self {
88            permits,
89            secrets,
90            owner: owner.into(),
91            root,
92            global_limit,
93            tenant_limit,
94            allow_shell,
95        })
96    }
97}
98
99#[async_trait]
100impl WorkspaceCommandExecutor for LocalHostExecutor {
101    async fn execute(
102        &self,
103        tenant_id: &str,
104        _workspace_id: &str,
105        command_id: &str,
106        spec: &CommandSpec,
107    ) -> Result<CmdOutput> {
108        anyhow::ensure!(!tenant_id.is_empty(), "tenant id is required");
109        anyhow::ensure!(!spec.argv.is_empty(), "argv must not be empty");
110        anyhow::ensure!(spec.argv.len() <= 256, "argv item limit exceeded");
111        anyhow::ensure!(
112            spec.max_output_bytes <= WORKSPACE_MAX_COMMAND_OUTPUT_BYTES,
113            "command output limit exceeds policy"
114        );
115        anyhow::ensure!(
116            spec.timeout_ms > 0 && spec.timeout_ms <= MAX_COMMAND_TIMEOUT_MS,
117            "command timeout exceeds policy"
118        );
119        anyhow::ensure!(
120            !spec.shell || self.allow_shell,
121            "shell mode is not permitted"
122        );
123        let acquired = self
124            .permits
125            .acquire_command_permit(CommandPermitRequest {
126                command_id: command_id.into(),
127                tenant_id: tenant_id.into(),
128                owner: self.owner.clone(),
129                now_ms: now_ms(),
130                lease_ms: spec.timeout_ms.saturating_add(5_000),
131                global_limit: self.global_limit,
132                tenant_limit: self.tenant_limit,
133            })
134            .await?;
135        anyhow::ensure!(acquired, "workspace command quota exceeded");
136        let result = self.execute_acquired(tenant_id, command_id, spec).await;
137        let release = self
138            .permits
139            .release_command_permit(command_id, &self.owner)
140            .await;
141        match (result, release) {
142            (Ok(output), Ok(())) => Ok(output),
143            (Err(error), _) => Err(error),
144            (Ok(_), Err(error)) => Err(error),
145        }
146    }
147
148    async fn cancel(&self, tenant_id: &str, command_id: &str) -> Result<()> {
149        anyhow::ensure!(!tenant_id.is_empty(), "tenant id is required");
150        anyhow::ensure!(
151            self.permits.cancel_command_permit(command_id).await?,
152            "command not found"
153        );
154        Ok(())
155    }
156}
157
158impl LocalHostExecutor {
159    async fn execute_acquired(
160        &self,
161        tenant_id: &str,
162        command_id: &str,
163        spec: &CommandSpec,
164    ) -> Result<CmdOutput> {
165        let cwd = spec.cwd.as_deref().unwrap_or(".");
166        let cwd = crate::path_resolver::WorkspacePathResolver::new(&self.root)
167            .existing(cwd)
168            .await?;
169        let mut command = if spec.shell {
170            let mut value = tokio::process::Command::new("sh");
171            value.arg("-lc").arg(spec.argv.join(" "));
172            value
173        } else {
174            let mut value = tokio::process::Command::new(&spec.argv[0]);
175            value.args(&spec.argv[1..]);
176            value
177        };
178        command
179            .current_dir(cwd)
180            .stdin(Stdio::null())
181            .stdout(Stdio::piped())
182            .stderr(Stdio::piped())
183            .kill_on_drop(true);
184        for (name, reference) in &spec.env {
185            anyhow::ensure!(valid_env_name(name), "invalid environment variable name");
186            let value = self.secrets.resolve(tenant_id, reference).await?;
187            #[cfg(unix)]
188            {
189                use std::os::unix::ffi::OsStringExt;
190                command.env(name, std::ffi::OsString::from_vec(value));
191            }
192            #[cfg(not(unix))]
193            command.env(
194                name,
195                String::from_utf8(value).context("secret environment value is not UTF-8")?,
196            );
197        }
198        #[cfg(unix)]
199        configure_isolation(&mut command, spec)?;
200        let mut child = command.spawn().context("spawn local development command")?;
201        let process_id = child.id().context("spawned command has no process id")?;
202        let stdout = child.stdout.take().context("stdout not captured")?;
203        let stderr = child.stderr.take().context("stderr not captured")?;
204        let stdout_task = tokio::spawn(read_output(stdout, spec.max_output_bytes));
205        let stderr_task = tokio::spawn(read_output(stderr, spec.max_output_bytes));
206        let timeout = tokio::time::sleep(Duration::from_millis(spec.timeout_ms));
207        tokio::pin!(timeout);
208        let status = loop {
209            tokio::select! {
210                result=child.wait()=>break result?,
211                _=&mut timeout=>{
212                    kill_process_tree(process_id); let _=child.kill().await; let _=child.wait().await;
213                    anyhow::bail!("workspace command timed out");
214                }
215                _=tokio::time::sleep(Duration::from_millis(50))=>{
216                    if self.permits.command_canceled(command_id).await?{
217                        kill_process_tree(process_id); let _=child.kill().await; let _=child.wait().await;
218                        anyhow::bail!("workspace command canceled");
219                    }
220                }
221            }
222        };
223        let (stdout, stdout_truncated) = stdout_task.await.context("join stdout")??;
224        let (stderr, stderr_truncated) = stderr_task.await.context("join stderr")??;
225        Ok(CmdOutput {
226            exit_code: status.code().unwrap_or(-1),
227            stdout: String::from_utf8_lossy(&stdout).into_owned(),
228            stderr: String::from_utf8_lossy(&stderr).into_owned(),
229            stdout_truncated,
230            stderr_truncated,
231        })
232    }
233}
234
235async fn read_output<R: AsyncRead + Unpin>(
236    reader: R,
237    limit: usize,
238) -> std::io::Result<(Vec<u8>, bool)> {
239    read_bounded(reader, limit).await
240}
241fn valid_env_name(value: &str) -> bool {
242    !value.is_empty()
243        && value
244            .bytes()
245            .all(|value| value == b'_' || value.is_ascii_alphanumeric())
246        && !value.bytes().next().unwrap().is_ascii_digit()
247}
248fn now_ms() -> u64 {
249    std::time::SystemTime::now()
250        .duration_since(std::time::UNIX_EPOCH)
251        .unwrap_or_default()
252        .as_millis()
253        .try_into()
254        .unwrap_or(u64::MAX)
255}
256
257#[cfg(unix)]
258fn configure_isolation(command: &mut tokio::process::Command, spec: &CommandSpec) -> Result<()> {
259    use std::os::unix::process::CommandExt;
260    let memory = spec.memory_bytes;
261    let cpu_secs = spec.cpu_millis.div_ceil(1000).max(1);
262    let processes = u64::from(spec.max_processes.max(1));
263    command.as_std_mut().process_group(0);
264    unsafe {
265        command.as_std_mut().pre_exec(move || {
266            #[cfg(not(target_os = "macos"))]
267            {
268                set_limit(libc::RLIMIT_AS, memory)?;
269                set_limit(libc::RLIMIT_NPROC, processes)?;
270            }
271            #[cfg(target_os = "macos")]
272            let _ = (memory, processes);
273            set_limit(libc::RLIMIT_CPU, cpu_secs)?;
274            Ok(())
275        });
276    }
277    Ok(())
278}
279
280#[cfg(unix)]
281fn set_limit(resource: RlimitResource, value: u64) -> std::io::Result<()> {
282    let mut current = libc::rlimit {
283        rlim_cur: 0,
284        rlim_max: 0,
285    };
286    if unsafe { libc::getrlimit(resource, &mut current) } != 0 {
287        return Err(std::io::Error::last_os_error());
288    }
289    let limit = libc::rlimit {
290        rlim_cur: (value as libc::rlim_t).min(current.rlim_max),
291        rlim_max: current.rlim_max,
292    };
293    if unsafe { libc::setrlimit(resource, &limit) } == 0 {
294        Ok(())
295    } else {
296        Err(std::io::Error::last_os_error())
297    }
298}
299
300#[cfg(all(unix, any(target_os = "linux", target_os = "android")))]
301type RlimitResource = libc::c_uint;
302
303#[cfg(all(unix, not(any(target_os = "linux", target_os = "android"))))]
304type RlimitResource = libc::c_int;
305
306#[cfg(unix)]
307fn kill_process_tree(pid: u32) {
308    unsafe {
309        libc::kill(-(pid as i32), libc::SIGKILL);
310    }
311}
312#[cfg(not(unix))]
313fn kill_process_tree(_pid: u32) {}
agent_workspace_adapter_local/executor.rs

agent_workspace_adapter_local/
executor.rs
