microclaw_engine/
acp_subagent.rs

1use std::collections::HashMap;
2use std::path::{Path, PathBuf};
3use std::process::Stdio;
4use std::sync::atomic::AtomicBool;
5use std::sync::Arc;
6
7use agent_client_protocol::{
8    self as acp, Agent as _, Client, ClientCapabilities, ClientSideConnection,
9    FileSystemCapabilities, InitializeRequest, ProtocolVersion, RequestPermissionOutcome,
10    SelectedPermissionOutcome,
11};
12use async_trait::async_trait;
13use serde_json::json;
14use tokio::io::AsyncReadExt;
15use tokio::sync::Mutex;
16use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
17
18use crate::config::{Config, ResolvedSubagentAcpTargetConfig};
19use crate::internal::storage::db::{call_blocking, Database};
20use crate::tools::{resolve_tool_working_dir, ToolAuthContext};
21use microclaw_core::text::floor_char_boundary;
22
23const ACP_RUNTIME_PROVIDER: &str = "acp";
24const ACP_AGENT_STDERR_LIMIT_BYTES: usize = 16 * 1024;
25const ACP_TERMINAL_OUTPUT_LIMIT_BYTES: usize = 256 * 1024;
26const ACP_CANCEL_POLL_MS: u64 = 350;
27
28pub struct AcpSubagentTaskParams {
29    pub config: Config,
30    pub db: Arc<Database>,
31    pub auth_context: ToolAuthContext,
32    pub run_id: String,
33    pub task: String,
34    pub context: String,
35    pub local_cancel: Arc<AtomicBool>,
36    pub target: ResolvedSubagentAcpTargetConfig,
37}
38
39#[derive(Default)]
40struct RetainedText {
41    text: String,
42    truncated: bool,
43}
44
45#[derive(Default)]
46struct TranscriptState {
47    message_text: String,
48    notes: Vec<String>,
49}
50
51struct TerminalSession {
52    child: Mutex<tokio::process::Child>,
53    output: Arc<Mutex<RetainedText>>,
54    exit_status: Mutex<Option<acp::TerminalExitStatus>>,
55}
56
57impl TerminalSession {
58    async fn output_response(&self) -> Result<acp::TerminalOutputResponse, acp::Error> {
59        let exit_status = self.try_refresh_exit_status().await?;
60        let output = self.output.lock().await;
61        Ok(
62            acp::TerminalOutputResponse::new(output.text.clone(), output.truncated)
63                .exit_status(exit_status),
64        )
65    }
66
67    async fn try_refresh_exit_status(&self) -> Result<Option<acp::TerminalExitStatus>, acp::Error> {
68        if let Some(existing) = self.exit_status.lock().await.clone() {
69            return Ok(Some(existing));
70        }
71        let maybe_status = {
72            let mut child = self.child.lock().await;
73            child.try_wait().map_err(internal_error)?
74        };
75        if let Some(status) = maybe_status {
76            let mapped = map_exit_status(status);
77            *self.exit_status.lock().await = Some(mapped.clone());
78            Ok(Some(mapped))
79        } else {
80            Ok(None)
81        }
82    }
83
84    async fn wait_for_exit(&self) -> Result<acp::WaitForTerminalExitResponse, acp::Error> {
85        if let Some(existing) = self.exit_status.lock().await.clone() {
86            return Ok(acp::WaitForTerminalExitResponse::new(existing));
87        }
88        let status = {
89            let mut child = self.child.lock().await;
90            child.wait().await.map_err(internal_error)?
91        };
92        let mapped = map_exit_status(status);
93        *self.exit_status.lock().await = Some(mapped.clone());
94        Ok(acp::WaitForTerminalExitResponse::new(mapped))
95    }
96
97    async fn kill(&self) -> Result<(), acp::Error> {
98        if self.exit_status.lock().await.is_some() {
99            return Ok(());
100        }
101        let status = {
102            let mut child = self.child.lock().await;
103            let _ = child.start_kill();
104            child.wait().await.map_err(internal_error)?
105        };
106        *self.exit_status.lock().await = Some(map_exit_status(status));
107        Ok(())
108    }
109}
110
111#[derive(Clone)]
112struct AcpSessionClient {
113    working_dir: PathBuf,
114    auto_approve: bool,
115    db: Arc<Database>,
116    run_id: String,
117    transcript: Arc<Mutex<TranscriptState>>,
118    terminals: Arc<Mutex<HashMap<String, Arc<TerminalSession>>>>,
119}
120
121impl AcpSessionClient {
122    fn new(working_dir: PathBuf, auto_approve: bool, db: Arc<Database>, run_id: String) -> Self {
123        Self {
124            working_dir,
125            auto_approve,
126            db,
127            run_id,
128            transcript: Arc::new(Mutex::new(TranscriptState::default())),
129            terminals: Arc::new(Mutex::new(HashMap::new())),
130        }
131    }
132
133    async fn final_text(&self) -> String {
134        let transcript = self.transcript.lock().await;
135        if transcript.message_text.trim().is_empty() {
136            transcript.notes.join("\n")
137        } else if transcript.notes.is_empty() {
138            transcript.message_text.clone()
139        } else {
140            format!(
141                "{}\n\n{}",
142                transcript.message_text,
143                transcript.notes.join("\n")
144            )
145        }
146    }
147
148    async fn kill_all_terminals(&self) {
149        let sessions = {
150            let guard = self.terminals.lock().await;
151            guard.values().cloned().collect::<Vec<_>>()
152        };
153        for session in sessions {
154            let _ = session.kill().await;
155        }
156    }
157
158    async fn append_note(&self, note: String) {
159        if note.trim().is_empty() {
160            return;
161        }
162        self.transcript.lock().await.notes.push(note);
163    }
164
165    async fn log_event(&self, event_type: &str, detail: Option<String>) {
166        let db = self.db.clone();
167        let run_id = self.run_id.clone();
168        let event_type = event_type.to_string();
169        let _ = call_blocking(db, move |db| {
170            db.append_subagent_event(&run_id, &event_type, detail.as_deref())
171        })
172        .await;
173    }
174}
175
176#[async_trait(?Send)]
177impl Client for AcpSessionClient {
178    async fn request_permission(
179        &self,
180        args: acp::RequestPermissionRequest,
181    ) -> Result<acp::RequestPermissionResponse, acp::Error> {
182        let title = args
183            .tool_call
184            .fields
185            .title
186            .clone()
187            .unwrap_or_else(|| "ACP tool call".to_string());
188        self.append_note(format!("[acp] permission requested: {title}"))
189            .await;
190        self.log_event("acp_permission_requested", Some(format!("title={title}")))
191            .await;
192
193        let preferred = if self.auto_approve {
194            args.options
195                .iter()
196                .find(|opt| {
197                    matches!(
198                        opt.kind,
199                        acp::PermissionOptionKind::AllowOnce
200                            | acp::PermissionOptionKind::AllowAlways
201                    )
202                })
203                .or_else(|| args.options.first())
204        } else {
205            args.options.iter().find(|opt| {
206                matches!(
207                    opt.kind,
208                    acp::PermissionOptionKind::RejectOnce | acp::PermissionOptionKind::RejectAlways
209                )
210            })
211        };
212
213        let Some(selected) = preferred else {
214            return Ok(acp::RequestPermissionResponse::new(
215                RequestPermissionOutcome::Cancelled,
216            ));
217        };
218        self.log_event(
219            "acp_permission_selected",
220            Some(format!("option_id={}", selected.option_id.0)),
221        )
222        .await;
223
224        Ok(acp::RequestPermissionResponse::new(
225            RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new(
226                selected.option_id.clone(),
227            )),
228        ))
229    }
230
231    async fn session_notification(&self, args: acp::SessionNotification) -> Result<(), acp::Error> {
232        match args.update {
233            acp::SessionUpdate::AgentMessageChunk(chunk) => {
234                if let acp::ContentBlock::Text(text) = chunk.content {
235                    self.transcript
236                        .lock()
237                        .await
238                        .message_text
239                        .push_str(&text.text);
240                }
241            }
242            acp::SessionUpdate::ToolCall(tool_call) => {
243                let title = if tool_call.title.trim().is_empty() {
244                    format!("{:?}", tool_call.kind)
245                } else {
246                    tool_call.title
247                };
248                self.append_note(format!("[acp] tool call: {title}")).await;
249                self.log_event("acp_tool_call", Some(format!("title={title}")))
250                    .await;
251            }
252            acp::SessionUpdate::ToolCallUpdate(update) => {
253                let title = update
254                    .fields
255                    .title
256                    .clone()
257                    .unwrap_or_else(|| update.tool_call_id.0.to_string());
258                if let Some(status) = update.fields.status {
259                    self.append_note(format!("[acp] tool update: {title} ({status:?})"))
260                        .await;
261                    self.log_event(
262                        "acp_tool_update",
263                        Some(format!("title={title} status={status:?}")),
264                    )
265                    .await;
266                }
267            }
268            acp::SessionUpdate::Plan(plan) if !plan.entries.is_empty() => {
269                let items = plan
270                    .entries
271                    .iter()
272                    .map(|entry| entry.content.as_str())
273                    .collect::<Vec<_>>()
274                    .join(" | ");
275                self.append_note(format!("[acp] plan: {items}")).await;
276                self.log_event("acp_plan", Some(items)).await;
277            }
278            _ => {}
279        }
280        Ok(())
281    }
282
283    async fn write_text_file(
284        &self,
285        args: acp::WriteTextFileRequest,
286    ) -> Result<acp::WriteTextFileResponse, acp::Error> {
287        let path = resolve_client_path(&self.working_dir, &args.path)?;
288        if let Some(parent) = path.parent() {
289            tokio::fs::create_dir_all(parent)
290                .await
291                .map_err(internal_error)?;
292        }
293        tokio::fs::write(path, args.content)
294            .await
295            .map_err(internal_error)?;
296        self.log_event(
297            "acp_write_text_file",
298            Some(format!("path={}", args.path.display())),
299        )
300        .await;
301        Ok(acp::WriteTextFileResponse::new())
302    }
303
304    async fn read_text_file(
305        &self,
306        args: acp::ReadTextFileRequest,
307    ) -> Result<acp::ReadTextFileResponse, acp::Error> {
308        let path = resolve_client_path(&self.working_dir, &args.path)?;
309        let content = tokio::fs::read_to_string(path)
310            .await
311            .map_err(internal_error)?;
312        let selected = slice_lines(&content, args.line, args.limit);
313        self.log_event(
314            "acp_read_text_file",
315            Some(format!("path={}", args.path.display())),
316        )
317        .await;
318        Ok(acp::ReadTextFileResponse::new(selected))
319    }
320
321    async fn create_terminal(
322        &self,
323        args: acp::CreateTerminalRequest,
324    ) -> Result<acp::CreateTerminalResponse, acp::Error> {
325        let cwd = match args.cwd {
326            Some(path) => resolve_client_path(&self.working_dir, &path)?,
327            None => self.working_dir.clone(),
328        };
329        let terminal_id = format!("acp-term-{}", uuid::Uuid::new_v4());
330        let output_limit = args
331            .output_byte_limit
332            .map(|value| value as usize)
333            .unwrap_or(ACP_TERMINAL_OUTPUT_LIMIT_BYTES)
334            .clamp(1024, ACP_TERMINAL_OUTPUT_LIMIT_BYTES);
335        let mut cmd = tokio::process::Command::new(&args.command);
336        cmd.args(&args.args)
337            .current_dir(&cwd)
338            .stdin(Stdio::null())
339            .stdout(Stdio::piped())
340            .stderr(Stdio::piped())
341            .kill_on_drop(true);
342        for env in args.env {
343            cmd.env(env.name, env.value);
344        }
345        let mut child = cmd.spawn().map_err(internal_error)?;
346        let stdout = child.stdout.take();
347        let stderr = child.stderr.take();
348        let session = Arc::new(TerminalSession {
349            child: Mutex::new(child),
350            output: Arc::new(Mutex::new(RetainedText::default())),
351            exit_status: Mutex::new(None),
352        });
353
354        if let Some(stdout) = stdout {
355            tokio::spawn(read_pipe_into_output(
356                stdout,
357                session.output.clone(),
358                output_limit,
359            ));
360        }
361        if let Some(stderr) = stderr {
362            tokio::spawn(read_pipe_into_output(
363                stderr,
364                session.output.clone(),
365                output_limit,
366            ));
367        }
368
369        self.terminals
370            .lock()
371            .await
372            .insert(terminal_id.clone(), session);
373        self.log_event(
374            "acp_create_terminal",
375            Some(format!("command={} cwd={}", args.command, cwd.display())),
376        )
377        .await;
378        Ok(acp::CreateTerminalResponse::new(terminal_id))
379    }
380
381    async fn terminal_output(
382        &self,
383        args: acp::TerminalOutputRequest,
384    ) -> Result<acp::TerminalOutputResponse, acp::Error> {
385        let session = self
386            .terminals
387            .lock()
388            .await
389            .get(args.terminal_id.0.as_ref())
390            .cloned()
391            .ok_or_else(acp::Error::invalid_params)?;
392        session.output_response().await
393    }
394
395    async fn release_terminal(
396        &self,
397        args: acp::ReleaseTerminalRequest,
398    ) -> Result<acp::ReleaseTerminalResponse, acp::Error> {
399        if let Some(session) = self
400            .terminals
401            .lock()
402            .await
403            .remove(args.terminal_id.0.as_ref())
404        {
405            let _ = session.kill().await;
406        }
407        self.log_event(
408            "acp_release_terminal",
409            Some(format!("terminal_id={}", args.terminal_id.0)),
410        )
411        .await;
412        Ok(acp::ReleaseTerminalResponse::new())
413    }
414
415    async fn wait_for_terminal_exit(
416        &self,
417        args: acp::WaitForTerminalExitRequest,
418    ) -> Result<acp::WaitForTerminalExitResponse, acp::Error> {
419        let session = self
420            .terminals
421            .lock()
422            .await
423            .get(args.terminal_id.0.as_ref())
424            .cloned()
425            .ok_or_else(acp::Error::invalid_params)?;
426        let response = session.wait_for_exit().await?;
427        self.log_event(
428            "acp_wait_terminal_exit",
429            Some(format!("terminal_id={}", args.terminal_id.0)),
430        )
431        .await;
432        Ok(response)
433    }
434
435    async fn kill_terminal(
436        &self,
437        args: acp::KillTerminalRequest,
438    ) -> Result<acp::KillTerminalResponse, acp::Error> {
439        let session = self
440            .terminals
441            .lock()
442            .await
443            .get(args.terminal_id.0.as_ref())
444            .cloned()
445            .ok_or_else(acp::Error::invalid_params)?;
446        session.kill().await?;
447        self.log_event(
448            "acp_kill_terminal",
449            Some(format!("terminal_id={}", args.terminal_id.0)),
450        )
451        .await;
452        Ok(acp::KillTerminalResponse::new())
453    }
454}
455
456pub async fn run_acp_subagent_task(
457    params: AcpSubagentTaskParams,
458) -> Result<(String, String, i64, i64), String> {
459    tokio::task::spawn_blocking(move || {
460        let runtime = tokio::runtime::Builder::new_current_thread()
461            .enable_all()
462            .build()
463            .map_err(|e| format!("Failed creating ACP runtime: {e}"))?;
464        let local = tokio::task::LocalSet::new();
465        runtime.block_on(local.run_until(run_acp_subagent_task_inner(params)))
466    })
467    .await
468    .map_err(|e| format!("ACP runtime join failed: {e}"))?
469}
470
471async fn run_acp_subagent_task_inner(
472    params: AcpSubagentTaskParams,
473) -> Result<(String, String, i64, i64), String> {
474    let working_dir = resolve_session_working_dir(&params.config, &params.auth_context)
475        .map_err(|e| format!("Failed resolving ACP working directory: {e}"))?;
476    tokio::fs::create_dir_all(&working_dir)
477        .await
478        .map_err(|e| format!("Failed creating ACP working directory: {e}"))?;
479
480    let command = params.target.command.clone();
481    if command.trim().is_empty() {
482        return Err("ACP runtime target command is empty".into());
483    }
484
485    let mut child = tokio::process::Command::new(&command);
486    child
487        .args(&params.target.args)
488        .current_dir(&working_dir)
489        .stdin(Stdio::piped())
490        .stdout(Stdio::piped())
491        .stderr(Stdio::piped())
492        .kill_on_drop(true)
493        .env("MICROCLAW_SUBAGENT_RUN_ID", &params.run_id)
494        .env(
495            "MICROCLAW_SUBAGENT_CHAT_ID",
496            params.auth_context.caller_chat_id.to_string(),
497        )
498        .env(
499            "MICROCLAW_SUBAGENT_WORKDIR",
500            working_dir.display().to_string(),
501        );
502    if let Some(target_name) = params.target.name.as_deref() {
503        child.env("MICROCLAW_SUBAGENT_RUNTIME_TARGET", target_name);
504    }
505    for (key, value) in &params.target.env {
506        child.env(key, value);
507    }
508
509    let mut child = child
510        .spawn()
511        .map_err(|e| format!("Failed spawning ACP agent command '{command}': {e}"))?;
512    let stdin = child
513        .stdin
514        .take()
515        .ok_or_else(|| "ACP agent stdin pipe unavailable".to_string())?;
516    let stdout = child
517        .stdout
518        .take()
519        .ok_or_else(|| "ACP agent stdout pipe unavailable".to_string())?;
520    let stderr = child.stderr.take();
521    let stderr_capture = Arc::new(Mutex::new(RetainedText::default()));
522    if let Some(stderr) = stderr {
523        tokio::spawn(read_pipe_into_output(
524            stderr,
525            stderr_capture.clone(),
526            ACP_AGENT_STDERR_LIMIT_BYTES,
527        ));
528    }
529
530    let client = Arc::new(AcpSessionClient::new(
531        working_dir.clone(),
532        params.target.auto_approve,
533        params.db.clone(),
534        params.run_id.clone(),
535    ));
536    let prompt_text = if params.context.is_empty() {
537        params.task.clone()
538    } else {
539        format!("Context: {}\n\nTask: {}", params.context, params.task)
540    };
541
542    let db = params.db.clone();
543    let run_id = params.run_id.clone();
544    let local_cancel = params.local_cancel.clone();
545    let (conn, handle_io) = ClientSideConnection::new(
546        client.clone(),
547        stdin.compat_write(),
548        stdout.compat(),
549        |fut| {
550            tokio::task::spawn_local(fut);
551        },
552    );
553    let io_handle = tokio::task::spawn_local(handle_io);
554
555    conn.initialize(
556        InitializeRequest::new(ProtocolVersion::V1)
557            .client_info(
558                acp::Implementation::new("microclaw-acp-client", env!("CARGO_PKG_VERSION"))
559                    .title("MicroClaw ACP Client"),
560            )
561            .client_capabilities(
562                ClientCapabilities::new()
563                    .fs(FileSystemCapabilities::new()
564                        .read_text_file(true)
565                        .write_text_file(true))
566                    .terminal(true),
567            ),
568    )
569    .await
570    .map_err(|e| format!("ACP initialize failed: {e}"))?;
571
572    let session = conn
573        .new_session(acp::NewSessionRequest::new(working_dir.clone()))
574        .await
575        .map_err(|e| format!("ACP new_session failed: {e}"))?;
576    let session_id = session.session_id.clone();
577    let prompt_future = conn.prompt(acp::PromptRequest::new(
578        session_id.clone(),
579        vec![prompt_text.into()],
580    ));
581    tokio::pin!(prompt_future);
582
583    loop {
584        tokio::select! {
585            prompt_result = &mut prompt_future => {
586                prompt_result.map_err(|e| format!("ACP prompt failed: {e}"))?;
587                break;
588            }
589            _ = tokio::time::sleep(std::time::Duration::from_millis(ACP_CANCEL_POLL_MS)) => {
590                if crate::tools::subagents::is_cancelled(db.clone(), &run_id, &local_cancel).await? {
591                    let _ = conn.cancel(acp::CancelNotification::new(session_id.clone())).await;
592                    client.kill_all_terminals().await;
593                    let _ = terminate_child_process(&mut child).await;
594                    io_handle.abort();
595                    return Err("cancelled".to_string());
596                }
597            }
598        }
599    }
600
601    client.kill_all_terminals().await;
602    let _ = terminate_child_process(&mut child).await;
603    io_handle.abort();
604
605    let source = client.final_text().await;
606    let stderr_text = stderr_capture.lock().await.text.clone();
607    if source.trim().is_empty() && !stderr_text.trim().is_empty() {
608        return Err(format!(
609            "ACP agent produced no response. stderr:\n{stderr_text}"
610        ));
611    }
612    let source = if source.trim().is_empty() {
613        "(sub-agent produced no output)".to_string()
614    } else {
615        source
616    };
617    let (final_text, artifact_json) =
618        crate::tools::subagents::normalize_subagent_artifact_payload(&source);
619    Ok((final_text, artifact_json, 0, 0))
620}
621
622pub fn acp_runtime_model(target: &ResolvedSubagentAcpTargetConfig) -> String {
623    target.model_label()
624}
625
626pub fn acp_runtime_provider(target_name: Option<&str>) -> String {
627    match target_name.map(str::trim).filter(|name| !name.is_empty()) {
628        Some(name) => format!("{ACP_RUNTIME_PROVIDER}:{name}"),
629        None => ACP_RUNTIME_PROVIDER.to_string(),
630    }
631}
632
633pub fn acp_runtime_target_from_provider(provider: &str) -> Option<String> {
634    provider
635        .strip_prefix(&format!("{ACP_RUNTIME_PROVIDER}:"))
636        .map(ToOwned::to_owned)
637        .filter(|name| !name.trim().is_empty())
638}
639
640fn resolve_session_working_dir(
641    config: &Config,
642    auth_context: &ToolAuthContext,
643) -> Result<PathBuf, std::io::Error> {
644    let input = json!({
645        "__microclaw_auth": {
646            "caller_channel": auth_context.caller_channel,
647            "caller_chat_id": auth_context.caller_chat_id,
648            "control_chat_ids": auth_context.control_chat_ids,
649            "env_files": auth_context.env_files,
650        }
651    });
652    let path = resolve_tool_working_dir(
653        Path::new(&config.working_dir),
654        config.working_dir_isolation,
655        &input,
656    );
657    std::fs::create_dir_all(&path)?;
658    Ok(path)
659}
660
661fn resolve_client_path(root: &Path, requested: &Path) -> Result<PathBuf, acp::Error> {
662    let raw_root = normalize_path(root);
663    let root = std::fs::canonicalize(root).map_err(internal_error)?;
664    let requested = normalize_path(requested);
665    let candidate = if requested.is_absolute() {
666        if let Ok(relative) = requested.strip_prefix(&root) {
667            root.join(relative)
668        } else if let Ok(relative) = requested.strip_prefix(&raw_root) {
669            root.join(relative)
670        } else {
671            return Err(acp::Error::invalid_params());
672        }
673    } else {
674        root.join(requested)
675    };
676    let resolved = std::fs::canonicalize(&candidate).unwrap_or_else(|_| normalize_path(&candidate));
677    if !resolved.starts_with(&root) {
678        return Err(acp::Error::invalid_params());
679    }
680    let path_str = resolved.to_string_lossy().to_string();
681    if let Err(msg) = crate::internal::tool_runtime::path_guard::check_path(&path_str) {
682        let mut err = acp::Error::invalid_request();
683        err.message = msg;
684        return Err(err);
685    }
686    Ok(resolved)
687}
688
689fn normalize_path(path: &Path) -> PathBuf {
690    let mut normalized = PathBuf::new();
691    for component in path.components() {
692        match component {
693            std::path::Component::Prefix(prefix) => normalized.push(prefix.as_os_str()),
694            std::path::Component::RootDir => normalized.push(Path::new("/")),
695            std::path::Component::CurDir => {}
696            std::path::Component::ParentDir => {
697                normalized.pop();
698            }
699            std::path::Component::Normal(part) => normalized.push(part),
700        }
701    }
702    normalized
703}
704
705fn slice_lines(content: &str, line: Option<u32>, limit: Option<u32>) -> String {
706    let lines: Vec<&str> = content.lines().collect();
707    let start = line.unwrap_or(1).saturating_sub(1) as usize;
708    let max = limit.unwrap_or(lines.len() as u32).max(1) as usize;
709    let end = (start + max).min(lines.len());
710    lines[start.min(lines.len())..end].join("\n")
711}
712
713async fn read_pipe_into_output<R>(mut reader: R, output: Arc<Mutex<RetainedText>>, limit: usize)
714where
715    R: tokio::io::AsyncRead + Unpin,
716{
717    let mut buf = [0_u8; 4096];
718    loop {
719        let read = match reader.read(&mut buf).await {
720            Ok(0) => break,
721            Ok(n) => n,
722            Err(_) => break,
723        };
724        let chunk = String::from_utf8_lossy(&buf[..read]).into_owned();
725        let mut retained = output.lock().await;
726        retained.text.push_str(&chunk);
727        truncate_retained_text(&mut retained, limit);
728    }
729}
730
731fn truncate_retained_text(output: &mut RetainedText, limit: usize) {
732    if output.text.len() <= limit {
733        return;
734    }
735    output.truncated = true;
736    let overflow = output.text.len().saturating_sub(limit);
737    let boundary = floor_char_boundary(&output.text, overflow);
738    output.text.replace_range(..boundary, "");
739}
740
741fn map_exit_status(status: std::process::ExitStatus) -> acp::TerminalExitStatus {
742    #[cfg(unix)]
743    let mapped = {
744        use std::os::unix::process::ExitStatusExt;
745        acp::TerminalExitStatus::new().signal(status.signal().map(|signal| signal.to_string()))
746    };
747    #[cfg(not(unix))]
748    let mapped = acp::TerminalExitStatus::new();
749    mapped.exit_code(status.code().map(|code| code as u32))
750}
751
752async fn terminate_child_process(child: &mut tokio::process::Child) -> Result<(), std::io::Error> {
753    match child.try_wait()? {
754        Some(_) => Ok(()),
755        None => {
756            let _ = child.start_kill();
757            let _ = child.wait().await?;
758            Ok(())
759        }
760    }
761}
762
763fn internal_error(err: impl std::fmt::Display) -> acp::Error {
764    let mut error = acp::Error::internal_error();
765    error.message = err.to_string();
766    error
767}
768
769#[cfg(test)]
770mod tests {
771    use super::*;
772
773    fn test_db() -> Arc<Database> {
774        let dir = std::env::temp_dir().join(format!(
775            "microclaw_acp_subagent_test_{}",
776            uuid::Uuid::new_v4()
777        ));
778        std::fs::create_dir_all(&dir).unwrap();
779        Arc::new(Database::new(dir.to_str().unwrap()).unwrap())
780    }
781
782    #[cfg(unix)]
783    fn test_terminal_command() -> (&'static str, Vec<String>) {
784        ("/bin/sh", vec!["-c".into(), "printf 'hello'".into()])
785    }
786
787    #[cfg(windows)]
788    fn test_terminal_command() -> (&'static str, Vec<String>) {
789        ("cmd.exe", vec!["/C".into(), "echo hello".into()])
790    }
791
792    #[tokio::test]
793    async fn test_acp_client_read_write_stays_inside_workdir() {
794        let root =
795            std::env::temp_dir().join(format!("microclaw_acp_client_{}", uuid::Uuid::new_v4()));
796        tokio::fs::create_dir_all(&root).await.unwrap();
797        let client = AcpSessionClient::new(root.clone(), true, test_db(), "run-1".into());
798        let session_id = acp::SessionId::new("test-session");
799
800        client
801            .write_text_file(acp::WriteTextFileRequest::new(
802                session_id.clone(),
803                root.join("notes/todo.txt"),
804                "hello",
805            ))
806            .await
807            .unwrap();
808        let read = client
809            .read_text_file(acp::ReadTextFileRequest::new(
810                session_id.clone(),
811                root.join("notes/todo.txt"),
812            ))
813            .await
814            .unwrap();
815        assert_eq!(read.content, "hello");
816
817        let err = client
818            .read_text_file(acp::ReadTextFileRequest::new(
819                session_id,
820                PathBuf::from("/etc/hosts"),
821            ))
822            .await
823            .unwrap_err();
824        assert_eq!(err.code, acp::Error::invalid_params().code);
825        let _ = tokio::fs::remove_dir_all(root).await;
826    }
827
828    #[tokio::test]
829    async fn test_acp_client_terminal_lifecycle() {
830        let root =
831            std::env::temp_dir().join(format!("microclaw_acp_term_{}", uuid::Uuid::new_v4()));
832        tokio::fs::create_dir_all(&root).await.unwrap();
833        let client = AcpSessionClient::new(root.clone(), true, test_db(), "run-2".into());
834        let session_id = acp::SessionId::new("test-session");
835        let (command, args) = test_terminal_command();
836
837        let created = client
838            .create_terminal(
839                acp::CreateTerminalRequest::new(session_id.clone(), command).args(args),
840            )
841            .await
842            .unwrap();
843        let waited = client
844            .wait_for_terminal_exit(acp::WaitForTerminalExitRequest::new(
845                session_id.clone(),
846                created.terminal_id.clone(),
847            ))
848            .await
849            .unwrap();
850        assert_eq!(waited.exit_status.exit_code, Some(0));
851
852        let output = client
853            .terminal_output(acp::TerminalOutputRequest::new(
854                session_id.clone(),
855                created.terminal_id.clone(),
856            ))
857            .await
858            .unwrap();
859        assert!(output.output.contains("hello"));
860
861        client
862            .release_terminal(acp::ReleaseTerminalRequest::new(
863                session_id,
864                created.terminal_id,
865            ))
866            .await
867            .unwrap();
868        let _ = tokio::fs::remove_dir_all(root).await;
869    }
870
871    #[tokio::test]
872    async fn test_permission_auto_approve_prefers_allow() {
873        let client = AcpSessionClient::new(std::env::temp_dir(), true, test_db(), "run-3".into());
874        let session_id = acp::SessionId::new("test-session");
875        let response = client
876            .request_permission(acp::RequestPermissionRequest::new(
877                session_id,
878                acp::ToolCallUpdate::new(
879                    "tool-1",
880                    acp::ToolCallUpdateFields::new().title("Run terminal".to_string()),
881                ),
882                vec![
883                    acp::PermissionOption::new(
884                        "reject",
885                        "Reject",
886                        acp::PermissionOptionKind::RejectOnce,
887                    ),
888                    acp::PermissionOption::new(
889                        "allow",
890                        "Allow",
891                        acp::PermissionOptionKind::AllowOnce,
892                    ),
893                ],
894            ))
895            .await
896            .unwrap();
897        let RequestPermissionOutcome::Selected(selected) = response.outcome else {
898            panic!("expected selected permission outcome");
899        };
900        assert_eq!(selected.option_id.0.as_ref(), "allow");
901    }
902
903    #[test]
904    fn test_acp_runtime_provider_round_trips_target_name() {
905        let provider = acp_runtime_provider(Some("worker"));
906        assert_eq!(provider, "acp:worker");
907        assert_eq!(
908            acp_runtime_target_from_provider(&provider).as_deref(),
909            Some("worker")
910        );
911        assert_eq!(acp_runtime_target_from_provider("acp"), None);
912    }
913}
microclaw_engine/acp_subagent.rs

microclaw_engine/
acp_subagent.rs
