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::time::Duration;
16
17use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
18use tokio::process::{Child, ChildStdin, Command};
19use tokio::sync::mpsc;
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/// Bound every command frame, including Init, when a sidecar stops reading.
29const COMMAND_WRITE_TIMEOUT: Duration = Duration::from_secs(2);
30
31/// High-level events emitted by the manager. This is a curated subset
32/// of [`SidecarFrame`] tailored for chatui consumers; plugin-specific
33/// frames that are not actionable by core are dropped.
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub enum SidecarLifecycleEvent {
36    /// Sidecar handshake complete.
37    Ready {
38        protocol_version: u16,
39        extension: String,
40        capabilities: Vec<String>,
41    },
42    /// Sidecar reports a plugin-defined state transition.
43    StateChanged {
44        state: String,
45        label: Option<String>,
46    },
47    /// Sidecar wants text applied to the input buffer.
48    InsertText { text: String, mode: InsertTextMode },
49    /// Sidecar reported an error message.
50    Error(String),
51    /// Sidecar process exited (clean or otherwise).
52    Exited,
53}
54
55/// Errors surfaced by the manager.
56#[derive(Debug, thiserror::Error)]
57pub enum SidecarError {
58    #[error("failed to spawn sidecar {bin}: {source}")]
59    Spawn {
60        bin: String,
61        #[source]
62        source: std::io::Error,
63    },
64    #[error("sidecar stdin/stdout was not captured")]
65    PipesUnavailable,
66    #[error("sidecar IO error: {0}")]
67    Io(#[from] std::io::Error),
68    /// No more commands can be sent (shutdown, failed write, or cancelled write).
69    #[error("sidecar command input is closed")]
70    AlreadyShutDown,
71    #[error("failed to encode sidecar command: {0}")]
72    Encode(#[from] serde_json::Error),
73    #[error("sidecar protocol error: {0}")]
74    Protocol(String),
75}
76
77/// Supervises one sidecar process and its line-JSON streams.
78///
79/// Construct via [`SidecarManager::spawn`]; drive with [`press`],
80/// [`release`], [`shutdown`]. Receive events with [`next_event`].
81pub struct SidecarManager {
82    hello_capabilities: Vec<String>,
83    child: Option<Child>,
84    stdin: Option<ChildStdin>,
85    rx: mpsc::Receiver<SidecarLifecycleEvent>,
86    reader_handle: Option<tokio::task::JoinHandle<()>>,
87    stderr_handle: Option<tokio::task::JoinHandle<()>>,
88}
89
90impl SidecarManager {
91    /// Spawn `bin` with `args`, wait for Hello, send [`Init`], and start
92    /// the background reader task. Hello is readiness; no post-Init status
93    /// is required. Hello has a 10s deadline and command writes have a 2s
94    /// deadline. Cancelling startup drops the child and reader tasks.
95    ///
96    /// [`Init`]: SidecarCommand::Init
97    pub async fn spawn(
98        bin: &Path,
99        args: &[String],
100        config: serde_json::Value,
101    ) -> Result<Self, SidecarError> {
102        let mut command = Command::new(bin);
103        command
104            .args(args.iter().map(OsStr::new))
105            .stdin(Stdio::piped())
106            .stdout(Stdio::piped())
107            .stderr(Stdio::piped())
108            .kill_on_drop(true);
109
110        let mut child = command.spawn().map_err(|source| SidecarError::Spawn {
111            bin: bin.display().to_string(),
112            source,
113        })?;
114
115        let stdin = child.stdin.take().ok_or(SidecarError::PipesUnavailable)?;
116        let stdout = child.stdout.take().ok_or(SidecarError::PipesUnavailable)?;
117        let stderr = child.stderr.take();
118
119        let (tx, rx) = mpsc::channel(EVENT_CHANNEL_CAPACITY);
120
121        // Reader task: parse line-JSON events and forward as SidecarLifecycleEvent.
122        let event_tx = tx.clone();
123        let reader_handle = tokio::spawn(async move {
124            let mut reader = BufReader::new(stdout);
125            let mut line = String::new();
126            loop {
127                line.clear();
128                let mut limited = (&mut reader).take(MAX_SIDECAR_LINE_BYTES + 1);
129                match tokio::io::AsyncBufReadExt::read_line(&mut limited, &mut line).await {
130                    Ok(0) => break, // EOF
131                    Ok(_) => {}
132                    Err(_) => break,
133                }
134                if line.len() as u64 > MAX_SIDECAR_LINE_BYTES {
135                    let _ = event_tx
136                        .send(SidecarLifecycleEvent::Error(format!(
137                            "sidecar stdout line exceeded {MAX_SIDECAR_LINE_BYTES} bytes, dropped"
138                        )))
139                        .await;
140                    continue;
141                }
142                let line = line.trim_end();
143                if line.trim().is_empty() {
144                    continue;
145                }
146                let event = match serde_json::from_str::<SidecarFrame>(line) {
147                    Ok(ev) => ev,
148                    Err(err) => {
149                        let _ = event_tx
150                            .send(SidecarLifecycleEvent::Error(format!(
151                                "failed to parse sidecar line: {err}: {line}"
152                            )))
153                            .await;
154                        continue;
155                    }
156                };
157                let mapped = match event {
158                    SidecarFrame::Hello {
159                        protocol_version,
160                        extension,
161                        capabilities,
162                    } => {
163                        if protocol_version < SIDECAR_PROTOCOL_VERSION {
164                            Some(SidecarLifecycleEvent::Error(format!(
165                                "sidecar protocol v{protocol_version} is too old; host requires v{SIDECAR_PROTOCOL_VERSION}. Update the plugin via /plugins."
166                            )))
167                        } else {
168                            Some(SidecarLifecycleEvent::Ready {
169                                protocol_version,
170                                extension,
171                                capabilities,
172                            })
173                        }
174                    }
175                    SidecarFrame::Status { state, label, .. } => {
176                        Some(SidecarLifecycleEvent::StateChanged { state, label })
177                    }
178                    SidecarFrame::InsertText { text, mode } => {
179                        Some(SidecarLifecycleEvent::InsertText { text, mode })
180                    }
181                    SidecarFrame::Error { message } => Some(SidecarLifecycleEvent::Error(message)),
182                    SidecarFrame::Custom => None,
183                };
184                if let Some(event) = mapped {
185                    if event_tx.send(event).await.is_err() {
186                        // Receiver dropped — give up.
187                        break;
188                    }
189                }
190            }
191            let _ = event_tx.send(SidecarLifecycleEvent::Exited).await;
192        });
193
194        // Stderr task: forward sidecar stderr to tracing for diagnostics.
195        let stderr_handle = stderr.map(|stderr| {
196            tokio::spawn(async move {
197                let mut reader = BufReader::new(stderr);
198                let mut line = String::new();
199                loop {
200                    line.clear();
201                    let mut limited = (&mut reader).take(MAX_SIDECAR_LINE_BYTES + 1);
202                    match tokio::io::AsyncBufReadExt::read_line(&mut limited, &mut line).await {
203                        Ok(0) => break,
204                        Ok(_) => {}
205                        Err(_) => break,
206                    }
207                    if line.len() as u64 > MAX_SIDECAR_LINE_BYTES {
208                        tracing::warn!(target: "sidecar::manager", "stderr line exceeded bound, dropped");
209                        continue;
210                    }
211                    tracing::debug!(target: "sidecar::manager", "{}", line.trim_end());
212                }
213            })
214        });
215
216        let mut manager = Self {
217            hello_capabilities: Vec::new(),
218            child: Some(child),
219            stdin: Some(stdin),
220            rx,
221            reader_handle: Some(reader_handle),
222            stderr_handle,
223        };
224
225        // Wait for the sidecar's Hello frame before sending Init.
226        // The sidecar must announce its protocol version first so we can
227        // reject incompatible versions before committing to the handshake.
228        // Timeout: 10s — if the sidecar can't say Hello in 10s, it's broken.
229        let hello_timeout =
230            tokio::time::timeout(std::time::Duration::from_secs(10), manager.rx.recv())
231                .await
232                .map_err(|_| {
233                    SidecarError::Protocol("sidecar did not send Hello within 10s".to_string())
234                })?;
235
236        match hello_timeout {
237            Some(SidecarLifecycleEvent::Ready { capabilities, .. }) => {
238                manager.hello_capabilities = capabilities;
239            }
240            Some(SidecarLifecycleEvent::Error(e)) => {
241                return Err(SidecarError::Protocol(format!("sidecar Hello failed: {e}")));
242            }
243            Some(SidecarLifecycleEvent::Exited) | None => {
244                return Err(SidecarError::Protocol(
245                    "sidecar exited before sending Hello".to_string(),
246                ));
247            }
248            Some(other) => {
249                return Err(SidecarError::Protocol(format!(
250                    "expected Hello from sidecar, got: {:?}",
251                    other
252                )));
253            }
254        }
255
256        manager.send(SidecarCommand::Init { config }).await?;
257        Ok(manager)
258    }
259
260    /// Optional initialization contract. Without it Hello+Init is the legacy
261    /// protocol-ready boundary, not proof that model/device loading finished.
262    pub fn ready_after_init(&self) -> bool {
263        self.hello_capabilities
264            .iter()
265            .any(|c| c == "ready_after_init")
266    }
267
268    /// Nonblocking drain for hosts about to publish readiness or activate.
269    pub fn try_next_event(&mut self) -> Option<SidecarLifecycleEvent> {
270        match self.rx.try_recv() {
271            Ok(event) => Some(event),
272            Err(mpsc::error::TryRecvError::Disconnected) => Some(SidecarLifecycleEvent::Exited),
273            Err(mpsc::error::TryRecvError::Empty) => None,
274        }
275    }
276
277    /// Send a trigger press command.
278    pub async fn press(&mut self) -> Result<(), SidecarError> {
279        self.send(SidecarCommand::Trigger {
280            name: "press".into(),
281            payload: None,
282        })
283        .await
284    }
285
286    /// Send a trigger release command.
287    pub async fn release(&mut self) -> Result<(), SidecarError> {
288        self.send(SidecarCommand::Trigger {
289            name: "release".into(),
290            payload: None,
291        })
292        .await
293    }
294
295    /// Send a graceful `shutdown` command and reap the child process.
296    /// A write error is returned only after cleanup; an already closed input
297    /// still permits cleanup and repeated shutdown calls.
298    pub async fn shutdown(&mut self) -> Result<(), SidecarError> {
299        let send_result = match self.send(SidecarCommand::Shutdown).await {
300            Err(SidecarError::AlreadyShutDown) => Ok(()),
301            result => result,
302        };
303        // Closing the pipe is synchronous; no unbounded flush/shutdown await.
304        // The sidecar sees EOF even if it ignored the shutdown command.
305        drop(self.stdin.take());
306        if let Some(mut child) = self.child.take() {
307            // Grace period: wait up to 2s, then kill. Matches MCP/extension patterns.
308            match tokio::time::timeout(std::time::Duration::from_secs(2), child.wait()).await {
309                Ok(_) => {} // exited gracefully
310                Err(_) => {
311                    let _ = child.kill().await;
312                }
313            }
314        }
315        if let Some(handle) = self.reader_handle.take() {
316            handle.abort();
317        }
318        if let Some(handle) = self.stderr_handle.take() {
319            handle.abort();
320        }
321        send_result
322    }
323
324    /// Receive the next high-level event, or `None` if the channel
325    /// closed (sidecar exited and reader task drained).
326    pub async fn next_event(&mut self) -> Option<SidecarLifecycleEvent> {
327        self.rx.recv().await
328    }
329
330    async fn send(&mut self, cmd: SidecarCommand) -> Result<(), SidecarError> {
331        let mut buf = serde_json::to_vec(&cmd)?;
332        buf.push(b'\n');
333        // All callers hold &mut self. Take ownership across the await so an IO
334        // error, timeout, or cancelled future closes stdin and leaves it None.
335        // A partially written JSON frame must never be followed by a retry.
336        let mut stdin = self.stdin.take().ok_or(SidecarError::AlreadyShutDown)?;
337        tokio::time::timeout(COMMAND_WRITE_TIMEOUT, async {
338            stdin.write_all(&buf).await?;
339            stdin.flush().await
340        })
341        .await
342        .map_err(|_| {
343            std::io::Error::new(
344                std::io::ErrorKind::TimedOut,
345                "sidecar command write did not complete within 2s; command input closed",
346            )
347        })??;
348        self.stdin = Some(stdin);
349        Ok(())
350    }
351}
352
353impl Drop for SidecarManager {
354    fn drop(&mut self) {
355        // Best-effort: kill the child if shutdown wasn't called.
356        if let Some(handle) = self.reader_handle.take() {
357            handle.abort();
358        }
359        if let Some(handle) = self.stderr_handle.take() {
360            handle.abort();
361        }
362    }
363}
agent_engine/sidecar/manager.rs

agent_engine/sidecar/
manager.rs
