brainos_delegate/
subprocess.rs

1//! Generic subprocess-backed agent delegate.
2//!
3//! Spawns a configured binary with templated args, streams the task spec
4//! through stdin, captures stdout/stderr, and returns an [`AgentResult`].
5//! Every CLI delegate the orchestrator can hand work to — discovered or
6//! manually configured — runs through this single adapter.
7
8use std::path::PathBuf;
9use std::process::Stdio;
10use std::time::Duration;
11
12use async_trait::async_trait;
13use chrono::Utc;
14use tokio::io::AsyncWriteExt;
15use tokio::process::Command;
16use tokio::time::timeout;
17
18use crate::traits::{
19    AgentCapabilities, AgentDelegate, AgentError, AgentResult, AgentTask, AgentTaskStatus,
20};
21
22/// How many bytes of stdout/stderr we keep — beyond this we truncate
23/// and annotate. Keeps delegates from blowing the orchestrator's
24/// in-memory task state.
25const CAPTURE_CAP: usize = 64 * 1024;
26
27/// Configuration for a subprocess delegate.
28#[derive(Debug, Clone)]
29pub struct SubprocessAgentConfig {
30    /// Logical name registered with the registry.
31    pub name: String,
32    /// Binary to spawn. Resolved via `$PATH` unless absolute.
33    pub binary: String,
34    /// Arguments passed before the prompt. Use `{prompt}` or `{task_id}`
35    /// in any element and they're substituted at spawn time; otherwise
36    /// the prompt goes on stdin.
37    pub args: Vec<String>,
38    /// Optional working directory override. When `None`, the task's
39    /// `workdir` is used, falling back to the current process cwd.
40    pub workdir: Option<PathBuf>,
41    /// Declared capabilities.
42    pub capabilities: AgentCapabilities,
43    /// If `true`, the rendered prompt is written to the child's stdin
44    /// instead of being templated into `args`.
45    pub prompt_via_stdin: bool,
46    /// Args used by [`AgentDelegate::health_check`] to probe whether the
47    /// binary is installed and runnable. Empty disables the probe (the
48    /// delegate then optimistically reports healthy).
49    pub version_args: Vec<String>,
50}
51
52impl SubprocessAgentConfig {
53    pub fn new(name: impl Into<String>, binary: impl Into<String>) -> Self {
54        Self {
55            name: name.into(),
56            binary: binary.into(),
57            args: Vec::new(),
58            workdir: None,
59            capabilities: AgentCapabilities::default(),
60            prompt_via_stdin: true,
61            version_args: vec!["--version".to_string()],
62        }
63    }
64
65    pub fn with_args<I, S>(mut self, args: I) -> Self
66    where
67        I: IntoIterator<Item = S>,
68        S: Into<String>,
69    {
70        self.args = args.into_iter().map(Into::into).collect();
71        self
72    }
73
74    pub fn with_workdir(mut self, workdir: impl Into<PathBuf>) -> Self {
75        self.workdir = Some(workdir.into());
76        self
77    }
78
79    pub fn with_capabilities(mut self, caps: AgentCapabilities) -> Self {
80        self.capabilities = caps;
81        self
82    }
83
84    pub fn with_prompt_via_stdin(mut self, via_stdin: bool) -> Self {
85        self.prompt_via_stdin = via_stdin;
86        self
87    }
88
89    pub fn with_version_args<I, S>(mut self, args: I) -> Self
90    where
91        I: IntoIterator<Item = S>,
92        S: Into<String>,
93    {
94        self.version_args = args.into_iter().map(Into::into).collect();
95        self
96    }
97}
98
99pub struct SubprocessAgentDelegate {
100    config: SubprocessAgentConfig,
101}
102
103impl SubprocessAgentDelegate {
104    pub fn new(config: SubprocessAgentConfig) -> Self {
105        Self { config }
106    }
107
108    /// Binary path as configured.
109    pub fn binary(&self) -> &str {
110        &self.config.binary
111    }
112
113    /// The prompt we'd hand to the underlying binary. Exposed so tests
114    /// (and other crates wiring custom delegates) can reuse the format.
115    pub fn render_prompt(task: &AgentTask) -> String {
116        let mut out = String::new();
117        out.push_str("# Task\n");
118        out.push_str(task.description.trim());
119        out.push_str("\n\n");
120        let ctx = task.context.render();
121        if !ctx.is_empty() {
122            out.push_str("# Context\n");
123            out.push_str(&ctx);
124            out.push_str("\n\n");
125        }
126        out.push_str("# Workdir\n");
127        match &task.workdir {
128            Some(p) => out.push_str(&p.display().to_string()),
129            None => out.push_str("<inherit>"),
130        }
131        out.push('\n');
132        out
133    }
134
135    fn build_args(&self, prompt: &str, task: &AgentTask) -> Vec<String> {
136        self.config
137            .args
138            .iter()
139            .map(|a| a.replace("{prompt}", prompt).replace("{task_id}", &task.id))
140            .collect()
141    }
142}
143
144#[async_trait]
145impl AgentDelegate for SubprocessAgentDelegate {
146    fn name(&self) -> &str {
147        &self.config.name
148    }
149
150    fn capabilities(&self) -> AgentCapabilities {
151        self.config.capabilities.clone()
152    }
153
154    async fn delegate(&self, task: AgentTask) -> Result<AgentResult, AgentError> {
155        let started_at = Utc::now();
156        let prompt = Self::render_prompt(&task);
157        let args = self.build_args(&prompt, &task);
158
159        let workdir = self
160            .config
161            .workdir
162            .clone()
163            .or_else(|| task.workdir.clone())
164            .unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")));
165
166        let mut cmd = Command::new(&self.config.binary);
167        cmd.args(&args)
168            .current_dir(&workdir)
169            .stdin(if self.config.prompt_via_stdin {
170                Stdio::piped()
171            } else {
172                Stdio::null()
173            })
174            .stdout(Stdio::piped())
175            .stderr(Stdio::piped())
176            .kill_on_drop(true);
177
178        tracing::debug!(
179            agent = %self.config.name,
180            task_id = %task.id,
181            binary = %self.config.binary,
182            workdir = %workdir.display(),
183            "Spawning subprocess delegate"
184        );
185
186        let mut child = cmd
187            .spawn()
188            .map_err(|e| AgentError::Launch(format!("spawn {}: {e}", self.config.binary)))?;
189
190        if self.config.prompt_via_stdin {
191            if let Some(stdin) = child.stdin.as_mut() {
192                match stdin.write_all(prompt.as_bytes()).await {
193                    Ok(()) => {}
194                    // A child that exits before reading stdin (e.g. `false`,
195                    // or a CLI that rejects the prompt) closes the pipe and
196                    // our write lands on EPIPE. That's not our error to
197                    // surface — the exit status below is the real signal.
198                    Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => {
199                        tracing::debug!(
200                            agent = %self.config.name,
201                            task_id = %task.id,
202                            "child closed stdin before prompt was fully written"
203                        );
204                    }
205                    Err(e) => return Err(AgentError::Io(format!("writing stdin: {e}"))),
206                }
207            }
208            // Drop stdin so the child gets EOF and can finish.
209            drop(child.stdin.take());
210        }
211
212        let wait = async {
213            child
214                .wait_with_output()
215                .await
216                .map_err(|e| AgentError::Io(format!("waiting on {}: {e}", self.config.name)))
217        };
218
219        let output = match timeout(Duration::from_secs(task.timeout_secs), wait).await {
220            Ok(res) => res?,
221            Err(_) => {
222                tracing::warn!(
223                    agent = %self.config.name,
224                    task_id = %task.id,
225                    secs = task.timeout_secs,
226                    "Delegate timed out"
227                );
228                return Err(AgentError::Timeout {
229                    task_id: task.id.clone(),
230                    secs: task.timeout_secs,
231                });
232            }
233        };
234
235        let completed_at = Utc::now();
236        let stdout = truncate(String::from_utf8_lossy(&output.stdout).into_owned());
237        let stderr = truncate(String::from_utf8_lossy(&output.stderr).into_owned());
238        let code = output.status.code().unwrap_or(-1);
239
240        let status = if output.status.success() {
241            AgentTaskStatus::Succeeded
242        } else {
243            AgentTaskStatus::Failed
244        };
245
246        let summary = if status == AgentTaskStatus::Succeeded {
247            first_line(&stdout, &task.description)
248        } else {
249            format!("{} failed (exit {})", self.config.name, code)
250        };
251
252        Ok(AgentResult {
253            task_id: task.id,
254            status,
255            summary,
256            artifacts: vec![],
257            stdout,
258            stderr,
259            exit_code: Some(code),
260            started_at,
261            completed_at,
262        })
263    }
264
265    /// Probe by running the configured `version_args` with a short
266    /// timeout. Confirms the binary is installed and runnable without
267    /// consuming task quota. Returns `true` when no `version_args` are
268    /// configured (caller opted out of probing).
269    async fn health_check(&self) -> bool {
270        if self.config.version_args.is_empty() {
271            return true;
272        }
273        let probe = Command::new(&self.config.binary)
274            .args(&self.config.version_args)
275            .stdin(Stdio::null())
276            .stdout(Stdio::piped())
277            .stderr(Stdio::piped())
278            .kill_on_drop(true)
279            .output();
280        matches!(
281            timeout(Duration::from_secs(5), probe).await,
282            Ok(Ok(o)) if o.status.success()
283        )
284    }
285}
286
287fn truncate(mut s: String) -> String {
288    if s.len() <= CAPTURE_CAP {
289        return s;
290    }
291    s.truncate(CAPTURE_CAP);
292    s.push_str("\n…[truncated]");
293    s
294}
295
296fn first_line(stdout: &str, fallback: &str) -> String {
297    stdout
298        .lines()
299        .map(str::trim)
300        .find(|l| !l.is_empty())
301        .map(|l| l.to_string())
302        .unwrap_or_else(|| format!("completed: {fallback}"))
303}
304
305#[cfg(test)]
306mod tests {
307    use super::*;
308    use crate::traits::AgentContext;
309
310    #[test]
311    fn render_prompt_includes_description_and_context() {
312        let task = AgentTask::new("add a test").with_context(
313            AgentContext::new()
314                .push("memory", "project uses clap 4")
315                .push("conventions", "tabs, trailing newline"),
316        );
317        let rendered = SubprocessAgentDelegate::render_prompt(&task);
318        assert!(rendered.starts_with("# Task"));
319        assert!(rendered.contains("add a test"));
320        assert!(rendered.contains("### memory"));
321        assert!(rendered.contains("### conventions"));
322        assert!(rendered.contains("# Workdir"));
323    }
324
325    #[test]
326    fn truncate_respects_cap() {
327        let s = "a".repeat(CAPTURE_CAP + 500);
328        let out = truncate(s);
329        assert!(out.ends_with("…[truncated]"));
330        assert!(out.len() <= CAPTURE_CAP + "\n…[truncated]".len());
331    }
332
333    #[test]
334    fn build_args_substitutes_placeholders() {
335        let cfg = SubprocessAgentConfig::new("mock", "echo")
336            .with_args(["--task", "{task_id}", "--prompt", "{prompt}"])
337            .with_prompt_via_stdin(false);
338        let delegate = SubprocessAgentDelegate::new(cfg);
339        let task = AgentTask::new("hi");
340        let args = delegate.build_args("PROMPT", &task);
341        assert_eq!(args[0], "--task");
342        assert_eq!(args[1], task.id);
343        assert_eq!(args[2], "--prompt");
344        assert_eq!(args[3], "PROMPT");
345    }
346
347    // End-to-end using `/bin/cat` (POSIX; skipped on Windows CI).
348    #[cfg(unix)]
349    #[tokio::test]
350    async fn echoes_prompt_via_stdin() {
351        let cfg = SubprocessAgentConfig::new("cat-agent", "/bin/cat");
352        let delegate = SubprocessAgentDelegate::new(cfg);
353        let task = AgentTask::new("say hello").with_timeout_secs(5);
354        let result = delegate.delegate(task).await.unwrap();
355        assert_eq!(result.status, AgentTaskStatus::Succeeded);
356        assert!(result.stdout.contains("# Task"));
357        assert!(result.stdout.contains("say hello"));
358    }
359
360    #[cfg(unix)]
361    #[tokio::test]
362    async fn times_out_when_child_hangs() {
363        let cfg = SubprocessAgentConfig::new("sleep-agent", "/bin/sleep").with_args(["30"]);
364        let delegate = SubprocessAgentDelegate::new(cfg);
365        let task = AgentTask::new("noop").with_timeout_secs(1);
366        let err = delegate.delegate(task).await.unwrap_err();
367        matches!(err, AgentError::Timeout { .. });
368    }
369
370    #[cfg(unix)]
371    #[tokio::test]
372    async fn nonzero_exit_surfaces_as_failed_status() {
373        let cfg = SubprocessAgentConfig::new("false-agent", "/usr/bin/false");
374        let delegate = SubprocessAgentDelegate::new(cfg);
375        let task = AgentTask::new("noop").with_timeout_secs(5);
376        let result = delegate.delegate(task).await.unwrap();
377        assert_eq!(result.status, AgentTaskStatus::Failed);
378        assert_ne!(result.exit_code, Some(0));
379    }
380}
brainos_delegate/subprocess.rs

brainos_delegate/
subprocess.rs
