soma_provider_adapters/
sidecar.rs

1//! Bounded child-process sidecar execution shared by the ai-sdk and python
2//! adapters (and any other adapter that shells out to a runtime process for
3//! one bounded, stdin-in/stdout-out call). Ported from
4//! `soma-service::providers::sidecar` with the env-var prefix generalized to
5//! a caller-supplied parameter — see the crate-level docs on why generic
6//! shared crates must not hard-code a product's env prefix.
7
8use std::{
9    ffi::OsString,
10    io,
11    path::{Path, PathBuf},
12    process::{Output, Stdio},
13    time::Duration,
14};
15
16use serde::Serialize;
17use soma_provider_core::{EnvRequirement, ProviderCall, ProviderError, ProviderSurface};
18use tokio::{
19    io::{AsyncRead, AsyncReadExt, AsyncWriteExt},
20    process::Command,
21    time::timeout,
22};
23
24use crate::error::SidecarError;
25
26/// The stdin wire envelope sent to every sidecar-executing adapter (ai-sdk,
27/// python). Field names and shape are load-bearing: they match the
28/// pre-extraction `ProviderExecutionEnvelope` byte-for-byte so drop-in
29/// TypeScript/Python provider handlers written against the documented input
30/// shape keep working unchanged.
31#[derive(Debug, Serialize)]
32pub struct ExecutionEnvelope<'a> {
33    pub schema_version: u32,
34    pub provider: &'a str,
35    pub action: &'a str,
36    pub params: &'a serde_json::Value,
37    pub surface: ProviderSurface,
38    pub snapshot_id: &'a str,
39}
40
41impl<'a> ExecutionEnvelope<'a> {
42    pub fn new(call: &'a ProviderCall) -> Self {
43        Self {
44            schema_version: 1,
45            provider: &call.provider,
46            action: &call.action,
47            params: &call.params,
48            surface: call.surface,
49            snapshot_id: &call.snapshot_id,
50        }
51    }
52}
53
54/// Serializes `call` into the sidecar stdin wire envelope.
55pub fn execution_payload(call: &ProviderCall) -> Result<Vec<u8>, serde_json::Error> {
56    serde_json::to_vec(&ExecutionEnvelope::new(call))
57}
58
59/// Attempts before giving up on a still-busy interpreter image, and the pause
60/// between them. Ten 5ms attempts bound the wait at ~50ms, which is far below
61/// any sidecar or worker startup timeout while comfortably outlasting the
62/// close-to-exec window in practice.
63const EXEC_BUSY_ATTEMPTS: u32 = 10;
64const EXEC_BUSY_BACKOFF: Duration = Duration::from_millis(5);
65
66/// `true` when the kernel refused to exec an image because it is still open
67/// for writing somewhere.
68///
69/// A prepared interpreter is materialized (copied or hard-linked into a
70/// generation tree) and then executed almost immediately. On Linux `execve`
71/// fails with `ETXTBSY` while *any* thread in this process still holds a write
72/// descriptor to that image — including a descriptor the writing thread has
73/// already dropped but whose close has not yet been observed. The condition is
74/// transient by construction and clears on its own, so it must be retried
75/// rather than reported as a broken provider.
76fn image_is_busy(error: &io::Error) -> bool {
77    error.kind() == io::ErrorKind::ExecutableFileBusy
78}
79
80/// Spawns `command`, retrying only the transient "image still open for
81/// writing" condition detected by `image_is_busy`. Every other spawn error
82/// is returned immediately and untouched.
83pub async fn spawn_retrying_busy_image(command: &mut Command) -> io::Result<tokio::process::Child> {
84    for _ in 0..EXEC_BUSY_ATTEMPTS {
85        match command.spawn() {
86            Err(error) if image_is_busy(&error) => {
87                tokio::time::sleep(EXEC_BUSY_BACKOFF).await;
88            }
89            result => return result,
90        }
91    }
92    command.spawn()
93}
94
95/// Blocking counterpart to [`spawn_retrying_busy_image`] for callers that
96/// drive a `std::process::Command` (the catalog sidecar runs on a blocking
97/// worker and has no reactor to await on).
98pub fn spawn_retrying_busy_image_blocking(
99    command: &mut std::process::Command,
100) -> io::Result<std::process::Child> {
101    for _ in 0..EXEC_BUSY_ATTEMPTS {
102        match command.spawn() {
103            Err(error) if image_is_busy(&error) => {
104                std::thread::sleep(EXEC_BUSY_BACKOFF);
105            }
106            result => return result,
107        }
108    }
109    command.spawn()
110}
111
112pub struct BoundedOutput {
113    pub output: Output,
114    pub stdout_exceeded: bool,
115    pub stderr_exceeded: bool,
116}
117
118pub async fn run_bounded_sidecar(
119    command: &str,
120    args: &[&str],
121    env: Vec<(String, String)>,
122    input: &[u8],
123    timeout_ms: u64,
124    max_output_bytes: usize,
125) -> Result<BoundedOutput, SidecarError> {
126    let resolved_command = resolve_sidecar_command(command);
127    let mut command = Command::new(resolved_command);
128    command
129        .args(args)
130        .kill_on_drop(true)
131        .env_clear()
132        .stdin(Stdio::piped())
133        .stdout(Stdio::piped())
134        .stderr(Stdio::piped());
135    apply_sidecar_base_env(&mut command);
136    command.envs(env);
137
138    let mut child = spawn_retrying_busy_image(&mut command)
139        .await
140        .map_err(SidecarError::Io)?;
141
142    let stdout = child
143        .stdout
144        .take()
145        .ok_or_else(|| io::Error::other("sidecar stdout pipe was not captured"))
146        .map_err(SidecarError::Io)?;
147    let stderr = child
148        .stderr
149        .take()
150        .ok_or_else(|| io::Error::other("sidecar stderr pipe was not captured"))
151        .map_err(SidecarError::Io)?;
152    let stdout_task = tokio::spawn(read_bounded(stdout, max_output_bytes));
153    let stderr_task = tokio::spawn(read_bounded(stderr, max_output_bytes));
154
155    if let Some(mut stdin) = child.stdin.take() {
156        stdin.write_all(input).await.map_err(SidecarError::Io)?;
157    }
158
159    let status = match timeout(Duration::from_millis(timeout_ms), child.wait()).await {
160        Ok(status) => status.map_err(SidecarError::Io)?,
161        Err(_) => {
162            let _ = child.kill().await;
163            let _ = child.wait().await;
164            stdout_task.abort();
165            stderr_task.abort();
166            return Err(SidecarError::Timeout);
167        }
168    };
169
170    let (stdout, stdout_exceeded) = stdout_task
171        .await
172        .map_err(SidecarError::Join)?
173        .map_err(SidecarError::Io)?;
174    let (stderr, stderr_exceeded) = stderr_task
175        .await
176        .map_err(SidecarError::Join)?
177        .map_err(SidecarError::Io)?;
178
179    Ok(BoundedOutput {
180        output: Output {
181            status,
182            stdout,
183            stderr,
184        },
185        stdout_exceeded,
186        stderr_exceeded,
187    })
188}
189
190fn apply_sidecar_base_env(command: &mut Command) {
191    for (key, value) in sidecar_base_env() {
192        command.env(key, value);
193    }
194}
195
196#[cfg(windows)]
197pub fn sidecar_base_env() -> Vec<(OsString, OsString)> {
198    let mut env = Vec::new();
199    for key in ["SystemRoot", "WINDIR", "COMSPEC", "PATHEXT", "TEMP", "TMP"] {
200        if let Some(value) = std::env::var_os(key) {
201            env.push((OsString::from(key), value));
202        }
203    }
204    env
205}
206
207#[cfg(not(windows))]
208pub fn sidecar_base_env() -> Vec<(OsString, OsString)> {
209    let mut env = Vec::new();
210    for key in ["HOME", "TMPDIR", "TEMP", "TMP"] {
211        if let Some(value) = std::env::var_os(key) {
212            env.push((OsString::from(key), value));
213        }
214    }
215    env
216}
217
218pub fn resolve_sidecar_command(command: &str) -> PathBuf {
219    resolve_sidecar_command_with_env(
220        command,
221        std::env::var_os("PATH"),
222        std::env::var_os("PATHEXT"),
223    )
224}
225
226fn resolve_sidecar_command_with_env(
227    command: &str,
228    path_env: Option<OsString>,
229    pathext_env: Option<OsString>,
230) -> PathBuf {
231    let command_path = Path::new(command);
232    if command_path.components().count() > 1 || command_path.is_absolute() {
233        return command_path.to_path_buf();
234    }
235
236    let Some(path_env) = path_env else {
237        return command_path.to_path_buf();
238    };
239    for dir in std::env::split_paths(&path_env) {
240        if command_path.extension().is_some() {
241            let candidate = dir.join(command_path);
242            if candidate.is_file() {
243                return resolve_runtime_shim(command, candidate);
244            }
245            continue;
246        }
247        let direct_candidate = dir.join(command_path);
248        if direct_candidate.is_file() {
249            return resolve_runtime_shim(command, direct_candidate);
250        }
251        #[cfg(windows)]
252        for extension in windows_path_extensions(pathext_env.as_ref()) {
253            let candidate = dir.join(format!("{command}{extension}"));
254            if candidate.is_file() {
255                return resolve_runtime_shim(command, candidate);
256            }
257        }
258    }
259    #[cfg(not(windows))]
260    let _ = pathext_env;
261    command_path.to_path_buf()
262}
263
264fn resolve_runtime_shim(command: &str, candidate: PathBuf) -> PathBuf {
265    resolve_mise_shim(command, &candidate).unwrap_or(candidate)
266}
267
268fn resolve_mise_shim(command: &str, candidate: &Path) -> Option<PathBuf> {
269    let canonical = candidate.canonicalize().ok()?;
270    if canonical.file_stem()?.to_string_lossy() != "mise" {
271        return None;
272    }
273    let output = std::process::Command::new(&canonical)
274        .args(["which", command])
275        .output()
276        .ok()?;
277    if !output.status.success() {
278        return None;
279    }
280    let resolved = PathBuf::from(String::from_utf8(output.stdout).ok()?.trim());
281    resolved.is_file().then_some(resolved)
282}
283
284#[cfg(windows)]
285fn windows_path_extensions(pathext_env: Option<&OsString>) -> Vec<String> {
286    pathext_env
287        .and_then(|value| value.to_str().map(ToOwned::to_owned))
288        .unwrap_or_else(|| ".COM;.EXE;.BAT;.CMD".to_owned())
289        .split(';')
290        .filter(|extension| !extension.is_empty())
291        .map(|extension| {
292            if extension.starts_with('.') {
293                extension.to_owned()
294            } else {
295                format!(".{extension}")
296            }
297        })
298        .collect()
299}
300
301pub fn output_exceeded_message(stream: &str, max_output_bytes: usize) -> String {
302    format!("sidecar {stream} output exceeds {max_output_bytes} bytes")
303}
304
305/// Resolves a provider/tool's declared env requirements against the process
306/// environment. `prefix` is the caller's product env-namespace (e.g.
307/// `"SOMA"`) — this crate has no product identity of its own, so callers
308/// must supply it explicitly rather than this module hard-coding one.
309pub fn collect_provider_env(
310    provider_requirements: &[EnvRequirement],
311    tool_requirements: &[EnvRequirement],
312    prefix: &str,
313    provider: &str,
314    action: &str,
315) -> Result<Vec<(String, String)>, ProviderError> {
316    let mut env = Vec::new();
317    for requirement in provider_requirements.iter().chain(tool_requirements) {
318        let name = requirement.runtime_name(prefix);
319        let value = std::env::var(&name)
320            .ok()
321            .or_else(|| {
322                requirement
323                    .allow_unprefixed
324                    .then(|| std::env::var(&requirement.name).ok())
325                    .flatten()
326            })
327            .or_else(|| {
328                requirement
329                    .default
330                    .as_ref()
331                    .and_then(serde_json::Value::as_str)
332                    .map(ToOwned::to_owned)
333            });
334        match value {
335            Some(value) => env.push((name, value)),
336            None if requirement.required => {
337                return Err(ProviderError::validation(
338                    provider,
339                    action,
340                    "missing_provider_env",
341                    format!("missing required provider env `{name}`"),
342                ));
343            }
344            None => {}
345        }
346    }
347    Ok(env)
348}
349
350async fn read_bounded<R>(mut reader: R, max_output_bytes: usize) -> io::Result<(Vec<u8>, bool)>
351where
352    R: AsyncRead + Unpin,
353{
354    let mut bytes = Vec::new();
355    let mut exceeded = false;
356    let mut chunk = [0u8; 8192];
357    loop {
358        let read = reader.read(&mut chunk).await?;
359        if read == 0 {
360            return Ok((bytes, exceeded));
361        }
362        let remaining = max_output_bytes.saturating_sub(bytes.len());
363        if remaining >= read && !exceeded {
364            bytes.extend_from_slice(&chunk[..read]);
365        } else {
366            exceeded = true;
367            if remaining > 0 {
368                bytes.extend_from_slice(&chunk[..remaining]);
369            }
370        }
371    }
372}
373
374#[cfg(test)]
375#[path = "sidecar_tests.rs"]
376mod tests;
soma_provider_adapters/sidecar.rs

soma_provider_adapters/
sidecar.rs
