agentmux_srv\backend\blockcontroller/
acp.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! AcpController: manages agent CLIs that speak the Agent Client Protocol (ACP).
5//!
6//! ACP is a JSON-RPC 2.0 protocol over stdin/stdout — the "LSP for AI agents."
7//! Instead of custom per-provider output parsing, ACP provides a standardized
8//! protocol for session management, prompting, and streaming events.
9//!
10//! Lifecycle:
11//!   1. Spawn the ACP agent process (e.g., `gemini --acp`, `openclaw acp`)
12//!   2. Send `initialize` request, receive capabilities
13//!   3. Send `initialized` notification
14//!   4. Create a session via `session/create`
15//!   5. For each user turn: send `session/prompt`, stream `session/update` notifications
16//!   6. On close: send `shutdown` + `exit`
17//!
18//! I/O model (similar to PersistentSubprocessController):
19//!   - stdin_writer: sends JSON-RPC requests/notifications to agent
20//!   - stdout_reader: reads JSON-RPC responses/notifications, persists + broadcasts
21//!   - process_waiter: monitors process lifecycle
22//!
23//! See: https://github.com/agentclientprotocol/agent-client-protocol
24
25use std::collections::HashMap;
26use std::sync::atomic::{AtomicU64, Ordering};
27use std::sync::{Arc, Mutex};
28
29use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
30use tokio::sync::mpsc;
31
32use super::{
33    BlockControllerRuntimeStatus, BlockInputUnion, Controller, STATUS_DONE, STATUS_INIT,
34    STATUS_RUNNING,
35};
36use super::core;
37use super::health::HealthMonitor;
38use crate::backend::eventbus::EventBus;
39use crate::backend::storage::filestore::FileStore;
40use crate::backend::storage::store::Store;
41use crate::backend::wps;
42
43/// WPS file subject name for ACP output.
44pub const ACP_OUTPUT_SUBJECT: &str = "output";
45
46pub const BLOCK_CONTROLLER_ACP: &str = "acp";
47
48/// Inner state protected by mutex.
49struct AcpInner {
50    proc_status: String,
51    proc_exit_code: i32,
52    status_version: i32,
53    session_id: Option<String>,
54    current_pid: Option<u32>,
55    stdin_tx: Option<mpsc::Sender<String>>,
56    kill_tx: Option<tokio::sync::oneshot::Sender<bool>>,
57    /// First user prompt, deferred until session/create completes.
58    pending_prompt: Option<String>,
59}
60
61/// AcpController manages an ACP-speaking agent process.
62pub struct AcpController {
63    #[allow(dead_code)]
64    tab_id: String,
65    block_id: String,
66    inner: Arc<Mutex<AcpInner>>,
67    broker: Option<Arc<wps::Broker>>,
68    event_bus: Option<Arc<EventBus>>,
69    wstore: Option<Arc<Store>>,
70    filestore: Option<Arc<FileStore>>,
71    health_monitor: Arc<HealthMonitor>,
72    /// Monotonically increasing JSON-RPC request ID.
73    next_rpc_id: Arc<AtomicU64>,
74}
75
76impl AcpController {
77    pub fn new(
78        tab_id: String,
79        block_id: String,
80        broker: Option<Arc<wps::Broker>>,
81        event_bus: Option<Arc<EventBus>>,
82        wstore: Option<Arc<Store>>,
83        filestore: Option<Arc<FileStore>>,
84    ) -> Self {
85        let health_monitor = Arc::new(HealthMonitor::new(
86            block_id.clone(),
87            broker.clone(),
88        ));
89        Self {
90            tab_id,
91            block_id,
92            inner: Arc::new(Mutex::new(AcpInner {
93                proc_status: STATUS_INIT.to_string(),
94                proc_exit_code: 0,
95                status_version: 0,
96                session_id: None,
97                current_pid: None,
98                stdin_tx: None,
99                kill_tx: None,
100                pending_prompt: None,
101            })),
102            broker,
103            event_bus,
104            wstore,
105            filestore,
106            health_monitor,
107            next_rpc_id: Arc::new(AtomicU64::new(1)),
108        }
109    }
110
111    fn next_id(&self) -> u64 {
112        self.next_rpc_id.fetch_add(1, Ordering::Relaxed)
113    }
114
115    fn set_status(inner: &mut AcpInner, status: &str) {
116        inner.proc_status = status.to_string();
117        inner.status_version += 1;
118    }
119
120    fn get_status_snapshot(&self) -> BlockControllerRuntimeStatus {
121        let inner = self.inner.lock().unwrap();
122        BlockControllerRuntimeStatus {
123            blockid: self.block_id.clone(),
124            version: inner.status_version,
125            shellprocstatus: inner.proc_status.clone(),
126            shellprocconnname: "local".to_string(),
127            shellprocexitcode: inner.proc_exit_code,
128            spawn_ts_ms: None,
129            is_agent_pane: true,
130        }
131    }
132
133    fn publish_status(&self) {
134        if let Some(ref broker) = self.broker {
135            let status = self.get_status_snapshot();
136            super::publish_controller_status(broker, &status);
137        }
138    }
139
140    fn is_running(&self) -> bool {
141        let inner = self.inner.lock().unwrap();
142        inner.stdin_tx.is_some()
143    }
144
145    /// Build a JSON-RPC 2.0 request.
146    fn make_request(&self, method: &str, params: serde_json::Value) -> String {
147        let id = self.next_id();
148        serde_json::json!({
149            "jsonrpc": "2.0",
150            "id": id,
151            "method": method,
152            "params": params,
153        }).to_string()
154    }
155
156    /// Build a JSON-RPC 2.0 notification (no id field).
157    fn make_notification(&self, method: &str, params: serde_json::Value) -> String {
158        serde_json::json!({
159            "jsonrpc": "2.0",
160            "method": method,
161            "params": params,
162        }).to_string()
163    }
164
165    /// Send a user message via ACP session/prompt.
166    /// If the process isn't spawned yet, spawns it first (the first prompt is
167    /// deferred until the session is established — see `pending_prompt`).
168    pub fn send_message(&self, message: String, cli_command: String, cli_args: Vec<String>, working_dir: String, env_vars: HashMap<String, String>) -> Result<(), String> {
169        if !self.is_running() {
170            // First turn: spawn and stash the prompt — the stdout reader will
171            // send it once the session/create response arrives.
172            {
173                let mut inner = self.inner.lock().unwrap();
174                inner.pending_prompt = Some(message.clone());
175            }
176            self.health_monitor.set_active_turn(true);
177            return self.spawn_process(cli_command, cli_args, working_dir, env_vars);
178        }
179
180        // Subsequent turns: session_id is already populated.
181        let session_id = {
182            let inner = self.inner.lock().unwrap();
183            inner.session_id.clone().unwrap_or_default()
184        };
185
186        let req = self.make_request("session/prompt", serde_json::json!({
187            "sessionId": session_id,
188            "prompt": {
189                "type": "text",
190                "text": message,
191            }
192        }));
193
194        self.health_monitor.set_active_turn(true);
195
196        let inner = self.inner.lock().unwrap();
197        let tx = inner.stdin_tx.as_ref()
198            .ok_or("ACP process not running after spawn")?;
199        tx.try_send(req)
200            .map_err(|e| format!("ACP stdin send failed: {e}"))
201    }
202
203    /// Spawn the ACP agent process and perform the initialize handshake.
204    fn spawn_process(&self, cli_command: String, cli_args: Vec<String>, working_dir: String, env_vars: HashMap<String, String>) -> Result<(), String> {
205        let mut cmd = crate::server::cli_handlers::make_cli_cmd(&cli_command);
206        cmd.args(&cli_args);
207
208        core::apply_working_dir(&mut cmd, &self.block_id, &working_dir, &env_vars);
209        // On Windows: suppress console-window allocation. Without CREATE_NO_WINDOW,
210        // node.exe spawned from a windowless sidecar may try to create/attach to a
211        // console, causing stdout to go to that console rather than the pipe.
212        #[cfg(windows)]
213        {
214            const CREATE_NO_WINDOW: u32 = 0x0800_0000;
215            cmd.creation_flags(CREATE_NO_WINDOW);
216        }
217
218        cmd.stdin(std::process::Stdio::piped());
219        cmd.stdout(std::process::Stdio::piped());
220        cmd.stderr(std::process::Stdio::piped());
221
222        let mut child = cmd.spawn().map_err(|e| {
223            tracing::error!(block_id = %self.block_id, error = %e, "ACP process spawn failed");
224            format!("failed to spawn ACP process: {e}")
225        })?;
226
227        let pid = child.id().unwrap_or(0);
228
229        tracing::info!(
230            block_id = %self.block_id,
231            pid = pid,
232            cmd = %cli_command,
233            args = ?cli_args,
234            "ACP agent process spawned"
235        );
236
237        let (kill_tx, kill_rx) = tokio::sync::oneshot::channel::<bool>();
238        let stdin = child.stdin.take()
239            .ok_or_else(|| format!("[acp] stdin not captured for block {}", self.block_id))?;
240        let stdout = child.stdout.take()
241            .ok_or_else(|| format!("[acp] stdout not captured for block {}", self.block_id))?;
242        let stderr = child.stderr.take();
243
244        // Drain stderr with explicit error/EOF logging
245        if let Some(stderr_pipe) = stderr {
246            let block_id_stderr = self.block_id.clone();
247            tokio::spawn(async move {
248                let mut reader = BufReader::new(stderr_pipe).lines();
249                loop {
250                    match reader.next_line().await {
251                        Err(e) => {
252                            tracing::warn!(block_id = %block_id_stderr, error = %e, "ACP stderr read error");
253                            break;
254                        }
255                        Ok(None) => break,
256                        Ok(Some(line)) => {
257                            if !line.trim().is_empty() {
258                                tracing::warn!(
259                                    block_id = %block_id_stderr,
260                                    line = %line,
261                                    "ACP agent stderr"
262                                );
263                            }
264                        }
265                    }
266                }
267            });
268        }
269
270        // Stdin writer channel
271        let (msg_tx, mut msg_rx) = mpsc::channel::<String>(32);
272
273        {
274            let mut inner = self.inner.lock().unwrap();
275            inner.current_pid = Some(pid);
276            inner.kill_tx = Some(kill_tx);
277            inner.stdin_tx = Some(msg_tx.clone());
278            Self::set_status(&mut inner, STATUS_RUNNING);
279        }
280        self.publish_status();
281
282        // Spawn stdin writer task
283        let block_id_stdin = self.block_id.clone();
284        tokio::spawn(async move {
285            let mut stdin = tokio::io::BufWriter::new(stdin);
286            while let Some(line) = msg_rx.recv().await {
287                if let Err(e) = stdin.write_all(line.as_bytes()).await {
288                    tracing::error!(block_id = %block_id_stdin, error = %e, "ACP stdin write error");
289                    break;
290                }
291                if let Err(e) = stdin.write_all(b"\n").await {
292                    tracing::error!(block_id = %block_id_stdin, error = %e, "ACP stdin newline error");
293                    break;
294                }
295                if let Err(e) = stdin.flush().await {
296                    tracing::error!(block_id = %block_id_stdin, error = %e, "ACP stdin flush error");
297                    break;
298                }
299            }
300        });
301
302        // Spawn stdout reader task — reads NDJSON lines and broadcasts via WPS
303        let block_id_stdout = self.block_id.clone();
304        let broker_clone = self.broker.clone();
305        let filestore_clone = self.filestore.clone();
306        let inner_clone = self.inner.clone();
307        let health_clone = self.health_monitor.clone();
308        let rpc_id_clone = self.next_rpc_id.clone();
309        let wstore_clone = self.wstore.clone();
310        let event_bus_clone = self.event_bus.clone();
311        // Resolve the agent's GLOBAL transcript zone once (see persistent.rs).
312        let global_output_zone =
313            super::shell::resolve_global_output_zone(&self.wstore, &self.block_id);
314        tokio::spawn(async move {
315            let mut reader = BufReader::new(stdout).lines();
316            tracing::info!(block_id = %block_id_stdout, "ACP stdout_reader started");
317
318            loop {
319                let line = match reader.next_line().await {
320                    Err(e) => {
321                        tracing::warn!(block_id = %block_id_stdout, error = %e, "ACP stdout read error");
322                        break;
323                    }
324                    Ok(None) => {
325                        tracing::info!(block_id = %block_id_stdout, "ACP stdout EOF");
326                        break;
327                    }
328                    Ok(Some(l)) => l,
329                };
330                if line.is_empty() {
331                    continue;
332                }
333
334                // Parse as JSON to check for session/update notifications
335                if let Ok(json) = serde_json::from_str::<serde_json::Value>(&line) {
336                    // Extract session ID from initialize result or session/create result
337                    if let Some(result) = json.get("result") {
338                        if let Some(sid) = result.get("sessionId").and_then(|v| v.as_str()) {
339                            let sid_owned = sid.to_string();
340                            {
341                                let mut inner = inner_clone.lock().unwrap();
342                                inner.session_id = Some(sid_owned.clone());
343                                // Flush pending prompt now that session is ready.
344                                if let Some(prompt) = inner.pending_prompt.take() {
345                                    let id = rpc_id_clone.fetch_add(1, Ordering::Relaxed);
346                                    let req = serde_json::json!({
347                                        "jsonrpc": "2.0",
348                                        "id": id,
349                                        "method": "session/prompt",
350                                        "params": {
351                                            "sessionId": sid,
352                                            "prompt": { "type": "text", "text": prompt },
353                                        }
354                                    }).to_string();
355                                    if let Some(ref tx) = inner.stdin_tx {
356                                        if tx.try_send(req).is_err() {
357                                            tracing::warn!(block_id = %block_id_stdout, "[acp] session/prompt send failed — channel full or closed");
358                                        }
359                                    }
360                                }
361                            }
362                            tracing::info!(
363                                block_id = %block_id_stdout,
364                                session_id = %sid_owned,
365                                "ACP session established"
366                            );
367                            // Persist to block metadata and broadcast so the frontend's
368                            // "My Agents" reattach path can read agent:sessionid from
369                            // block.meta. ACP previously captured the ID in memory only —
370                            // this mirrors the careful path from persistent.rs / subprocess.rs.
371                            core::persist_session_id(&block_id_stdout, &sid_owned, &wstore_clone, &event_bus_clone);
372                        }
373                    }
374
375                    // Reset health monitor on prompt result (turn complete)
376                    if json.get("id").is_some() && json.get("result").is_some() {
377                        // This is a response to a request (e.g., session/prompt result)
378                        if let Some(result) = json.get("result") {
379                            if result.get("stopReason").is_some() {
380                                health_clone.set_active_turn(false);
381                            }
382                        }
383                    }
384                }
385
386                // Persist + broadcast via the shared helper (same as subprocess.rs)
387                if let Some(ref broker) = broker_clone {
388                    let line_with_newline = format!("{}\n", line);
389                    super::shell::handle_append_block_file(
390                        broker,
391                        &block_id_stdout,
392                        ACP_OUTPUT_SUBJECT,
393                        line_with_newline.as_bytes(),
394                        filestore_clone.as_ref(),
395                        global_output_zone.as_deref(),
396                    );
397                }
398            }
399        });
400
401        // Spawn process waiter task
402        let block_id_wait = self.block_id.clone();
403        let inner_wait = self.inner.clone();
404        let broker_wait = self.broker.clone();
405        let health_wait = self.health_monitor.clone();
406        tokio::spawn(async move {
407            tokio::select! {
408                _ = kill_rx => {
409                    let _ = child.kill().await;
410                    tracing::info!(block_id = %block_id_wait, "ACP process killed");
411
412                    let mut inner = inner_wait.lock().unwrap();
413                    inner.stdin_tx = None;
414                    inner.current_pid = None;
415                    AcpController::set_status(&mut inner, STATUS_DONE);
416                    drop(inner);
417
418                    health_wait.set_active_turn(false);
419
420                    if let Some(ref broker) = broker_wait {
421                        let status = BlockControllerRuntimeStatus {
422                            blockid: block_id_wait.clone(),
423                            version: 0,
424                            shellprocstatus: STATUS_DONE.to_string(),
425                            shellprocconnname: "local".to_string(),
426                            shellprocexitcode: -1,
427                            spawn_ts_ms: None,
428                            is_agent_pane: true,
429                        };
430                        super::publish_controller_status(broker, &status);
431                    }
432                }
433                status = child.wait() => {
434                    let exit_code = status.map(|s| s.code().unwrap_or(-1)).unwrap_or(-1);
435                    tracing::info!(
436                        block_id = %block_id_wait,
437                        exit_code = exit_code,
438                        "ACP process exited"
439                    );
440                    let mut inner = inner_wait.lock().unwrap();
441                    inner.proc_exit_code = exit_code;
442                    inner.stdin_tx = None;
443                    AcpController::set_status(&mut inner, STATUS_DONE);
444                    drop(inner);
445
446                    health_wait.set_active_turn(false);
447
448                    if let Some(ref broker) = broker_wait {
449                        let status = BlockControllerRuntimeStatus {
450                            blockid: block_id_wait.clone(),
451                            version: 0,
452                            shellprocstatus: STATUS_DONE.to_string(),
453                            shellprocconnname: "local".to_string(),
454                            shellprocexitcode: exit_code,
455                            spawn_ts_ms: None,
456                            is_agent_pane: true,
457                        };
458                        super::publish_controller_status(broker, &status);
459                    }
460                }
461            }
462        });
463
464        // Send ACP initialize handshake
465        let init_req = self.make_request("initialize", serde_json::json!({
466            "clientInfo": {
467                "name": "AgentMux",
468                "version": env!("CARGO_PKG_VERSION"),
469            },
470            "capabilities": {
471                "tools": true,
472                "fileAccess": true,
473            },
474            "workspaceRoots": [working_dir],
475        }));
476        let init_notification = self.make_notification("initialized", serde_json::json!({}));
477        let session_req = self.make_request("session/create", serde_json::json!({
478            "cwd": working_dir,
479        }));
480
481        // Queue the handshake messages
482        let inner = self.inner.lock().unwrap();
483        if let Some(ref tx) = inner.stdin_tx {
484            for (label, msg) in [("initialize", init_req), ("initialized", init_notification), ("session/create", session_req)] {
485                if tx.try_send(msg).is_err() {
486                    tracing::error!(block_id = %self.block_id, method = label, "[acp] handshake message dropped — channel full or closed; agent will not start");
487                }
488            }
489        }
490
491        Ok(())
492    }
493}
494
495impl Controller for AcpController {
496    fn start(
497        &self,
498        block_meta: super::super::obj::MetaMapType,
499        _rt_opts: Option<serde_json::Value>,
500        _force: bool,
501    ) -> Result<(), String> {
502        // Extract spawn config from block metadata
503        let cmd = super::super::obj::meta_get_string(&block_meta, super::META_KEY_CMD, "");
504        let cwd = super::super::obj::meta_get_string(&block_meta, super::META_KEY_CMD_CWD, "");
505        let args_str = super::super::obj::meta_get_string(&block_meta, super::META_KEY_CMD_ARGS, "[]");
506        let env_str = super::super::obj::meta_get_string(&block_meta, super::META_KEY_CMD_ENV, "{}");
507
508        if cmd.is_empty() {
509            return Err("ACP controller: no cmd specified in block meta".to_string());
510        }
511
512        let args: Vec<String> = serde_json::from_str(&args_str).unwrap_or_default();
513        let env_vars: HashMap<String, String> = serde_json::from_str(&env_str).unwrap_or_default();
514
515        self.spawn_process(cmd, args, cwd, env_vars)
516    }
517
518    fn stop(&self, _graceful: bool, _new_status: &str) -> Result<(), String> {
519        // Send shutdown request before killing
520        {
521            let inner = self.inner.lock().unwrap();
522            if let Some(ref tx) = inner.stdin_tx {
523                let shutdown = self.make_request("shutdown", serde_json::json!({}));
524                let exit = self.make_notification("exit", serde_json::json!({}));
525                for (label, msg) in [("shutdown", shutdown), ("exit", exit)] {
526                    if tx.try_send(msg).is_err() {
527                        tracing::debug!(block_id = %self.block_id, method = label, "[acp] shutdown message dropped — process likely already gone");
528                    }
529                }
530            }
531        }
532
533        // Kill the process
534        let kill_tx = {
535            let mut inner = self.inner.lock().unwrap();
536            inner.stdin_tx = None;
537            inner.kill_tx.take()
538        };
539        if let Some(tx) = kill_tx {
540            let _ = tx.send(true);
541        }
542
543        {
544            let mut inner = self.inner.lock().unwrap();
545            Self::set_status(&mut inner, STATUS_DONE);
546        }
547        self.publish_status();
548        Ok(())
549    }
550
551    fn get_runtime_status(&self) -> BlockControllerRuntimeStatus {
552        self.get_status_snapshot()
553    }
554
555    fn send_input(&self, input: BlockInputUnion, _seq: Option<u64>) -> Result<(), String> {
556        if let Some(data) = input.input_data {
557            // Raw input from the frontend — treat as a user message.
558            // The frontend sends the user prompt as UTF-8 bytes.
559            let message = String::from_utf8_lossy(&data).to_string();
560            if message.trim().is_empty() {
561                return Ok(());
562            }
563
564            if !self.is_running() {
565                // Process not running — stash as pending so start() picks it up.
566                let mut inner = self.inner.lock().unwrap();
567                inner.pending_prompt = Some(message);
568                return Err("ACP process not running — message queued for next start()".to_string());
569            }
570
571            let session_id = {
572                let inner = self.inner.lock().unwrap();
573                inner.session_id.clone().unwrap_or_default()
574            };
575            let req = self.make_request("session/prompt", serde_json::json!({
576                "sessionId": session_id,
577                "prompt": {
578                    "type": "text",
579                    "text": message,
580                }
581            }));
582            self.health_monitor.set_active_turn(true);
583            let inner = self.inner.lock().unwrap();
584            if let Some(ref tx) = inner.stdin_tx {
585                tx.try_send(req)
586                    .map_err(|e| format!("ACP stdin send failed: {e}"))?;
587            }
588        }
589
590        if let Some(sig) = input.sig_name {
591            if sig == "SIGTERM" || sig == "SIGINT" {
592                return self.stop(true, STATUS_DONE);
593            }
594        }
595
596        Ok(())
597    }
598
599    fn controller_type(&self) -> &str {
600        BLOCK_CONTROLLER_ACP
601    }
602
603    fn block_id(&self) -> &str {
604        &self.block_id
605    }
606
607    fn as_any(&self) -> &dyn std::any::Any {
608        self
609    }
610}
agentmux_srv\backend\blockcontroller/acp.rs

agentmux_srv\backend\blockcontroller/
acp.rs
