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

agentmux_srv\backend\lsp/
supervisor.rs
