agent_engine/sidecar/
manager.rs

1//! Sidecar lifecycle and IO.
2//!
3//! [`SidecarManager`] spawns a sidecar process, writes line-JSON
4//! [`SidecarCommand`] values to its stdin, and surfaces the
5//! deserialized [`SidecarFrame`] stream as higher-level
6//! [`SidecarLifecycleEvent`] values on an mpsc channel.
7//!
8//! Modality-agnostic. Plugin-specific work lives in the plugin process;
9//! this module is intentionally small and dependency-free beyond `tokio`
10//! and `serde_json`.
11
12use std::ffi::OsStr;
13use std::path::Path;
14use std::process::Stdio;
15use std::sync::Arc;
16
17use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
18use tokio::process::{Child, ChildStdin, Command};
19use tokio::sync::{mpsc, Mutex};
20
21/// Maximum bytes per sidecar stdout/stderr line. A hostile or buggy sidecar
22/// sending a single unbounded line could OOM the host without this cap.
23const MAX_SIDECAR_LINE_BYTES: u64 = 1024 * 1024; // 1 MiB, matches MCP
24
25use super::protocol::{InsertTextMode, SidecarCommand, SidecarFrame, SIDECAR_PROTOCOL_VERSION};
26
27const EVENT_CHANNEL_CAPACITY: usize = 64;
28
29/// High-level events emitted by the manager. This is a curated subset
30/// of [`SidecarFrame`] tailored for chatui consumers; plugin-specific
31/// frames that are not actionable by core are dropped.
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub enum SidecarLifecycleEvent {
34    /// Sidecar handshake complete.
35    Ready {
36        protocol_version: u16,
37        extension: String,
38        capabilities: Vec<String>,
39    },
40    /// Sidecar reports a plugin-defined state transition.
41    StateChanged {
42        state: String,
43        label: Option<String>,
44    },
45    /// Sidecar wants text applied to the input buffer.
46    InsertText { text: String, mode: InsertTextMode },
47    /// Sidecar reported an error message.
48    Error(String),
49    /// Sidecar process exited (clean or otherwise).
50    Exited,
51}
52
53/// Errors surfaced by the manager.
54#[derive(Debug, thiserror::Error)]
55pub enum SidecarError {
56    #[error("failed to spawn sidecar {bin}: {source}")]
57    Spawn {
58        bin: String,
59        #[source]
60        source: std::io::Error,
61    },
62    #[error("sidecar stdin/stdout was not captured")]
63    PipesUnavailable,
64    #[error("sidecar IO error: {0}")]
65    Io(#[from] std::io::Error),
66    #[error("sidecar process has already shut down")]
67    AlreadyShutDown,
68    #[error("failed to encode sidecar command: {0}")]
69    Encode(#[from] serde_json::Error),
70    #[error("sidecar protocol error: {0}")]
71    Protocol(String),
72}
73
74/// Supervises one sidecar process and its line-JSON streams.
75///
76/// Construct via [`SidecarManager::spawn`]; drive with [`press`],
77/// [`release`], [`shutdown`]. Receive events with [`next_event`].
78pub struct SidecarManager {
79    child: Option<Child>,
80    stdin: Arc<Mutex<Option<ChildStdin>>>,
81    rx: mpsc::Receiver<SidecarLifecycleEvent>,
82    reader_handle: Option<tokio::task::JoinHandle<()>>,
83    stderr_handle: Option<tokio::task::JoinHandle<()>>,
84}
85
86impl SidecarManager {
87    /// Spawn `bin` with `args`, send the [`Init`] handshake, and start
88    /// the background reader task.
89    ///
90    /// [`Init`]: SidecarCommand::Init
91    pub async fn spawn(
92        bin: &Path,
93        args: &[String],
94        config: serde_json::Value,
95    ) -> Result<Self, SidecarError> {
96        let mut command = Command::new(bin);
97        command
98            .args(args.iter().map(OsStr::new))
99            .stdin(Stdio::piped())
100            .stdout(Stdio::piped())
101            .stderr(Stdio::piped())
102            .kill_on_drop(true);
103
104        let mut child = command.spawn().map_err(|source| SidecarError::Spawn {
105            bin: bin.display().to_string(),
106            source,
107        })?;
108
109        let stdin = child.stdin.take().ok_or(SidecarError::PipesUnavailable)?;
110        let stdout = child.stdout.take().ok_or(SidecarError::PipesUnavailable)?;
111        let stderr = child.stderr.take();
112
113        let (tx, rx) = mpsc::channel(EVENT_CHANNEL_CAPACITY);
114        let stdin = Arc::new(Mutex::new(Some(stdin)));
115
116        // Reader task: parse line-JSON events and forward as SidecarLifecycleEvent.
117        let event_tx = tx.clone();
118        let reader_handle = tokio::spawn(async move {
119            let mut reader = BufReader::new(stdout);
120            let mut line = String::new();
121            loop {
122                line.clear();
123                let mut limited = (&mut reader).take(MAX_SIDECAR_LINE_BYTES + 1);
124                match tokio::io::AsyncBufReadExt::read_line(&mut limited, &mut line).await {
125                    Ok(0) => break, // EOF
126                    Ok(_) => {}
127                    Err(_) => break,
128                }
129                if line.len() as u64 > MAX_SIDECAR_LINE_BYTES {
130                    let _ = event_tx
131                        .send(SidecarLifecycleEvent::Error(format!(
132                            "sidecar stdout line exceeded {MAX_SIDECAR_LINE_BYTES} bytes, dropped"
133                        )))
134                        .await;
135                    continue;
136                }
137                let line = line.trim_end();
138                if line.trim().is_empty() {
139                    continue;
140                }
141                let event = match serde_json::from_str::<SidecarFrame>(line) {
142                    Ok(ev) => ev,
143                    Err(err) => {
144                        let _ = event_tx
145                            .send(SidecarLifecycleEvent::Error(format!(
146                                "failed to parse sidecar line: {err}: {line}"
147                            )))
148                            .await;
149                        continue;
150                    }
151                };
152                let mapped = match event {
153                    SidecarFrame::Hello {
154                        protocol_version,
155                        extension,
156                        capabilities,
157                    } => {
158                        if protocol_version < SIDECAR_PROTOCOL_VERSION {
159                            Some(SidecarLifecycleEvent::Error(format!(
160                                "sidecar protocol v{protocol_version} is too old; host requires v{SIDECAR_PROTOCOL_VERSION}. Update the plugin via /plugins."
161                            )))
162                        } else {
163                            Some(SidecarLifecycleEvent::Ready {
164                                protocol_version,
165                                extension,
166                                capabilities,
167                            })
168                        }
169                    }
170                    SidecarFrame::Status { state, label, .. } => {
171                        Some(SidecarLifecycleEvent::StateChanged { state, label })
172                    }
173                    SidecarFrame::InsertText { text, mode } => {
174                        Some(SidecarLifecycleEvent::InsertText { text, mode })
175                    }
176                    SidecarFrame::Error { message } => Some(SidecarLifecycleEvent::Error(message)),
177                    SidecarFrame::Custom => None,
178                };
179                if let Some(event) = mapped {
180                    if event_tx.send(event).await.is_err() {
181                        // Receiver dropped — give up.
182                        break;
183                    }
184                }
185            }
186            let _ = event_tx.send(SidecarLifecycleEvent::Exited).await;
187        });
188
189        // Stderr task: forward sidecar stderr to tracing for diagnostics.
190        let stderr_handle = stderr.map(|stderr| {
191            tokio::spawn(async move {
192                let mut reader = BufReader::new(stderr);
193                let mut line = String::new();
194                loop {
195                    line.clear();
196                    let mut limited = (&mut reader).take(MAX_SIDECAR_LINE_BYTES + 1);
197                    match tokio::io::AsyncBufReadExt::read_line(&mut limited, &mut line).await {
198                        Ok(0) => break,
199                        Ok(_) => {}
200                        Err(_) => break,
201                    }
202                    if line.len() as u64 > MAX_SIDECAR_LINE_BYTES {
203                        tracing::warn!(target: "sidecar::manager", "stderr line exceeded bound, dropped");
204                        continue;
205                    }
206                    tracing::debug!(target: "sidecar::manager", "{}", line.trim_end());
207                }
208            })
209        });
210
211        let mut manager = Self {
212            child: Some(child),
213            stdin,
214            rx,
215            reader_handle: Some(reader_handle),
216            stderr_handle,
217        };
218
219        // Wait for the sidecar's Hello frame before sending Init.
220        // The sidecar must announce its protocol version first so we can
221        // reject incompatible versions before committing to the handshake.
222        // Timeout: 10s — if the sidecar can't say Hello in 10s, it's broken.
223        let hello_timeout =
224            tokio::time::timeout(std::time::Duration::from_secs(10), manager.rx.recv())
225                .await
226                .map_err(|_| {
227                    SidecarError::Protocol("sidecar did not send Hello within 10s".to_string())
228                })?;
229
230        match hello_timeout {
231            Some(SidecarLifecycleEvent::Ready { .. }) => {
232                // Hello received and protocol version is acceptable — proceed with Init
233            }
234            Some(SidecarLifecycleEvent::Error(e)) => {
235                return Err(SidecarError::Protocol(format!("sidecar Hello failed: {e}")));
236            }
237            Some(other) => {
238                return Err(SidecarError::Protocol(format!(
239                    "expected Hello from sidecar, got: {:?}",
240                    other
241                )));
242            }
243            None => {
244                return Err(SidecarError::Protocol(
245                    "sidecar exited before sending Hello".to_string(),
246                ));
247            }
248        }
249
250        manager.send(SidecarCommand::Init { config }).await?;
251        Ok(manager)
252    }
253
254    /// Send a trigger press command.
255    pub async fn press(&mut self) -> Result<(), SidecarError> {
256        self.send(SidecarCommand::Trigger {
257            name: "press".into(),
258            payload: None,
259        })
260        .await
261    }
262
263    /// Send a trigger release command.
264    pub async fn release(&mut self) -> Result<(), SidecarError> {
265        self.send(SidecarCommand::Trigger {
266            name: "release".into(),
267            payload: None,
268        })
269        .await
270    }
271
272    /// Send a graceful `shutdown` command and reap the child process.
273    pub async fn shutdown(&mut self) -> Result<(), SidecarError> {
274        let _ = self.send(SidecarCommand::Shutdown).await;
275        // Drop the stdin so the sidecar sees EOF if it ignored shutdown.
276        if let Some(mut stdin) = self.stdin.lock().await.take() {
277            let _ = stdin.shutdown().await;
278        }
279        if let Some(mut child) = self.child.take() {
280            // Grace period: wait up to 2s, then kill. Matches MCP/extension patterns.
281            match tokio::time::timeout(
282                std::time::Duration::from_secs(2),
283                child.wait(),
284            )
285            .await
286            {
287                Ok(_) => {} // exited gracefully
288                Err(_) => {
289                    let _ = child.kill().await;
290                }
291            }
292        }
293        if let Some(handle) = self.reader_handle.take() {
294            handle.abort();
295        }
296        if let Some(handle) = self.stderr_handle.take() {
297            handle.abort();
298        }
299        Ok(())
300    }
301
302    /// Receive the next high-level event, or `None` if the channel
303    /// closed (sidecar exited and reader task drained).
304    pub async fn next_event(&mut self) -> Option<SidecarLifecycleEvent> {
305        self.rx.recv().await
306    }
307
308    async fn send(&self, cmd: SidecarCommand) -> Result<(), SidecarError> {
309        let mut buf = serde_json::to_vec(&cmd)?;
310        buf.push(b'\n');
311        let mut guard = self.stdin.lock().await;
312        let stdin = guard.as_mut().ok_or(SidecarError::AlreadyShutDown)?;
313        stdin.write_all(&buf).await?;
314        stdin.flush().await?;
315        Ok(())
316    }
317}
318
319impl Drop for SidecarManager {
320    fn drop(&mut self) {
321        // Best-effort: kill the child if shutdown wasn't called.
322        if let Some(handle) = self.reader_handle.take() {
323            handle.abort();
324        }
325        if let Some(handle) = self.stderr_handle.take() {
326            handle.abort();
327        }
328    }
329}
agent_engine/sidecar/manager.rs

agent_engine/sidecar/
manager.rs
