返回 CodeWhale
client.rs
根目录 / crates / tui / src / lsp / client.rs
1 //! Thin JSON-RPC over stdio client for LSP servers.
2 //!
3 //! We deliberately do **not** depend on `tower-lsp` — it is a server-side
4 //! framework and dragging it in here would add hundreds of unnecessary
5 //! transitive dependencies and slow down `cargo build` for every contributor.
6 //! The LSP wire protocol is small enough that handling it ourselves is a
7 //! self-contained ~400 LOC and lets us keep total control of the spawn
8 //! lifecycle, timeouts, and the async surface.
9 //!
10 //! Architecture:
11 //!
12 //! - [`LspTransport`] is the trait the [`super::LspManager`] talks to. The
13 //! real implementation is [`StdioLspTransport`] (forks an LSP server with
14 //! `tokio::process::Command`); tests use `super::tests::FakeTransport`.
15 //! - [`StdioLspTransport`] runs three tokio tasks: a reader, a writer, and
16 //! the public API. Communication uses tokio mpsc channels.
17 //! - We parse `Content-Length`-framed JSON-RPC and route inbound messages
18 //! either to a per-request response slot (for replies) or to the
19 //! diagnostics queue (for `textDocument/publishDiagnostics` notifications).
20 //!
21 //! The transport is one-shot per file in MVP form: the manager spawns a
22 //! transport on demand for a language and reuses it. We do not implement
23 //! workspace sync beyond didOpen/didChange because the goal is "post-edit
24 //! diagnostics," not full IDE smartness.
25
26 use std::collections::HashMap;
27 use std::path::{Path, PathBuf};
28 use std::process::Stdio;
29 use std::sync::Arc;
30 use std::time::Duration;
31
32 use anyhow::{Context, Result, anyhow};
33 use async_trait::async_trait;
34 use serde_json::{Value, json};
35 use tokio::io::{AsyncReadExt, AsyncWriteExt};
36 use tokio::process::{Child, Command};
37 use tokio::sync::Mutex as AsyncMutex;
38 use tokio::sync::{mpsc, oneshot};
39 use tokio::time::timeout;
40
41 use super::diagnostics::{Diagnostic, Severity};
42 use crate::utils::spawn_supervised;
43
44 /// Trait the LSP manager talks to. A real LSP server speaks this via stdio;
45 /// tests use an in-process fake.
46 #[async_trait]
47 pub trait LspTransport: Send + Sync {
48 /// Notify the server that a file was opened or its contents updated, then
49 /// wait up to `wait` for a `publishDiagnostics` notification for that
50 /// file. Returns the diagnostics list (possibly empty). Implementations
51 /// must NOT block past `wait`.
52 async fn diagnostics_for(
53 &self,
54 path: &Path,
55 text: &str,
56 wait: Duration,
57 ) -> Result<Vec<Diagnostic>>;
58
59 /// Send a JSON-RPC request and wait up to `wait` for the reply.
60 ///
61 /// Default returns "unsupported" so diagnostic-only fakes keep working.
62 /// Real transports implement this for go-to-definition, symbols, and
63 /// references without spawning a second server lifecycle.
64 async fn request(&self, _method: &str, _params: Value, _wait: Duration) -> Result<Value> {
65 Err(anyhow!("LSP request not supported by this transport"))
66 }
67
68 /// Ensure `path` is open with `text` (didOpen/didChange) so position-based
69 /// requests can target it. Default is a no-op; real transports track opens.
70 async fn ensure_open(&self, _path: &Path, _text: &str) -> Result<()> {
71 Ok(())
72 }
73
74 /// Best-effort shutdown. Called via `LspManager::shutdown_all`.
75 #[allow(dead_code)]
76 async fn shutdown(&self);
77 }
78
79 /// Stdio-backed transport. Spawns the LSP server as a child process and
80 /// pipes JSON-RPC over stdin/stdout. Stderr is captured into a buffer so
81 /// callers can include it in error messages without polluting our own stderr.
82 pub struct StdioLspTransport {
83 /// JoinHandle for the running server. Held so the child stays alive for
84 /// the transport's lifetime; consumed during `shutdown`.
85 #[allow(dead_code)]
86 child: AsyncMutex<Option<Child>>,
87 /// Outgoing message sender to the writer task.
88 tx_outbound: mpsc::Sender<Vec<u8>>,
89 /// Inbound diagnostics queue. We push every `publishDiagnostics`
90 /// notification into here and the public API drains the relevant entries.
91 diagnostics_rx: AsyncMutex<mpsc::Receiver<(PathBuf, Vec<Diagnostic>)>>,
92 /// Map of in-flight request id -> reply slot for model-facing intelligence
93 /// requests (definition, references, symbols).
94 pending: Arc<AsyncMutex<HashMap<i64, oneshot::Sender<Value>>>>,
95 /// Monotonic request id counter for JSON-RPC request/reply methods.
96 next_id: AsyncMutex<i64>,
97 /// Language id passed in `textDocument/didOpen` (e.g. "rust").
98 language_id: String,
99 /// Track which files we have opened so the second touch sends
100 /// `didChange` instead of `didOpen`.
101 opened: AsyncMutex<HashMap<PathBuf, i64>>,
102 }
103
104 impl StdioLspTransport {
105 /// Spawn `command args…` and run the LSP `initialize` handshake. Returns
106 /// `Err` immediately if the binary is not on PATH or `initialize` fails.
107 pub async fn spawn(
108 command: &str,
109 args: &[String],
110 language_id: &str,
111 workspace: PathBuf,
112 ) -> Result<Self> {
113 let mut cmd = Command::new(command);
114 cmd.args(args);
115 cmd.stdin(Stdio::piped());
116 cmd.stdout(Stdio::piped());
117 cmd.stderr(Stdio::piped());
118 cmd.kill_on_drop(true);
119
120 let mut child = cmd
121 .spawn()
122 .with_context(|| format!("failed to spawn LSP server `{command}`"))?;
123
124 let stdin = child
125 .stdin
126 .take()
127 .context("LSP child has no stdin handle")?;
128 let stdout = child
129 .stdout
130 .take()
131 .context("LSP child has no stdout handle")?;
132
133 let (tx_outbound, rx_outbound) = mpsc::channel::<Vec<u8>>(64);
134 let (tx_inbound, rx_inbound) = mpsc::channel::<Value>(64);
135 let (tx_diag, rx_diag) = mpsc::channel::<(PathBuf, Vec<Diagnostic>)>(64);
136
137 // Writer task: drain outbound channel, frame with Content-Length, write to stdin.
138 spawn_supervised(
139 "lsp-writer",
140 std::panic::Location::caller(),
141 writer_task(stdin, rx_outbound),
142 );
143 // Reader task: parse Content-Length frames from stdout, push to inbound queue.
144 spawn_supervised(
145 "lsp-reader",
146 std::panic::Location::caller(),
147 reader_task(stdout, tx_inbound),
148 );
149 // Inbound dispatcher: routes notifications to `tx_diag`, replies to a
150 // pending map. We keep the pending map for completeness even though
151 // diagnostics polling itself does not reuse it.
152 let pending: Arc<AsyncMutex<HashMap<i64, oneshot::Sender<Value>>>> =
153 Arc::new(AsyncMutex::new(HashMap::new()));
154 spawn_supervised(
155 "lsp-dispatcher",
156 std::panic::Location::caller(),
157 dispatcher_task(rx_inbound, tx_diag, pending.clone()),
158 );
159
160 // Send `initialize` and wait for `initialized`. We synthesize id=1.
161 let init_payload = json!({
162 "jsonrpc": "2.0",
163 "id": 1,
164 "method": "initialize",
165 "params": {
166 "processId": std::process::id(),
167 "rootUri": uri_from_path(&workspace),
168 "capabilities": {
169 "textDocument": {
170 "publishDiagnostics": { "relatedInformation": false }
171 }
172 },
173 "workspaceFolders": [{
174 "uri": uri_from_path(&workspace),
175 "name": "workspace"
176 }]
177 }
178 });
179 send_message(&tx_outbound, &init_payload).await?;
180
181 // We do not actually wait for the initialize response here in MVP —
182 // most servers buffer notifications until they are ready, and waiting
183 // for `initialize` reply doubles the latency of the first edit. Send
184 // `initialized` immediately and let publishDiagnostics arrive on its
185 // own clock.
186 let initialized = json!({
187 "jsonrpc": "2.0",
188 "method": "initialized",
189 "params": {}
190 });
191 send_message(&tx_outbound, &initialized).await?;
192
193 Ok(Self {
194 child: AsyncMutex::new(Some(child)),
195 tx_outbound,
196 diagnostics_rx: AsyncMutex::new(rx_diag),
197 pending,
198 next_id: AsyncMutex::new(2),
199 language_id: language_id.to_string(),
200 opened: AsyncMutex::new(HashMap::new()),
201 })
202 }
203 }
204
205 impl StdioLspTransport {
206 async fn open_or_change(&self, path: &Path, text: &str) -> Result<String> {
207 let path_buf = path.to_path_buf();
208 let uri = uri_from_path(&path_buf);
209 let mut opened = self.opened.lock().await;
210 let is_new = !opened.contains_key(&path_buf);
211 let new_version = opened.get(&path_buf).copied().unwrap_or(0) + 1;
212 opened.insert(path_buf, new_version);
213 drop(opened);
214
215 let payload = if is_new {
216 json!({
217 "jsonrpc": "2.0",
218 "method": "textDocument/didOpen",
219 "params": {
220 "textDocument": {
221 "uri": uri.clone(),
222 "languageId": self.language_id,
223 "version": new_version,
224 "text": text
225 }
226 }
227 })
228 } else {
229 json!({
230 "jsonrpc": "2.0",
231 "method": "textDocument/didChange",
232 "params": {
233 "textDocument": {
234 "uri": uri.clone(),
235 "version": new_version
236 },
237 "contentChanges": [{ "text": text }]
238 }
239 })
240 };
241 send_message(&self.tx_outbound, &payload).await?;
242 Ok(uri)
243 }
244 }
245
246 #[async_trait]
247 impl LspTransport for StdioLspTransport {
248 async fn diagnostics_for(
249 &self,
250 path: &Path,
251 text: &str,
252 wait: Duration,
253 ) -> Result<Vec<Diagnostic>> {
254 let path_buf = path.to_path_buf();
255 self.open_or_change(path, text).await?;
256
257 // Drain matching `publishDiagnostics` notifications until `wait`
258 // elapses. Servers typically publish within a few hundred ms; for
259 // initial cold-start (rust-analyzer) it can be many seconds — but
260 // the manager guards us with a separate timeout.
261 let deadline = tokio::time::Instant::now() + wait;
262 let mut latest: Option<Vec<Diagnostic>> = None;
263
264 loop {
265 let now = tokio::time::Instant::now();
266 if now >= deadline {
267 break;
268 }
269 let remaining = deadline - now;
270 let mut rx = self.diagnostics_rx.lock().await;
271 let next = match timeout(remaining, rx.recv()).await {
272 Ok(Some(item)) => item,
273 Ok(None) => break, // channel closed
274 Err(_) => break, // timed out
275 };
276 drop(rx);
277 let (file, items) = next;
278 if file == path_buf {
279 latest = Some(items);
280 // We have a payload — return immediately. If the server
281 // re-publishes after rapid edits, the next call will sync.
282 break;
283 }
284 // Otherwise: notification was for a different file we previously
285 // opened. Discard and continue waiting.
286 }
287 Ok(latest.unwrap_or_default())
288 }
289
290 async fn ensure_open(&self, path: &Path, text: &str) -> Result<()> {
291 self.open_or_change(path, text).await?;
292 Ok(())
293 }
294
295 async fn request(&self, method: &str, params: Value, wait: Duration) -> Result<Value> {
296 let id = {
297 let mut next = self.next_id.lock().await;
298 let id = *next;
299 *next = next.saturating_add(1);
300 id
301 };
302 let (tx, rx) = oneshot::channel();
303 {
304 let mut pending = self.pending.lock().await;
305 pending.insert(id, tx);
306 }
307 let payload = json!({
308 "jsonrpc": "2.0",
309 "id": id,
310 "method": method,
311 "params": params,
312 });
313 if let Err(err) = send_message(&self.tx_outbound, &payload).await {
314 let mut pending = self.pending.lock().await;
315 pending.remove(&id);
316 return Err(err);
317 }
318 match timeout(wait, rx).await {
319 Ok(Ok(reply)) => {
320 if let Some(error) = reply.get("error") {
321 let message = error
322 .get("message")
323 .and_then(|v| v.as_str())
324 .unwrap_or("LSP request failed");
325 return Err(anyhow!("{message}"));
326 }
327 Ok(reply.get("result").cloned().unwrap_or(Value::Null))
328 }
329 Ok(Err(_)) => Err(anyhow!("LSP request channel closed")),
330 Err(_) => {
331 let mut pending = self.pending.lock().await;
332 pending.remove(&id);
333 Err(anyhow!("LSP request timed out for {method}"))
334 }
335 }
336 }
337
338 async fn shutdown(&self) {
339 let mut child = self.child.lock().await;
340 if let Some(mut c) = child.take() {
341 let _ = c.start_kill();
342 let _ = c.wait().await;
343 }
344 }
345 }
346
347 /// Send a JSON value as one Content-Length-framed JSON-RPC message.
348 async fn send_message(tx: &mpsc::Sender<Vec<u8>>, value: &Value) -> Result<()> {
349 let body = serde_json::to_vec(value).context("serialize LSP message")?;
350 let header = format!("Content-Length: {}\r\n\r\n", body.len());
351 let mut frame = Vec::with_capacity(header.len() + body.len());
352 frame.extend_from_slice(header.as_bytes());
353 frame.extend_from_slice(&body);
354 tx.send(frame)
355 .await
356 .map_err(|_| anyhow!("LSP outbound channel closed"))?;
357 Ok(())
358 }
359
360 /// Background task that drains the outbound queue and writes each frame to
361 /// the LSP server's stdin. Exits cleanly when the channel closes.
362 async fn writer_task(mut stdin: tokio::process::ChildStdin, mut rx: mpsc::Receiver<Vec<u8>>) {
363 while let Some(frame) = rx.recv().await {
364 if stdin.write_all(&frame).await.is_err() {
365 break;
366 }
367 if stdin.flush().await.is_err() {
368 break;
369 }
370 }
371 }
372
373 /// Background task that parses `Content-Length`-framed JSON-RPC frames from
374 /// the LSP server's stdout. Pushes each parsed JSON value to `tx`. Exits
375 /// when stdout closes or a frame is malformed (we choose to fail closed
376 /// rather than risk hanging).
377 async fn reader_task(mut stdout: tokio::process::ChildStdout, tx: mpsc::Sender<Value>) {
378 let mut buf: Vec<u8> = Vec::with_capacity(8 * 1024);
379 let mut tmp = [0u8; 4096];
380 loop {
381 let n = match stdout.read(&mut tmp).await {
382 Ok(0) => return,
383 Ok(n) => n,
384 Err(_) => return,
385 };
386 buf.extend_from_slice(&tmp[..n]);
387 // Try to parse as many frames as we can from the accumulated buffer.
388 while let Some((header_end, content_length)) = parse_header(&buf) {
389 if buf.len() < header_end + content_length {
390 break; // need more bytes
391 }
392 let body = &buf[header_end..header_end + content_length];
393 let parsed = serde_json::from_slice::<Value>(body).ok();
394 // Drop the consumed bytes regardless of parse result so a bad frame
395 // does not stall the loop.
396 buf.drain(..header_end + content_length);
397 if let Some(value) = parsed
398 && tx.send(value).await.is_err()
399 {
400 return;
401 }
402 }
403 }
404 }
405
406 /// Parse a JSON-RPC header block. Returns `Some((header_end, content_length))`
407 /// where `header_end` is the byte offset of the first body byte. The header
408 /// terminator is `\r\n\r\n`. We require a `Content-Length` header.
409 fn parse_header(buf: &[u8]) -> Option<(usize, usize)> {
410 let term = b"\r\n\r\n";
411 let pos = buf.windows(term.len()).position(|window| window == term)?;
412 let header = std::str::from_utf8(&buf[..pos]).ok()?;
413 let mut content_length: Option<usize> = None;
414 for line in header.split("\r\n") {
415 if let Some(rest) = line.strip_prefix("Content-Length:") {
416 content_length = rest.trim().parse::<usize>().ok();
417 }
418 }
419 content_length.map(|cl| (pos + term.len(), cl))
420 }
421
422 /// Background task that consumes inbound JSON values, classifies them as
423 /// notifications/responses, and routes accordingly.
424 async fn dispatcher_task(
425 mut rx: mpsc::Receiver<Value>,
426 tx_diag: mpsc::Sender<(PathBuf, Vec<Diagnostic>)>,
427 pending: Arc<AsyncMutex<HashMap<i64, oneshot::Sender<Value>>>>,
428 ) {
429 while let Some(value) = rx.recv().await {
430 // Notifications have a `method` and no `id`.
431 let method = value.get("method").and_then(|v| v.as_str());
432 if method == Some("textDocument/publishDiagnostics") {
433 if let Some((path, diags)) = parse_publish_diagnostics(&value) {
434 let _ = tx_diag.send((path, diags)).await;
435 }
436 continue;
437 }
438 // Replies have an `id` and a `result` or `error`.
439 if let Some(id) = value.get("id").and_then(|v| v.as_i64()) {
440 let mut map = pending.lock().await;
441 if let Some(slot) = map.remove(&id) {
442 let _ = slot.send(value);
443 }
444 }
445 }
446 }
447
448 /// Decode a `textDocument/publishDiagnostics` notification.
449 fn parse_publish_diagnostics(value: &Value) -> Option<(PathBuf, Vec<Diagnostic>)> {
450 let params = value.get("params")?;
451 let uri = params.get("uri")?.as_str()?;
452 let path = path_from_uri(uri)?;
453 let raw = params.get("diagnostics")?.as_array()?;
454 let mut out = Vec::with_capacity(raw.len());
455 for d in raw {
456 let range = d.get("range")?;
457 let start = range.get("start")?;
458 let line = start.get("line")?.as_u64()? as u32 + 1;
459 let column = start.get("character")?.as_u64()? as u32 + 1;
460 let severity = Severity::from_lsp(d.get("severity").and_then(|v| v.as_i64()))
461 .unwrap_or(Severity::Error);
462 let message = d
463 .get("message")
464 .and_then(|v| v.as_str())
465 .unwrap_or("")
466 .to_string();
467 out.push(Diagnostic {
468 line,
469 column,
470 severity,
471 message,
472 });
473 }
474 Some((path, out))
475 }
476
477 /// Convert a filesystem path to a `file://` URI. Best-effort — we do not
478 /// support Windows drive letters perfectly, but the LSP servers in our
479 /// registry accept percent-encoded paths well enough for the post-edit
480 /// diagnostics use case.
481 pub(crate) fn uri_from_path(path: &Path) -> String {
482 let canonical = path.canonicalize().unwrap_or_else(|_| path.to_path_buf());
483 let s = canonical.to_string_lossy();
484 if s.starts_with('/') {
485 format!("file://{s}")
486 } else {
487 format!("file:///{}", s.trim_start_matches('/'))
488 }
489 }
490
491 /// Inverse of [`uri_from_path`]. Returns `None` when the URI is not a `file://`.
492 fn path_from_uri(uri: &str) -> Option<PathBuf> {
493 let stripped = uri.strip_prefix("file://")?;
494 Some(PathBuf::from(stripped))
495 }
496
497 #[cfg(test)]
498 mod tests {
499 use super::*;
500
501 #[test]
502 fn parses_lsp_header() {
503 let frame = b"Content-Length: 5\r\n\r\nhello";
504 let (end, len) = parse_header(frame).expect("header parses");
505 assert_eq!(end, 21);
506 assert_eq!(len, 5);
507 }
508
509 #[test]
510 fn parse_header_returns_none_when_truncated() {
511 let frame = b"Content-Length: 5\r\nMissingTerm";
512 assert!(parse_header(frame).is_none());
513 }
514
515 #[test]
516 fn parses_publish_diagnostics_payload() {
517 let payload = json!({
518 "jsonrpc": "2.0",
519 "method": "textDocument/publishDiagnostics",
520 "params": {
521 "uri": "file:///tmp/foo.rs",
522 "diagnostics": [
523 {
524 "range": {
525 "start": { "line": 11, "character": 7 },
526 "end": { "line": 11, "character": 8 }
527 },
528 "severity": 1,
529 "message": "missing semicolon"
530 }
531 ]
532 }
533 });
534 let (path, diags) = parse_publish_diagnostics(&payload).expect("parses");
535 assert_eq!(path, PathBuf::from("/tmp/foo.rs"));
536 assert_eq!(diags.len(), 1);
537 assert_eq!(diags[0].line, 12);
538 assert_eq!(diags[0].column, 8);
539 assert_eq!(diags[0].severity, Severity::Error);
540 assert_eq!(diags[0].message, "missing semicolon");
541 }
542
543 #[test]
544 fn round_trips_uri_path() {
545 let path = PathBuf::from("/tmp/example/foo.rs");
546 let uri = format!("file://{}", path.display());
547 assert_eq!(path_from_uri(&uri), Some(path));
548 }
549 }
550
550 lines RUST