agentmux_srv/backend/lsp/
supervisor.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3//
4// LspSupervisor — spawns language-server child processes and proxies
5// LSP messages between the editor pane and the server.
6//
7// Design notes:
8//   * One server per (workspace_root, language). Same workspace open in
9//     two panes shares one server (refcounted).
10//   * Backend is a dumb proxy: it frames messages on the wire and
11//     forwards bodies. LSP semantics (initialize, didOpen, …) are
12//     enforced on the frontend in LspClient.
13//   * Server stdout is read on a tokio task; each framed message is
14//     broadcast as a `lsp:message` event on the EventBus.
15//   * Process is killed on supervisor drop (kill_on_drop=true).
16//
17// Spec: docs/specs/SPEC_EDITOR_LSP_AND_THEMES_2026-05-26.md (Tier 1).
18
19#[cfg(windows)]
20use agentmux_common::win32::NoWindow;
21use std::collections::HashMap;
22use std::path::PathBuf;
23use std::process::Stdio;
24use std::sync::Arc;
25
26use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
27use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command};
28use tokio::sync::Mutex;
29
30use crate::backend::eventbus::EventBus;
31
32pub type ServerId = String;
33
34pub struct StartArgs {
35    pub language: String,
36    pub workspace_root: PathBuf,
37}
38
39pub struct StartResult {
40    pub server_id: ServerId,
41    pub workspace_root: String,
42}
43
44#[derive(thiserror::Error, Debug)]
45pub enum LspError {
46    /// The configured server binary couldn't be found on PATH. The
47    /// frontend uses this to render the install banner.
48    #[error("server_binary_not_found:{language}")]
49    BinaryNotFound { language: String },
50
51    #[error("unsupported language: {0}")]
52    UnsupportedLanguage(String),
53
54    #[error("spawn failed: {0}")]
55    SpawnFailed(String),
56
57    #[error("server not running: {0}")]
58    ServerNotFound(String),
59
60    #[error("io: {0}")]
61    Io(#[from] std::io::Error),
62}
63
64impl LspError {
65    pub fn to_wire_string(&self) -> String {
66        self.to_string()
67    }
68}
69
70struct ServerHandle {
71    stdin: Mutex<ChildStdin>,
72    refcount: std::sync::atomic::AtomicUsize,
73    _child: Mutex<Child>, // hold ownership; kill_on_drop fires on supervisor drop
74}
75
76pub struct LspSupervisor {
77    servers: Mutex<HashMap<(String, String), Arc<ServerHandle>>>,
78    event_bus: Arc<EventBus>,
79}
80
81impl LspSupervisor {
82    pub fn new(event_bus: Arc<EventBus>) -> Self {
83        Self {
84            servers: Mutex::new(HashMap::new()),
85            event_bus,
86        }
87    }
88
89    /// Start (or attach to existing) server for `(workspace_root, language)`.
90    pub async fn start(&self, args: StartArgs) -> Result<StartResult, LspError> {
91        let lang = args.language.clone();
92        let root_str = args.workspace_root.to_string_lossy().to_string();
93        let key = (root_str.clone(), lang.clone());
94        let server_id = make_server_id(&root_str, &lang);
95
96        // Fast-path: existing server, bump refcount.
97        {
98            let servers = self.servers.lock().await;
99            if let Some(handle) = servers.get(&key) {
100                handle
101                    .refcount
102                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
103                return Ok(StartResult {
104                    server_id,
105                    workspace_root: root_str,
106                });
107            }
108        }
109
110        // Resolve binary on PATH. Phase 1 is opinionated about supported
111        // languages — see `binary_for_language`.
112        let binary = binary_for_language(&lang)
113            .ok_or_else(|| LspError::UnsupportedLanguage(lang.clone()))?;
114        let resolved = which::which(binary).map_err(|_| LspError::BinaryNotFound {
115            language: lang.clone(),
116        })?;
117
118        tracing::info!(
119            language = %lang,
120            workspace = %root_str,
121            binary = %resolved.display(),
122            "LSP: spawning server"
123        );
124
125        let mut cmd = Command::new(&resolved);
126        // A long-lived third-party language server. Not ours, not a helper, so it
127        // gets the strict policy — no instance identity, no API endpoint.
128        crate::backend::pane_env::sanitize_external_command(&mut cmd);
129        // typescript-language-server, pyright-langserver, gopls — all use --stdio
130        // by convention. rust-analyzer reads from stdio without a flag (the
131        // flag is ignored). Safe to pass on all.
132        cmd.arg("--stdio")
133            .current_dir(&args.workspace_root)
134            .stdin(Stdio::piped())
135            .stdout(Stdio::piped())
136            .stderr(Stdio::piped())
137            .kill_on_drop(true);
138
139        // On Windows: suppress console-window allocation. Spawned from the
140        // windowless srv without CREATE_NO_WINDOW, each LSP server (ts-language-
141        // server, pyright, gopls, rust-analyzer) opens a Windows Terminal window
142        // (Win11 default-terminal handler). stdio is piped, so no console is
143        // needed. See docs/retro/retro-windows-terminal-window-leak-2026-06-21.md.
144        #[cfg(windows)]
145        {
146            cmd.no_window();
147        }
148
149        let mut child = cmd
150            .spawn()
151            .map_err(|e| LspError::SpawnFailed(e.to_string()))?;
152        let stdin = child
153            .stdin
154            .take()
155            .ok_or_else(|| LspError::SpawnFailed("no stdin".to_string()))?;
156        let stdout = child
157            .stdout
158            .take()
159            .ok_or_else(|| LspError::SpawnFailed("no stdout".to_string()))?;
160        let stderr = child
161            .stderr
162            .take()
163            .ok_or_else(|| LspError::SpawnFailed("no stderr".to_string()))?;
164
165        // Spawn the stdout reader. Each framed message becomes an
166        // `lsp:message` WSEvent broadcast for the frontend to dispatch.
167        let event_bus = self.event_bus.clone();
168        let server_id_for_stdout = server_id.clone();
169        tokio::spawn(async move {
170            read_lsp_messages(stdout, server_id_for_stdout, event_bus).await;
171        });
172
173        // Drain stderr — a chatty server (rust-analyzer, tsserver on errors)
174        // can fill the pipe buffer and stall its own stdin/stdout. Log each
175        // line via tracing so the supervisor surface remains observable.
176        let server_id_for_stderr = server_id.clone();
177        tokio::spawn(async move {
178            drain_lsp_stderr(stderr, server_id_for_stderr).await;
179        });
180
181        let handle = Arc::new(ServerHandle {
182            stdin: Mutex::new(stdin),
183            refcount: std::sync::atomic::AtomicUsize::new(1),
184            _child: Mutex::new(child),
185        });
186
187        // Re-check under the second lock — a concurrent start() for the same
188        // (workspace, language) could have raced past the fast-path check
189        // above. If so, drop our just-spawned child (kill_on_drop reaps it)
190        // and bump the existing entry's refcount instead.
191        let mut servers = self.servers.lock().await;
192        if let Some(existing) = servers.get(&key) {
193            existing
194                .refcount
195                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
196            drop(servers);
197            drop(handle); // releases our spawned child via kill_on_drop
198            return Ok(StartResult {
199                server_id,
200                workspace_root: root_str,
201            });
202        }
203        servers.insert(key, handle);
204
205        Ok(StartResult {
206            server_id,
207            workspace_root: root_str,
208        })
209    }
210
211    /// Forward an LSP message (pre-serialized JSON) to the server's stdin.
212    pub async fn send(&self, server_id: &str, message_json: &str) -> Result<(), LspError> {
213        let key = parse_server_id(server_id)?;
214        let handle = {
215            let servers = self.servers.lock().await;
216            servers
217                .get(&key)
218                .cloned()
219                .ok_or_else(|| LspError::ServerNotFound(server_id.to_string()))?
220        };
221        let mut stdin = handle.stdin.lock().await;
222        let body = message_json.as_bytes();
223        // LSP header: Content-Length: <bytes>\r\n\r\n
224        let header = format!("Content-Length: {}\r\n\r\n", body.len());
225        stdin.write_all(header.as_bytes()).await?;
226        stdin.write_all(body).await?;
227        stdin.flush().await?;
228        Ok(())
229    }
230
231    /// Decrement refcount. When it reaches zero, drop the handle so the
232    /// child process exits via `kill_on_drop`. Phase 1 does not yet
233    /// implement the 60s idle grace from the spec (deferred — see
234    /// supervisor.rs Open Questions follow-up).
235    pub async fn stop(&self, server_id: &str) -> Result<(), LspError> {
236        let key = parse_server_id(server_id)?;
237        let mut servers = self.servers.lock().await;
238        if let Some(handle) = servers.get(&key) {
239            let prev = handle
240                .refcount
241                .fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
242            if prev <= 1 {
243                tracing::info!(server_id = %server_id, "LSP: shutting down (refcount=0)");
244                servers.remove(&key);
245                // Drop fires kill_on_drop → process exits.
246            }
247        }
248        Ok(())
249    }
250}
251
252fn make_server_id(workspace_root: &str, language: &str) -> ServerId {
253    // Use a sentinel separator unlikely to appear in either component.
254    format!("lsp://{language}::{workspace_root}")
255}
256
257fn parse_server_id(server_id: &str) -> Result<(String, String), LspError> {
258    let stripped = server_id
259        .strip_prefix("lsp://")
260        .ok_or_else(|| LspError::ServerNotFound(server_id.to_string()))?;
261    let (lang, root) = stripped
262        .split_once("::")
263        .ok_or_else(|| LspError::ServerNotFound(server_id.to_string()))?;
264    Ok((root.to_string(), lang.to_string()))
265}
266
267/// Phase 1 binary table — the cross-platform `--stdio`-mode binary name
268/// for each supported language. Extended in Phase 3.
269fn binary_for_language(language: &str) -> Option<&'static str> {
270    Some(match language {
271        "typescript" | "javascript" => "typescript-language-server",
272        // Phase 3 candidates — pre-listed so the discovery path works
273        // for follow-up phases without code churn.
274        "rust" => "rust-analyzer",
275        "python" => "pyright-langserver",
276        "go" => "gopls",
277        "c" | "cpp" => "clangd",
278        _ => return None,
279    })
280}
281
282/// Read framed LSP messages from the server's stdout. Each message is
283/// broadcast as an `lsp:message` event with `{ server_id, message }`.
284/// Loop exits on EOF (server died) — the supervisor's child handle
285/// then surfaces the exit via Drop on the next `stop` cycle.
286async fn read_lsp_messages(stdout: ChildStdout, server_id: ServerId, event_bus: Arc<EventBus>) {
287    let mut reader = BufReader::new(stdout);
288    loop {
289        let mut content_length: usize = 0;
290        // Read headers until blank line.
291        loop {
292            let mut line = String::new();
293            match reader.read_line(&mut line).await {
294                Ok(0) => {
295                    tracing::info!(server_id = %server_id, "LSP: stdout EOF (server exited)");
296                    return;
297                }
298                Ok(_) => {}
299                Err(e) => {
300                    tracing::warn!(server_id = %server_id, error = %e, "LSP: stdout read error");
301                    return;
302                }
303            }
304            let trimmed = line.trim_end_matches(['\r', '\n']);
305            if trimmed.is_empty() {
306                break;
307            }
308            if let Some(value) = trimmed.strip_prefix("Content-Length:") {
309                content_length = value.trim().parse().unwrap_or(0);
310            }
311            // Other headers (e.g. Content-Type) are ignored — LSP only
312            // requires Content-Length.
313        }
314        if content_length == 0 {
315            continue;
316        }
317        let mut buf = vec![0u8; content_length];
318        if reader.read_exact(&mut buf).await.is_err() {
319            tracing::warn!(server_id = %server_id, "LSP: body read failed");
320            return;
321        }
322        let body = match std::str::from_utf8(&buf) {
323            Ok(s) => s,
324            Err(_) => {
325                tracing::warn!(server_id = %server_id, "LSP: non-UTF8 body");
326                continue;
327            }
328        };
329        // Parse to a serde_json::Value so the frontend gets structured JSON,
330        // not a string-of-JSON.
331        let message = match serde_json::from_str::<serde_json::Value>(body) {
332            Ok(v) => v,
333            Err(e) => {
334                tracing::warn!(server_id = %server_id, error = %e, "LSP: invalid JSON body");
335                continue;
336            }
337        };
338        let event = crate::backend::mps_events::LspMessageEvent { server_id: server_id.to_string(), message };
339        event_bus.broadcast_event(&EventBus::typed_ws_event::<crate::backend::mps_events::LspMessage>(&event, String::new()));
340    }
341}
342
343/// Drain the server's stderr line-by-line, logging each line via tracing.
344/// Without this the OS pipe buffer fills up on chatty servers (rust-analyzer
345/// during indexing, tsserver on startup errors) and the server's own writes
346/// block, which can stall stdin/stdout handling.
347async fn drain_lsp_stderr(stderr: ChildStderr, server_id: ServerId) {
348    let mut reader = BufReader::new(stderr);
349    let mut line = String::new();
350    loop {
351        line.clear();
352        match reader.read_line(&mut line).await {
353            Ok(0) => return, // EOF
354            Ok(_) => {
355                tracing::debug!(
356                    server_id = %server_id,
357                    stderr = %line.trim_end_matches(['\r', '\n']),
358                    "LSP stderr"
359                );
360            }
361            Err(e) => {
362                tracing::warn!(server_id = %server_id, error = %e, "LSP: stderr read error");
363                return;
364            }
365        }
366    }
367}
368
369#[cfg(test)]
370mod tests {
371    use super::*;
372
373    #[test]
374    fn server_id_roundtrip() {
375        let id = make_server_id("/Users/asaf/project", "typescript");
376        let (root, lang) = parse_server_id(&id).unwrap();
377        assert_eq!(root, "/Users/asaf/project");
378        assert_eq!(lang, "typescript");
379    }
380
381    #[test]
382    fn server_id_with_double_colon_in_root() {
383        // Edge case — Windows paths can contain colons (`C:`); make sure
384        // our separator doesn't trip them.
385        let id = make_server_id("C:\\Users\\asaf\\project", "rust");
386        let (root, lang) = parse_server_id(&id).unwrap();
387        assert_eq!(root, "C:\\Users\\asaf\\project");
388        assert_eq!(lang, "rust");
389    }
390
391    #[test]
392    fn binary_lookup_table() {
393        assert_eq!(binary_for_language("typescript"), Some("typescript-language-server"));
394        assert_eq!(binary_for_language("javascript"), Some("typescript-language-server"));
395        assert_eq!(binary_for_language("rust"), Some("rust-analyzer"));
396        assert_eq!(binary_for_language("python"), Some("pyright-langserver"));
397        assert_eq!(binary_for_language("go"), Some("gopls"));
398        assert_eq!(binary_for_language("c"), Some("clangd"));
399        assert_eq!(binary_for_language("cpp"), Some("clangd"));
400        assert_eq!(binary_for_language("brainfuck"), None);
401    }
402}
agentmux_srv/backend/lsp/supervisor.rs

agentmux_srv/backend/lsp/
supervisor.rs
