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
59pub struct BoundedOutput {
60    pub output: Output,
61    pub stdout_exceeded: bool,
62    pub stderr_exceeded: bool,
63}
64
65pub async fn run_bounded_sidecar(
66    command: &str,
67    args: &[&str],
68    env: Vec<(String, String)>,
69    input: &[u8],
70    timeout_ms: u64,
71    max_output_bytes: usize,
72) -> Result<BoundedOutput, SidecarError> {
73    let resolved_command = resolve_sidecar_command(command);
74    let mut command = Command::new(resolved_command);
75    command
76        .args(args)
77        .kill_on_drop(true)
78        .env_clear()
79        .stdin(Stdio::piped())
80        .stdout(Stdio::piped())
81        .stderr(Stdio::piped());
82    apply_sidecar_base_env(&mut command);
83    command.envs(env);
84
85    let mut child = command.spawn().map_err(SidecarError::Io)?;
86
87    let stdout = child
88        .stdout
89        .take()
90        .ok_or_else(|| io::Error::other("sidecar stdout pipe was not captured"))
91        .map_err(SidecarError::Io)?;
92    let stderr = child
93        .stderr
94        .take()
95        .ok_or_else(|| io::Error::other("sidecar stderr pipe was not captured"))
96        .map_err(SidecarError::Io)?;
97    let stdout_task = tokio::spawn(read_bounded(stdout, max_output_bytes));
98    let stderr_task = tokio::spawn(read_bounded(stderr, max_output_bytes));
99
100    if let Some(mut stdin) = child.stdin.take() {
101        stdin.write_all(input).await.map_err(SidecarError::Io)?;
102    }
103
104    let status = match timeout(Duration::from_millis(timeout_ms), child.wait()).await {
105        Ok(status) => status.map_err(SidecarError::Io)?,
106        Err(_) => {
107            let _ = child.kill().await;
108            let _ = child.wait().await;
109            stdout_task.abort();
110            stderr_task.abort();
111            return Err(SidecarError::Timeout);
112        }
113    };
114
115    let (stdout, stdout_exceeded) = stdout_task
116        .await
117        .map_err(SidecarError::Join)?
118        .map_err(SidecarError::Io)?;
119    let (stderr, stderr_exceeded) = stderr_task
120        .await
121        .map_err(SidecarError::Join)?
122        .map_err(SidecarError::Io)?;
123
124    Ok(BoundedOutput {
125        output: Output {
126            status,
127            stdout,
128            stderr,
129        },
130        stdout_exceeded,
131        stderr_exceeded,
132    })
133}
134
135fn apply_sidecar_base_env(command: &mut Command) {
136    for (key, value) in sidecar_base_env() {
137        command.env(key, value);
138    }
139}
140
141#[cfg(windows)]
142pub fn sidecar_base_env() -> Vec<(OsString, OsString)> {
143    let mut env = Vec::new();
144    for key in ["SystemRoot", "WINDIR", "COMSPEC", "PATHEXT", "TEMP", "TMP"] {
145        if let Some(value) = std::env::var_os(key) {
146            env.push((OsString::from(key), value));
147        }
148    }
149    env
150}
151
152#[cfg(not(windows))]
153pub fn sidecar_base_env() -> Vec<(OsString, OsString)> {
154    let mut env = Vec::new();
155    for key in ["HOME", "TMPDIR", "TEMP", "TMP"] {
156        if let Some(value) = std::env::var_os(key) {
157            env.push((OsString::from(key), value));
158        }
159    }
160    env
161}
162
163pub fn resolve_sidecar_command(command: &str) -> PathBuf {
164    resolve_sidecar_command_with_env(
165        command,
166        std::env::var_os("PATH"),
167        std::env::var_os("PATHEXT"),
168    )
169}
170
171fn resolve_sidecar_command_with_env(
172    command: &str,
173    path_env: Option<OsString>,
174    pathext_env: Option<OsString>,
175) -> PathBuf {
176    let command_path = Path::new(command);
177    if command_path.components().count() > 1 || command_path.is_absolute() {
178        return command_path.to_path_buf();
179    }
180
181    let Some(path_env) = path_env else {
182        return command_path.to_path_buf();
183    };
184    for dir in std::env::split_paths(&path_env) {
185        if command_path.extension().is_some() {
186            let candidate = dir.join(command_path);
187            if candidate.is_file() {
188                return resolve_runtime_shim(command, candidate);
189            }
190            continue;
191        }
192        let direct_candidate = dir.join(command_path);
193        if direct_candidate.is_file() {
194            return resolve_runtime_shim(command, direct_candidate);
195        }
196        #[cfg(windows)]
197        for extension in windows_path_extensions(pathext_env.as_ref()) {
198            let candidate = dir.join(format!("{command}{extension}"));
199            if candidate.is_file() {
200                return resolve_runtime_shim(command, candidate);
201            }
202        }
203    }
204    #[cfg(not(windows))]
205    let _ = pathext_env;
206    command_path.to_path_buf()
207}
208
209fn resolve_runtime_shim(command: &str, candidate: PathBuf) -> PathBuf {
210    resolve_mise_shim(command, &candidate).unwrap_or(candidate)
211}
212
213fn resolve_mise_shim(command: &str, candidate: &Path) -> Option<PathBuf> {
214    let canonical = candidate.canonicalize().ok()?;
215    if canonical.file_stem()?.to_string_lossy() != "mise" {
216        return None;
217    }
218    let output = std::process::Command::new(&canonical)
219        .args(["which", command])
220        .output()
221        .ok()?;
222    if !output.status.success() {
223        return None;
224    }
225    let resolved = PathBuf::from(String::from_utf8(output.stdout).ok()?.trim());
226    resolved.is_file().then_some(resolved)
227}
228
229#[cfg(windows)]
230fn windows_path_extensions(pathext_env: Option<&OsString>) -> Vec<String> {
231    pathext_env
232        .and_then(|value| value.to_str().map(ToOwned::to_owned))
233        .unwrap_or_else(|| ".COM;.EXE;.BAT;.CMD".to_owned())
234        .split(';')
235        .filter(|extension| !extension.is_empty())
236        .map(|extension| {
237            if extension.starts_with('.') {
238                extension.to_owned()
239            } else {
240                format!(".{extension}")
241            }
242        })
243        .collect()
244}
245
246pub fn output_exceeded_message(stream: &str, max_output_bytes: usize) -> String {
247    format!("sidecar {stream} output exceeds {max_output_bytes} bytes")
248}
249
250/// Resolves a provider/tool's declared env requirements against the process
251/// environment. `prefix` is the caller's product env-namespace (e.g.
252/// `"SOMA"`) — this crate has no product identity of its own, so callers
253/// must supply it explicitly rather than this module hard-coding one.
254pub fn collect_provider_env(
255    provider_requirements: &[EnvRequirement],
256    tool_requirements: &[EnvRequirement],
257    prefix: &str,
258    provider: &str,
259    action: &str,
260) -> Result<Vec<(String, String)>, ProviderError> {
261    let mut env = Vec::new();
262    for requirement in provider_requirements.iter().chain(tool_requirements) {
263        let name = requirement.runtime_name(prefix);
264        let value = std::env::var(&name)
265            .ok()
266            .or_else(|| {
267                requirement
268                    .allow_unprefixed
269                    .then(|| std::env::var(&requirement.name).ok())
270                    .flatten()
271            })
272            .or_else(|| {
273                requirement
274                    .default
275                    .as_ref()
276                    .and_then(serde_json::Value::as_str)
277                    .map(ToOwned::to_owned)
278            });
279        match value {
280            Some(value) => env.push((name, value)),
281            None if requirement.required => {
282                return Err(ProviderError::validation(
283                    provider,
284                    action,
285                    "missing_provider_env",
286                    format!("missing required provider env `{name}`"),
287                ));
288            }
289            None => {}
290        }
291    }
292    Ok(env)
293}
294
295async fn read_bounded<R>(mut reader: R, max_output_bytes: usize) -> io::Result<(Vec<u8>, bool)>
296where
297    R: AsyncRead + Unpin,
298{
299    let mut bytes = Vec::new();
300    let mut exceeded = false;
301    let mut chunk = [0u8; 8192];
302    loop {
303        let read = reader.read(&mut chunk).await?;
304        if read == 0 {
305            return Ok((bytes, exceeded));
306        }
307        let remaining = max_output_bytes.saturating_sub(bytes.len());
308        if remaining >= read && !exceeded {
309            bytes.extend_from_slice(&chunk[..read]);
310        } else {
311            exceeded = true;
312            if remaining > 0 {
313                bytes.extend_from_slice(&chunk[..remaining]);
314            }
315        }
316    }
317}
318
319#[cfg(test)]
320#[path = "sidecar_tests.rs"]
321mod tests;
soma_provider_adapters/sidecar.rs

soma_provider_adapters/
sidecar.rs
