agentbridge/
executor.rs

1//! OpenCode executor adapter.
2//!
3//! AgentBridge is not a coding agent. This module starts a local OpenCode CLI
4//! process against the configured workspace, captures exit status, and returns
5//! a compact result. Internal model reasoning is discarded.
6
7use std::path::{Path, PathBuf};
8use std::process::Stdio;
9
10use tokio::io::AsyncReadExt;
11use tokio::process::{Child, Command};
12
13use crate::config::{ExecutorConfig, ExecutorMode, ALLOWED_EXECUTOR_TYPES};
14use crate::protocol::C2cPlan;
15
16const MAX_CAPTURE_BYTES: usize = 64 * 1024;
17const MAX_SUMMARY_CHARS: usize = 2_000;
18
19#[derive(Debug, thiserror::Error)]
20pub enum ExecutorError {
21    #[error(
22        "OpenCode is not installed or not found on PATH (looked for `{0}`). \
23         Install OpenCode and ensure the executable is available."
24    )]
25    NotInstalled(String),
26    #[error("executor type `{0}` is not allowed (allowlist: opencode, codex, claude)")]
27    TypeNotAllowed(String),
28    #[error("executor type `{0}` is not implemented; V0.2 only supports OpenCode")]
29    TypeNotImplemented(String),
30    #[error("an executor task is already running (task_id={0})")]
31    AlreadyRunning(String),
32    #[error("no running executor task")]
33    NotRunning,
34    #[error("executor command is invalid: {0}")]
35    InvalidCommand(String),
36    #[error("executor workspace is invalid: {0}")]
37    InvalidWorkspace(String),
38    #[error("failed to start OpenCode: {0}")]
39    Spawn(String),
40    #[error("executor crashed: {0}")]
41    Crashed(String),
42    #[error("executor was cancelled")]
43    Cancelled,
44    #[error("{0}")]
45    Other(String),
46}
47
48/// Local coding agent that may write files and run commands inside the workspace.
49pub trait Executor: Send + Sync {
50    fn name(&self) -> &'static str;
51    fn detect(&self) -> Result<PathBuf, ExecutorError>;
52    fn start_task(&self, plan: &C2cPlan, workspace: &Path) -> Result<SpawnedTask, ExecutorError>;
53    fn status(&self, task_id: &str) -> Result<ExecutorSnapshot, ExecutorError>;
54    fn result(&self, task_id: &str) -> Result<ExecutorSnapshot, ExecutorError>;
55    fn cancel(&self, task_id: &str) -> Result<(), ExecutorError>;
56}
57
58/// Handle to a live OpenCode process. `status` / `result` / `cancel` are
59/// implemented by [`crate::task::TaskRuntime`], which owns this handle.
60pub struct SpawnedTask {
61    pub child: Child,
62    pub pid: u32,
63    pub executable: PathBuf,
64}
65
66#[derive(Debug, Clone, PartialEq, Eq)]
67pub struct ExecutorSnapshot {
68    pub task_id: String,
69    pub status: String,
70    pub summary: Option<String>,
71    pub exit_code: Option<i32>,
72    pub tests: Option<String>,
73    pub changed_files: Vec<String>,
74    pub error: Option<String>,
75}
76
77#[derive(Debug, Clone, PartialEq, Eq)]
78pub struct ExecutorOutcome {
79    pub exit_code: Option<i32>,
80    pub summary: String,
81    pub tests_excerpt: Option<String>,
82    pub cancelled: bool,
83    pub error: Option<String>,
84}
85
86#[derive(Debug, Clone)]
87pub struct OpenCodeExecutor {
88    command: String,
89    mode: ExecutorMode,
90}
91
92impl OpenCodeExecutor {
93    pub fn from_config(cfg: &ExecutorConfig) -> Result<Self, ExecutorError> {
94        validate_executor_type(&cfg.kind)?;
95        let command = cfg.command.trim();
96        if command.is_empty() {
97            return Err(ExecutorError::InvalidCommand(
98                "command must not be empty".into(),
99            ));
100        }
101        if command.contains('\0') {
102            return Err(ExecutorError::InvalidCommand("command contains NUL".into()));
103        }
104        Ok(Self {
105            command: command.to_string(),
106            mode: cfg.mode,
107        })
108    }
109
110    pub fn command(&self) -> &str {
111        &self.command
112    }
113
114    pub fn mode(&self) -> ExecutorMode {
115        self.mode
116    }
117}
118
119impl Executor for OpenCodeExecutor {
120    fn name(&self) -> &'static str {
121        "opencode"
122    }
123
124    fn detect(&self) -> Result<PathBuf, ExecutorError> {
125        find_executable(&self.command)
126            .ok_or_else(|| ExecutorError::NotInstalled(self.command.clone()))
127    }
128
129    fn start_task(&self, plan: &C2cPlan, workspace: &Path) -> Result<SpawnedTask, ExecutorError> {
130        plan.validate()
131            .map_err(|e| ExecutorError::Other(e.to_string()))?;
132        if !workspace.is_dir() {
133            return Err(ExecutorError::InvalidWorkspace(
134                workspace.display().to_string(),
135            ));
136        }
137        let exe = self.detect()?;
138        let prompt = executor_argv_prompt();
139        let child = spawn_opencode(&exe, workspace, prompt)?;
140        let pid = child
141            .id()
142            .ok_or_else(|| ExecutorError::Spawn("OpenCode process has no pid".into()))?;
143        Ok(SpawnedTask {
144            child,
145            pid,
146            executable: exe,
147        })
148    }
149
150    fn status(&self, _task_id: &str) -> Result<ExecutorSnapshot, ExecutorError> {
151        Err(ExecutorError::Other(
152            "status is provided by TaskRuntime".into(),
153        ))
154    }
155
156    fn result(&self, _task_id: &str) -> Result<ExecutorSnapshot, ExecutorError> {
157        Err(ExecutorError::Other(
158            "result is provided by TaskRuntime".into(),
159        ))
160    }
161
162    fn cancel(&self, _task_id: &str) -> Result<(), ExecutorError> {
163        Err(ExecutorError::Other(
164            "cancel is provided by TaskRuntime".into(),
165        ))
166    }
167}
168
169pub fn validate_executor_type(kind: &str) -> Result<(), ExecutorError> {
170    let kind = kind.trim().to_ascii_lowercase();
171    if !ALLOWED_EXECUTOR_TYPES
172        .iter()
173        .any(|allowed| *allowed == kind)
174    {
175        return Err(ExecutorError::TypeNotAllowed(kind));
176    }
177    if kind != "opencode" {
178        return Err(ExecutorError::TypeNotImplemented(kind));
179    }
180    Ok(())
181}
182
183/// Locate `command` on PATH (Windows-aware). Absolute paths from config are allowed.
184pub fn find_executable(command: &str) -> Option<PathBuf> {
185    let command = command.trim();
186    if command.is_empty() || command.contains('\0') {
187        return None;
188    }
189    let path = Path::new(command);
190    if path.is_absolute() {
191        return existing_file(path);
192    }
193    if command.contains("..") {
194        return None;
195    }
196    if path.components().count() > 1 {
197        return existing_file(path);
198    }
199
200    let path_var = std::env::var_os("PATH")?;
201    for dir in std::env::split_paths(&path_var) {
202        for candidate in candidates_in_dir(&dir, command) {
203            if let Some(found) = existing_file(&candidate) {
204                return Some(found);
205            }
206        }
207    }
208    None
209}
210
211fn existing_file(path: &Path) -> Option<PathBuf> {
212    if path.is_file() {
213        return Some(path.to_path_buf());
214    }
215    #[cfg(windows)]
216    {
217        for ext in pathext() {
218            let with_ext = if has_extension(path) {
219                path.to_path_buf()
220            } else {
221                let mut p = path.as_os_str().to_os_string();
222                p.push(&ext);
223                PathBuf::from(p)
224            };
225            if with_ext.is_file() {
226                return Some(with_ext);
227            }
228        }
229    }
230    None
231}
232
233fn candidates_in_dir(dir: &Path, command: &str) -> Vec<PathBuf> {
234    let base = dir.join(command);
235    let mut out = vec![base.clone()];
236    #[cfg(windows)]
237    {
238        if !has_extension(&base) {
239            for ext in pathext() {
240                let mut p = base.as_os_str().to_os_string();
241                p.push(&ext);
242                out.push(PathBuf::from(p));
243            }
244        }
245    }
246    out
247}
248
249#[cfg(windows)]
250fn has_extension(path: &Path) -> bool {
251    path.extension().is_some()
252}
253
254#[cfg(windows)]
255fn pathext() -> Vec<String> {
256    let raw = std::env::var("PATHEXT").unwrap_or_else(|_| ".COM;.EXE;.BAT;.CMD".into());
257    raw.split(';')
258        .map(|s| s.trim().to_string())
259        .filter(|s| !s.is_empty())
260        .collect()
261}
262
263pub fn opencode_version(command: &str) -> Option<String> {
264    let exe = find_executable(command)?;
265    let output = std::process::Command::new(&exe)
266        .arg("--version")
267        .stdout(Stdio::piped())
268        .stderr(Stdio::piped())
269        .output()
270        .ok()?;
271    let stdout = String::from_utf8_lossy(&output.stdout);
272    let line = stdout
273        .lines()
274        .map(str::trim)
275        .find(|l| !l.is_empty())
276        .map(ToOwned::to_owned);
277    if let Some(line) = line {
278        return Some(line);
279    }
280    Some(exe.display().to_string())
281}
282
283/// Short argv payload. The full PLAN is in `.agentbridge/current.c2c` so Windows
284/// `.cmd` wrappers are not given multiline arguments.
285fn executor_argv_prompt() -> &'static str {
286    "Read .agentbridge/current.c2c and implement that PLAN. Stay inside this working directory. Run the TESTS. Print a short summary of changed files and test results. Do not paste source or internal reasoning."
287}
288
289fn spawn_opencode(exe: &Path, workspace: &Path, prompt: &str) -> Result<Child, ExecutorError> {
290    let mut cmd = Command::new(exe);
291    cmd.arg("run")
292        .arg("--auto")
293        .arg(prompt)
294        .current_dir(workspace)
295        .stdin(Stdio::null())
296        .stdout(Stdio::piped())
297        .stderr(Stdio::piped())
298        .kill_on_drop(true)
299        .env_remove("AGENTBRIDGE_AUTH_TOKEN");
300
301    #[cfg(windows)]
302    {
303        const CREATE_NEW_PROCESS_GROUP: u32 = 0x00000200;
304        cmd.creation_flags(CREATE_NEW_PROCESS_GROUP);
305    }
306
307    cmd.spawn().map_err(|err| {
308        if err.kind() == std::io::ErrorKind::NotFound {
309            ExecutorError::NotInstalled(exe.display().to_string())
310        } else {
311            ExecutorError::Spawn(err.to_string())
312        }
313    })
314}
315
316/// Wait for OpenCode, honouring cancel. Returns a compact outcome (no reasoning).
317pub async fn run_spawned(
318    mut child: Child,
319    cancel: tokio_util::sync::CancellationToken,
320    mode: ExecutorMode,
321) -> ExecutorOutcome {
322    let stream_to_stdout = mode == ExecutorMode::Stream;
323    let stdout = child.stdout.take();
324    let stderr = child.stderr.take();
325    let out_buf = tokio::spawn(read_limited(stdout, stream_to_stdout));
326    let err_buf = tokio::spawn(read_limited(stderr, stream_to_stdout));
327
328    enum Finish {
329        Status(std::io::Result<std::process::ExitStatus>),
330        Cancelled,
331    }
332
333    let finish = tokio::select! {
334        _ = cancel.cancelled() => {
335            if let Some(pid) = child.id() {
336                let _ = kill_process_tree(pid);
337            }
338            let _ = child.kill().await;
339            Finish::Cancelled
340        }
341        status = child.wait() => Finish::Status(status),
342    };
343
344    if matches!(finish, Finish::Cancelled) {
345        let _ = child.wait().await;
346    }
347
348    let stdout_bytes = out_buf.await.unwrap_or_default();
349    let stderr_bytes = err_buf.await.unwrap_or_default();
350    let stdout = String::from_utf8_lossy(&stdout_bytes);
351    let stderr = String::from_utf8_lossy(&stderr_bytes);
352    let stdout = strip_reasoning(&stdout);
353    let stderr = strip_reasoning(&stderr);
354
355    match finish {
356        Finish::Cancelled => ExecutorOutcome {
357            exit_code: None,
358            summary: "Executor was cancelled.".into(),
359            tests_excerpt: extract_tests_excerpt(&stdout),
360            cancelled: true,
361            error: None,
362        },
363        Finish::Status(Ok(status)) => {
364            let code = status.code();
365            let crashed = code.is_none() && !status.success();
366            let summary = sanitize_summary(&stdout, &stderr, code, crashed);
367            let error = if crashed {
368                Some("OpenCode process crashed (no exit code).".into())
369            } else if code.unwrap_or(0) != 0 {
370                Some(format!("OpenCode exited with code {}.", code.unwrap_or(-1)))
371            } else {
372                None
373            };
374            ExecutorOutcome {
375                exit_code: code,
376                summary,
377                tests_excerpt: extract_tests_excerpt(&stdout),
378                cancelled: false,
379                error,
380            }
381        }
382        Finish::Status(Err(err)) => ExecutorOutcome {
383            exit_code: None,
384            summary: format!("Executor crashed: {err}"),
385            tests_excerpt: extract_tests_excerpt(&stdout),
386            cancelled: false,
387            error: Some(err.to_string()),
388        },
389    }
390}
391
392async fn read_limited<R: tokio::io::AsyncRead + Unpin + Send + 'static>(
393    reader: Option<R>,
394    stream_to_stdout: bool,
395) -> Vec<u8> {
396    let Some(mut reader) = reader else {
397        return Vec::new();
398    };
399    let mut buf = Vec::new();
400    let mut chunk = [0u8; 4096];
401    loop {
402        match reader.read(&mut chunk).await {
403            Ok(0) => break,
404            Ok(n) => {
405                if stream_to_stdout {
406                    use std::io::Write;
407                    let _ = std::io::stdout().write_all(&chunk[..n]);
408                    let _ = std::io::stdout().flush();
409                }
410                if buf.len() < MAX_CAPTURE_BYTES {
411                    let take = n.min(MAX_CAPTURE_BYTES - buf.len());
412                    buf.extend_from_slice(&chunk[..take]);
413                }
414            }
415            Err(_) => break,
416        }
417    }
418    buf
419}
420
421pub fn kill_process_tree(pid: u32) -> Result<(), ExecutorError> {
422    #[cfg(windows)]
423    {
424        let _ = std::process::Command::new("taskkill")
425            .args(["/PID", &pid.to_string(), "/T", "/F"])
426            .stdout(Stdio::null())
427            .stderr(Stdio::null())
428            .status();
429        Ok(())
430    }
431    #[cfg(unix)]
432    {
433        let _ = std::process::Command::new("kill")
434            .args(["-TERM", &format!("-{pid}")])
435            .stdout(Stdio::null())
436            .stderr(Stdio::null())
437            .status();
438        let _ = std::process::Command::new("kill")
439            .args(["-TERM", &pid.to_string()])
440            .stdout(Stdio::null())
441            .stderr(Stdio::null())
442            .status();
443        Ok(())
444    }
445}
446
447pub fn process_is_alive(pid: u32) -> bool {
448    if pid == 0 {
449        return false;
450    }
451    #[cfg(windows)]
452    {
453        let output = std::process::Command::new("tasklist")
454            .args(["/FI", &format!("PID eq {pid}"), "/FO", "CSV", "/NH"])
455            .stdout(Stdio::piped())
456            .stderr(Stdio::null())
457            .output();
458        match output {
459            Ok(o) => {
460                let s = String::from_utf8_lossy(&o.stdout);
461                s.split(',')
462                    .any(|col| col.trim_matches('"') == pid.to_string())
463                    || s.contains(&pid.to_string())
464            }
465            Err(_) => false,
466        }
467    }
468    #[cfg(unix)]
469    {
470        std::process::Command::new("kill")
471            .args(["-0", &pid.to_string()])
472            .stdout(Stdio::null())
473            .stderr(Stdio::null())
474            .status()
475            .map(|s| s.success())
476            .unwrap_or(false)
477    }
478}
479
480fn strip_reasoning(text: &str) -> String {
481    let mut out = String::new();
482    let mut in_think = false;
483    for line in text.lines() {
484        let trimmed = line.trim();
485        if trimmed.contains("<think>") || trimmed.contains("<thinking>") {
486            in_think = true;
487        }
488        if in_think {
489            if trimmed.contains("</think>") || trimmed.contains("</thinking>") {
490                in_think = false;
491            }
492            continue;
493        }
494        if let Ok(value) = serde_json::from_str::<serde_json::Value>(trimmed) {
495            if is_reasoning_event(&value) {
496                continue;
497            }
498            if let Some(text) = json_text(&value) {
499                if !out.is_empty() {
500                    out.push('\n');
501                }
502                out.push_str(&text);
503                continue;
504            }
505        }
506        if trimmed.eq_ignore_ascii_case("thinking")
507            || trimmed.starts_with("thinking:")
508            || trimmed.starts_with("Reasoning:")
509        {
510            continue;
511        }
512        if !out.is_empty() {
513            out.push('\n');
514        }
515        out.push_str(line);
516    }
517    out
518}
519
520fn is_reasoning_event(value: &serde_json::Value) -> bool {
521    let ty = value
522        .get("type")
523        .or_else(|| value.get("kind"))
524        .and_then(|v| v.as_str())
525        .unwrap_or("")
526        .to_ascii_lowercase();
527    matches!(
528        ty.as_str(),
529        "thinking" | "reasoning" | "think" | "internal" | "thought"
530    )
531}
532
533fn json_text(value: &serde_json::Value) -> Option<String> {
534    if let Some(s) = value.get("text").and_then(|v| v.as_str()) {
535        return Some(s.to_string());
536    }
537    if let Some(s) = value.get("response").and_then(|v| v.as_str()) {
538        return Some(s.to_string());
539    }
540    if let Some(s) = value
541        .get("part")
542        .and_then(|v| v.get("text"))
543        .and_then(|v| v.as_str())
544    {
545        return Some(s.to_string());
546    }
547    None
548}
549
550fn sanitize_summary(stdout: &str, stderr: &str, exit_code: Option<i32>, crashed: bool) -> String {
551    if crashed {
552        return truncate_chars(
553            &format!(
554                "OpenCode crashed. {}",
555                stderr.trim().lines().next().unwrap_or("")
556            ),
557            MAX_SUMMARY_CHARS,
558        );
559    }
560    let body = stdout.trim();
561    if !body.is_empty() {
562        return truncate_chars(body, MAX_SUMMARY_CHARS);
563    }
564    let err = stderr.trim();
565    if !err.is_empty() {
566        return truncate_chars(err, MAX_SUMMARY_CHARS);
567    }
568    match exit_code {
569        Some(0) => "OpenCode finished successfully.".into(),
570        Some(code) => format!("OpenCode exited with code {code}."),
571        None => "OpenCode finished.".into(),
572    }
573}
574
575fn extract_tests_excerpt(stdout: &str) -> Option<String> {
576    let mut hits = Vec::new();
577    for line in stdout.lines() {
578        let l = line.trim();
579        let lower = l.to_ascii_lowercase();
580        if lower.contains("test result:")
581            || lower.contains("passed")
582            || lower.contains("failed")
583            || lower.contains("cargo test")
584            || lower.contains("npm test")
585            || lower.contains("pytest")
586        {
587            hits.push(l.to_string());
588        }
589        if hits.len() >= 8 {
590            break;
591        }
592    }
593    if hits.is_empty() {
594        None
595    } else {
596        Some(hits.join("\n"))
597    }
598}
599
600fn truncate_chars(s: &str, max: usize) -> String {
601    if s.chars().count() <= max {
602        return s.to_string();
603    }
604    let mut out: String = s.chars().take(max.saturating_sub(1)).collect();
605    out.push('…');
606    out
607}
608
609#[cfg(test)]
610mod tests {
611    use super::*;
612
613    #[test]
614    fn detects_missing_binary() {
615        let cfg = ExecutorConfig {
616            kind: "opencode".into(),
617            command: "opencode-not-installed-agentbridge-xyz".into(),
618            mode: ExecutorMode::Silent,
619        };
620        let exec = OpenCodeExecutor::from_config(&cfg).unwrap();
621        let err = exec.detect().unwrap_err();
622        assert!(matches!(err, ExecutorError::NotInstalled(_)));
623        assert!(err.to_string().contains("not installed"));
624    }
625
626    #[test]
627    fn rejects_unimplemented_types() {
628        let err = validate_executor_type("codex").unwrap_err();
629        assert!(matches!(err, ExecutorError::TypeNotImplemented(_)));
630        let err = validate_executor_type("not-a-real-agent").unwrap_err();
631        assert!(matches!(err, ExecutorError::TypeNotAllowed(_)));
632        validate_executor_type("opencode").unwrap();
633    }
634
635    #[test]
636    fn finds_git_on_path() {
637        assert!(find_executable("git").is_some());
638    }
639
640    #[test]
641    fn strip_reasoning_drops_think_blocks() {
642        let raw = "<think>\nsecret chain of thought\n</think>\ncreated TEST.md\n";
643        let out = strip_reasoning(raw);
644        assert!(!out.contains("chain of thought"));
645        assert!(out.contains("created TEST.md"));
646    }
647}
agentbridge/executor.rs

agentbridge/
executor.rs
