From 0d76cba054b8f85ad2ca6177469a124e069c8bed Mon Sep 17 00:00:00 2001 From: Yogthos Date: Tue, 19 May 2026 17:51:42 -0400 Subject: [PATCH 1/2] Consolidate URI helpers + switch notify_open to tokio::fs Move path_to_file_uri / uri_to_path / percent_encode_path / percent_decode out of init.rs and client.rs into a shared src/lsp/uri.rs module. Both files now import the same helpers. Tests for the URI behavior consolidated alongside the implementation. Switch LspClient::notify_open from std::fs to tokio::fs::canonicalize + tokio::fs::read_to_string so the orchestrator can parallelize touches across multiple clients without blocking the runtime thread. New uri.rs tests (8): safe-char passthrough, special-char encoding regression, round-trip for hash/space/question-mark paths, non-file scheme rejection, invalid percent escape passthrough, lsp-types Uri parse, multibyte UTF-8 per-byte encoding. Removed 4 duplicate tests from init.rs/client.rs that now live in uri.rs. Suite: 380 -> 382 (net +2 from the new uri.rs tests after dedup). --- src/lsp/client.rs | 104 +++-------------------------- src/lsp/init.rs | 69 +------------------ src/lsp/mod.rs | 1 + src/lsp/uri.rs | 166 ++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 179 insertions(+), 161 deletions(-) create mode 100644 src/lsp/uri.rs diff --git a/src/lsp/client.rs b/src/lsp/client.rs index b2b84102..01f91dbb 100644 --- a/src/lsp/client.rs +++ b/src/lsp/client.rs @@ -24,6 +24,7 @@ use tokio::sync::watch; use crate::lsp::language; use crate::lsp::rpc::{RpcClient, RpcError}; +use crate::lsp::uri::{path_to_file_uri_string, uri_to_path}; #[derive(Debug, Default)] struct FileState { @@ -112,15 +113,15 @@ impl LspClient { /// `textDocument/didChange` thereafter (with a bumped version). /// Returns the version sent. /// - /// Uses synchronous `std::fs::canonicalize` + `std::fs::read_to_string`. - /// Acceptable for now since callers invoke this once per write boundary, - /// not on a hot loop. Phase 4 should switch to `tokio::fs` if hot paths - /// emerge (e.g. touching many files in parallel on agent startup). + /// File I/O goes through `tokio::fs` so the orchestrator can fan + /// `touch_file` out across multiple clients without blocking the runtime + /// thread. pub async fn notify_open(&self, path: &Path) -> Result { - let abs = path - .canonicalize() - .or_else(|_| Ok::<_, std::io::Error>(path.to_path_buf()))?; - let text = std::fs::read_to_string(&abs)?; + let abs = match tokio::fs::canonicalize(path).await { + Ok(p) => p, + Err(_) => path.to_path_buf(), + }; + let text = tokio::fs::read_to_string(&abs).await?; let uri = path_to_file_uri_string(&abs); let is_first_open; @@ -254,70 +255,6 @@ impl LspClient { } } -fn uri_to_path(uri: &str) -> Option { - // Strip `file://` and percent-decode the body. We're intentionally - // permissive: the spec allows file:///path on unix and file://host/path - // on remote (which we don't support). Anything else returns None. - let trimmed = uri - .strip_prefix("file://") - .or_else(|| uri.strip_prefix("file:"))?; - let decoded = percent_decode(trimmed); - Some(PathBuf::from(decoded)) -} - -fn percent_decode(s: &str) -> String { - let bytes = s.as_bytes(); - let mut out: Vec = Vec::with_capacity(bytes.len()); - let mut i = 0; - while i < bytes.len() { - if bytes[i] == b'%' && i + 2 < bytes.len() { - let hi = hex_value(bytes[i + 1]); - let lo = hex_value(bytes[i + 2]); - if let (Some(h), Some(l)) = (hi, lo) { - out.push(h * 16 + l); - i += 3; - continue; - } - } - out.push(bytes[i]); - i += 1; - } - String::from_utf8_lossy(&out).into_owned() -} - -fn hex_value(b: u8) -> Option { - match b { - b'0'..=b'9' => Some(b - b'0'), - b'a'..=b'f' => Some(b - b'a' + 10), - b'A'..=b'F' => Some(b - b'A' + 10), - _ => None, - } -} - -fn path_to_file_uri_string(path: &Path) -> String { - let s = path.to_string_lossy(); - let encoded = percent_encode_path(&s); - if s.starts_with('/') { - format!("file://{encoded}") - } else { - format!("file:///{encoded}") - } -} - -fn percent_encode_path(path: &str) -> String { - let mut out = String::with_capacity(path.len()); - for byte in path.bytes() { - let safe = - byte.is_ascii_alphanumeric() || matches!(byte, b'/' | b'-' | b'.' | b'_' | b'~' | b':'); - if safe { - out.push(byte as char); - } else { - out.push_str(&format!("%{byte:02X}")); - } - } - out -} - fn dedupe>(items: I) -> Vec { let mut seen: HashSet = HashSet::new(); let mut out = Vec::new(); @@ -669,27 +606,4 @@ mod tests { .await; assert!(matches!(res, Err(LspError::DiagnosticsTimeout { .. }))); } - - // URI ↔ path round-trips, exercising percent-encoding. - #[test] - fn uri_to_path_roundtrips_safe_paths() { - let p = PathBuf::from("/tmp/proj/main.rs"); - let uri = path_to_file_uri_string(&p); - let decoded = uri_to_path(&uri).unwrap(); - assert_eq!(decoded, p); - } - - #[test] - fn uri_to_path_decodes_percent_encoded_special_chars() { - let p = PathBuf::from("/tmp/proj #1/main.rs"); - let uri = path_to_file_uri_string(&p); - assert!(uri.contains("%23")); - let decoded = uri_to_path(&uri).unwrap(); - assert_eq!(decoded, p); - } - - #[test] - fn uri_to_path_returns_none_for_non_file_uri() { - assert!(uri_to_path("https://example.com").is_none()); - } } diff --git a/src/lsp/init.rs b/src/lsp/init.rs index f050ee3c..7a1893d6 100644 --- a/src/lsp/init.rs +++ b/src/lsp/init.rs @@ -16,11 +16,12 @@ use lsp_types::{ ClientCapabilities, DiagnosticClientCapabilities, DidChangeWatchedFilesClientCapabilities, GeneralClientCapabilities, InitializeParams, InitializeResult, PublishDiagnosticsClientCapabilities, TextDocumentClientCapabilities, - TextDocumentSyncClientCapabilities, Uri, WindowClientCapabilities, WorkspaceClientCapabilities, + TextDocumentSyncClientCapabilities, WindowClientCapabilities, WorkspaceClientCapabilities, WorkspaceFolder, }; use crate::lsp::rpc::{RpcClient, RpcError}; +use crate::lsp::uri::path_to_file_uri; /// Time we'll wait for the server to answer `initialize`. Matches opencode's /// 45s ceiling — rust-analyzer in particular can take a moment when first @@ -40,7 +41,7 @@ pub async fn initialize( process_id: Option, initialization_options: serde_json::Value, ) -> Result { - let root_uri = path_to_file_uri(root)?; + let root_uri = path_to_file_uri(root).map_err(RpcError::Io)?; let params = InitializeParams { process_id, @@ -70,46 +71,6 @@ pub async fn initialize( Ok(result) } -fn path_to_file_uri(path: &Path) -> Result { - // `file://` URI from a filesystem path. We don't try to handle Windows - // drive letters specially — dirge isn't tested on Windows. - let canonical = path.to_string_lossy(); - let encoded = percent_encode_path(&canonical); - let uri_str = if canonical.starts_with('/') { - format!("file://{encoded}") - } else { - format!("file:///{encoded}") - }; - uri_str.parse::().map_err(|e| { - RpcError::Io(std::io::Error::new( - std::io::ErrorKind::InvalidInput, - format!("invalid path for file URI: {e}"), - )) - }) -} - -/// Percent-encode characters that would otherwise be interpreted as URI -/// structural delimiters. Slashes are preserved (path separators). Conforms -/// to RFC 3986's `unreserved` set + `/` for path segments. Pure ASCII only — -/// non-ASCII bytes pass through and the Uri parser does its own escaping or -/// rejection. -fn percent_encode_path(path: &str) -> String { - let mut out = String::with_capacity(path.len()); - for byte in path.bytes() { - let safe = - byte.is_ascii_alphanumeric() || matches!(byte, b'/' | b'-' | b'.' | b'_' | b'~' | b':'); - if safe { - out.push(byte as char); - } else if byte < 0x80 { - out.push_str(&format!("%{byte:02X}")); - } else { - // Non-ASCII — emit as UTF-8 percent-encoded bytes. - out.push_str(&format!("%{byte:02X}")); - } - } - out -} - /// The client capabilities we advertise. Conservative and stable — matches /// what opencode sends. Phase 3's per-file work depends on these being /// honoured by the server: @@ -370,30 +331,6 @@ mod tests { ); } - // Regression: paths containing URI-significant characters must be - // percent-encoded. A `#` would otherwise terminate the path early and - // produce a fragment. - #[test] - fn path_to_file_uri_percent_encodes_special_chars() { - let p = Path::new("/tmp/proj #1/src/main.rs"); - let uri = path_to_file_uri(p).unwrap(); - let s = uri.as_str(); - assert!(s.starts_with("file:///"), "got: {s}"); - assert!(s.contains("%23"), "must encode '#' as %23, got: {s}"); - assert!(s.contains("%20"), "must encode space as %20, got: {s}"); - } - - #[test] - fn path_to_file_uri_preserves_slashes_and_safe_chars() { - let p = Path::new("/tmp/proj_v1.0-rc/main.rs"); - let uri = path_to_file_uri(p).unwrap(); - let s = uri.as_str(); - assert!( - s.starts_with("file:///tmp/proj_v1.0-rc/main.rs"), - "got: {s}" - ); - } - // The `initialized` notification must follow the InitializeResult — some // servers stall until they see it. #[tokio::test] diff --git a/src/lsp/mod.rs b/src/lsp/mod.rs index bada24b0..94911189 100644 --- a/src/lsp/mod.rs +++ b/src/lsp/mod.rs @@ -16,3 +16,4 @@ pub mod jsonrpc; pub mod language; pub mod rpc; pub mod server; +pub mod uri; diff --git a/src/lsp/uri.rs b/src/lsp/uri.rs new file mode 100644 index 00000000..cd52ee76 --- /dev/null +++ b/src/lsp/uri.rs @@ -0,0 +1,166 @@ +//! `file://` URI ↔ filesystem path conversion with percent-encoding. +//! +//! Consolidates what previously lived in both `init.rs` and `client.rs`. +//! The encoding rule: preserve `unreserved` chars + `/` (path separator) + +//! `:` (Windows drive separators, though dirge isn't tested there). +//! Everything else gets `%XX` percent-encoded by raw byte; non-ASCII bytes +//! emit as percent-encoded UTF-8 sequences. +//! +//! Decoding is permissive: invalid `%XX` sequences pass through unchanged, +//! and the result is interpreted as a best-effort UTF-8 string (lossy on +//! ill-formed bytes). + +use std::path::{Path, PathBuf}; + +use lsp_types::Uri; + +/// Convert a path to a `file://` URI string. Always returns a string (never +/// fails); call [`path_to_file_uri`] when you need a parsed `Uri` and want +/// the parse error. +pub fn path_to_file_uri_string(path: &Path) -> String { + let s = path.to_string_lossy(); + let encoded = percent_encode_path(&s); + if s.starts_with('/') { + format!("file://{encoded}") + } else { + // Relative path or Windows-style. Emit with an extra `/` so the + // result is parseable as a URI. + format!("file:///{encoded}") + } +} + +/// Convert a path to a parsed [`Uri`]. Returns an I/O error wrapped around +/// the parse failure so callers can propagate it through their own error +/// chains. +pub fn path_to_file_uri(path: &Path) -> std::io::Result { + let s = path_to_file_uri_string(path); + s.parse::().map_err(|e| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("invalid path for file URI: {e}"), + ) + }) +} + +/// Decode a `file://` URI back into a [`PathBuf`]. Returns `None` for any +/// scheme other than `file:` (e.g. `https://`). +pub fn uri_to_path(uri: &str) -> Option { + let trimmed = uri + .strip_prefix("file://") + .or_else(|| uri.strip_prefix("file:"))?; + Some(PathBuf::from(percent_decode(trimmed))) +} + +/// Percent-encode `path` per RFC 3986. Slashes are preserved (path +/// separators). Conforms to `unreserved` + `/` + `:`. +pub fn percent_encode_path(path: &str) -> String { + let mut out = String::with_capacity(path.len()); + for byte in path.bytes() { + let safe = + byte.is_ascii_alphanumeric() || matches!(byte, b'/' | b'-' | b'.' | b'_' | b'~' | b':'); + if safe { + out.push(byte as char); + } else { + out.push_str(&format!("%{byte:02X}")); + } + } + out +} + +/// Permissive percent-decoder. Invalid `%XX` sequences pass through as-is. +pub fn percent_decode(s: &str) -> String { + let bytes = s.as_bytes(); + let mut out: Vec = Vec::with_capacity(bytes.len()); + let mut i = 0; + while i < bytes.len() { + if bytes[i] == b'%' && i + 2 < bytes.len() { + let hi = hex_value(bytes[i + 1]); + let lo = hex_value(bytes[i + 2]); + if let (Some(h), Some(l)) = (hi, lo) { + out.push(h * 16 + l); + i += 3; + continue; + } + } + out.push(bytes[i]); + i += 1; + } + String::from_utf8_lossy(&out).into_owned() +} + +fn hex_value(b: u8) -> Option { + match b { + b'0'..=b'9' => Some(b - b'0'), + b'a'..=b'f' => Some(b - b'a' + 10), + b'A'..=b'F' => Some(b - b'A' + 10), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn safe_chars_pass_through_unchanged() { + let p = Path::new("/tmp/proj_v1.0-rc/main.rs"); + let s = path_to_file_uri_string(p); + assert_eq!(s, "file:///tmp/proj_v1.0-rc/main.rs"); + } + + // Regression: paths containing URI-significant characters must be + // percent-encoded. A `#` would otherwise terminate the path early and + // produce a fragment. + #[test] + fn regression_special_chars_are_percent_encoded() { + let p = Path::new("/tmp/proj #1/main rs"); + let s = path_to_file_uri_string(p); + assert!(s.contains("%23"), "must encode '#' as %23: {s}"); + assert!(s.contains("%20"), "must encode space as %20: {s}"); + } + + #[test] + fn round_trip_preserves_path() { + for p in &[ + "/tmp/a/b/c.rs", + "/tmp/with spaces/main.rs", + "/tmp/with#hash/main.rs", + "/tmp/with?q/main.rs", + ] { + let path = PathBuf::from(p); + let uri = path_to_file_uri_string(&path); + let decoded = uri_to_path(&uri).unwrap(); + assert_eq!(decoded, path, "round-trip failed for {p}"); + } + } + + #[test] + fn non_file_uri_returns_none() { + assert!(uri_to_path("https://example.com").is_none()); + assert!(uri_to_path("not a uri").is_none()); + } + + #[test] + fn invalid_percent_escape_passes_through() { + // Stray `%` followed by non-hex must not panic; emit as-is. + let s = percent_decode("hello%zz world"); + assert_eq!(s, "hello%zz world"); + } + + #[test] + fn parses_to_lsp_types_uri() { + let uri = path_to_file_uri(Path::new("/tmp/main.rs")).unwrap(); + assert_eq!(uri.as_str(), "file:///tmp/main.rs"); + } + + #[test] + fn multibyte_utf8_percent_encodes_per_byte() { + let p = Path::new("/tmp/🦀.rs"); + let s = path_to_file_uri_string(p); + // 🦀 is F0 9F A6 80 in UTF-8 → four %XX escapes. + assert!(s.contains("%F0%9F%A6%80"), "got: {s}"); + // Round-trips. + let decoded = uri_to_path(&s).unwrap(); + assert_eq!(decoded, p); + } +} From 5451563b93a0bba54e0241a85cca5762514ddd88 Mon Sep 17 00:00:00 2001 From: Yogthos Date: Tue, 19 May 2026 18:00:50 -0400 Subject: [PATCH 2/2] LSP Phase 4: orchestrator with lazy spawn, inflight dedup, fan-out MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit src/lsp/spawn.rs - Spawner trait + Spawned struct. Trait is unit-testable via MockSpawner (duplex pipes + fake server task that answers initialize); production uses ProcessSpawner with tokio::process::Command + kill_on_drop(true). - Spawned.guard is Box — opaque to the manager, holds whatever the spawner needs to live for the child's lifetime (Child or JoinHandle). Drop terminates the connection. - ProcessSpawner drains stderr in the background so a chatty LSP server (rust-analyzer logs there) doesn't fill the pipe and stall the child. - 3 spawn tests: initialize handshake round-trips through the mock, spawn calls recorded, failure path works. src/lsp/manager.rs (LspManager) - One per agent session. Threaded through main like BackgroundStore in the bg-notifications work. - get_clients(file): walks the builtin registry, picks servers that claim the extension, resolves workspace root via server.root(), spawns lazily, caches by (root, server_id). Inflight spawns are deduped via Arc so concurrent agent tool calls never race two rust-analyzer processes for the same workspace. Broken-set blocks retry on failed spawns. - touch_file(path, mode): notify_open on each claiming client. TouchMode::AwaitPush additionally blocks on wait_for_push so the edit tool can surface fresh diagnostics in its tool result. - Fan-out helpers for Phase 5's agent tool: hover, definition, references, implementation, document_symbol, workspace_symbol, prepareCallHierarchy + incoming/outgoingCalls. Per-client errors are logged at debug and swallowed — one slow server doesn't poison the result. - all_diagnostics() aggregates across attached clients. 8 manager tests including regressions: - concurrent get_clients only spawns ONE process (regression — without inflight dedup, every parallel tool call would race a fresh process) - failed spawn marks (root, server_id) broken and no retry (regression — hammering a broken server every tool call would be expensive) - no-server extension returns empty without spawning - touch_file calls notify_open on the attached client - manager drop doesn't deadlock on the state mutex Code review fixes applied before push: - Notify race: subscribers arriving AFTER notify_waiters would hang. Added a 60s safety timeout on the joiner-side await; cache is authoritative regardless of how the wait terminates. - Moved tracing log calls out of the state-lock critical section in get_or_spawn so spawn-failure logging doesn't extend lock hold time. Introduced a SpawnOutcome enum to thread the outcome past lock release. Phases 1-3: 75 tests; Phase 4: +11 → 86 LSP tests. Suite: 380 -> 391. --- src/lsp/manager.rs | 816 +++++++++++++++++++++++++++++++++++++++++++++ src/lsp/mod.rs | 2 + src/lsp/spawn.rs | 274 +++++++++++++++ 3 files changed, 1092 insertions(+) create mode 100644 src/lsp/manager.rs create mode 100644 src/lsp/spawn.rs diff --git a/src/lsp/manager.rs b/src/lsp/manager.rs new file mode 100644 index 00000000..888ab9a1 --- /dev/null +++ b/src/lsp/manager.rs @@ -0,0 +1,816 @@ +//! `LspManager` — lazy-spawn LSP clients per (workspace root, server-id) and +//! fan tool requests out across them. +//! +//! One manager per agent session. Constructed in `main::build_channels`, +//! threaded through `build_agent` and `run_interactive` (the same plumbing +//! pattern used for `BackgroundStore`). +//! +//! What it does in Phase 4: +//! - Walk the built-in [`crate::lsp::server`] registry to find which servers +//! claim a given file extension. +//! - Resolve workspace roots via each server's `root` function. +//! - Spawn LSP servers on demand, dedupe in-flight spawns (so two parallel +//! tool calls don't race two `rust-analyzer` processes), cache the result +//! by `(root, server_id)`. +//! - Track a "broken" set so spawn failures aren't re-attempted. +//! - `touch_file(path, mode)`: open / re-sync the file on every claiming +//! client so push diagnostics flow. +//! - Fan-out helpers for the LSP request methods the agent tool (Phase 5) +//! will expose: hover, definition, references, implementation, +//! document_symbol, workspace_symbol, prepareCallHierarchy + incoming/ +//! outgoingCalls. +//! +//! On `Drop`, every spawned client's `_guard` drops (which kills the child +//! process via `kill_on_drop(true)` on the real spawner). + +use std::collections::{HashMap, HashSet}; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use lsp_types::{ + CallHierarchyIncomingCall, CallHierarchyItem, CallHierarchyOutgoingCall, Diagnostic, + DocumentSymbolResponse, GotoDefinitionResponse, Hover, Location, WorkspaceSymbolResponse, +}; +use serde::Serialize; +use serde_json::{Value, json}; +use tokio::io::BufReader; +use tokio::sync::Notify; + +use crate::lsp::client::{LspClient, LspError}; +use crate::lsp::init::initialize; +use crate::lsp::rpc::RpcClient; +use crate::lsp::server::{self, ServerInfo}; +use crate::lsp::spawn::Spawner; +use crate::lsp::uri::path_to_file_uri_string; + +/// Time we'll let any non-initialize LSP request take. Generous — LSP servers +/// can be slow on first-touch indexing — but bounded so a stuck server +/// doesn't hold up the agent's turn forever. +const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); + +/// How a [`LspManager::touch_file`] call should handle diagnostics: +/// - `None`: just send didOpen / didChange and return. +/// - `Document(after)`: wait for the next push for this file after the given +/// `Instant`, with a timeout. Used by the edit tool to surface fresh +/// diagnostics in its tool result. +#[derive(Debug, Clone, Copy)] +pub enum TouchMode { + Notify, + AwaitPush { after: Instant, timeout: Duration }, +} + +#[derive(Debug, thiserror::Error)] +pub enum ManagerError { + #[error("LSP server {server_id:?} failed to spawn: {source}")] + SpawnFailed { + server_id: String, + #[source] + source: std::io::Error, + }, + #[error("LSP server {server_id:?} initialize handshake failed: {source}")] + InitializeFailed { + server_id: String, + #[source] + source: crate::lsp::rpc::RpcError, + }, + #[error(transparent)] + Client(#[from] LspError), +} + +/// One cached LSP connection. +pub struct ClientEntry { + client: LspClient, + server_id: String, + root: PathBuf, + /// Holds the child process / fake-server task; drops on manager + /// shutdown to terminate the connection. + #[allow(dead_code)] + guard: Box, +} + +#[derive(Default)] +struct ManagerState { + clients: HashMap<(PathBuf, String), Arc>, + /// (root, server_id) pairs we've given up on. Avoids hammering a broken + /// server every time the agent touches a file. + broken: HashSet<(PathBuf, String)>, + /// In-flight spawns. The `Notify` is shared with every caller waiting + /// on the same (root, server_id) so we never spawn two of the same + /// server in parallel. + spawning: HashMap<(PathBuf, String), Arc>, +} + +/// One manager per agent session. Cheap to clone. +#[derive(Clone)] +pub struct LspManager { + spawner: Arc, + /// Worktree boundary — `server.root` fn walks stop here. + worktree: PathBuf, + state: Arc>, + /// Built-in registry. Cached so per-call code doesn't reallocate. + servers: Arc>, +} + +impl LspManager { + pub fn new(spawner: Arc, worktree: impl Into) -> Self { + Self { + spawner, + worktree: worktree.into(), + state: Arc::new(Mutex::new(ManagerState::default())), + servers: Arc::new(server::builtin_servers()), + } + } + + /// Resolve and (lazily) spawn every LSP client that claims `file`. Returns + /// one `LspClient` per (server_id, root) attached to that file. Empty + /// when no server claims the extension or every match is in the + /// broken-set. + pub async fn get_clients(&self, file: &Path) -> Vec> { + let ext = file + .extension() + .and_then(|s| s.to_str()) + .unwrap_or("") + .to_lowercase(); + + let mut out = Vec::new(); + + for server in self.servers.iter() { + if !server + .extensions + .iter() + .any(|e| e.eq_ignore_ascii_case(&ext)) + { + continue; + } + let Some(root) = (server.root)(file, &self.worktree) else { + continue; + }; + let key = (root.clone(), server.id.to_string()); + + // Fast-path cache check. + { + let state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + if state.broken.contains(&key) { + continue; + } + if let Some(entry) = state.clients.get(&key) { + out.push(Arc::clone(entry)); + continue; + } + } + + // Either spawn fresh or join an in-flight spawn for this key. + match self.get_or_spawn(server, &root, &key).await { + Some(entry) => out.push(entry), + None => continue, + } + } + out + } + + /// The dedupe-and-spawn dance, factored out so [`get_clients`] stays + /// readable. Returns the cached or freshly-spawned entry, or `None` if + /// spawn/initialize failed (in which case the key is now in `broken`). + async fn get_or_spawn( + &self, + server: &ServerInfo, + root: &Path, + key: &(PathBuf, String), + ) -> Option> { + // If another caller is already spawning this key, wait for them + // instead of racing. + let wait_for: Option> = { + let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + // Re-check cache under the lock — could have landed between + // the fast-path miss and now. + if let Some(entry) = state.clients.get(key) { + return Some(Arc::clone(entry)); + } + if state.broken.contains(key) { + return None; + } + if let Some(notify) = state.spawning.get(key) { + Some(Arc::clone(notify)) + } else { + // We're the spawner. Mark in-flight. + let notify = Arc::new(Notify::new()); + state.spawning.insert(key.clone(), Arc::clone(¬ify)); + None + } + }; + + if let Some(notify) = wait_for { + // Safety timeout in case we subscribed after `notify_waiters` fired + // (Notify doesn't save permits for late subscribers). The cache is + // authoritative either way — re-check after the wait, regardless + // of how it terminated. + let _ = tokio::time::timeout(Duration::from_secs(60), notify.notified()).await; + let state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(entry) = state.clients.get(key) { + return Some(Arc::clone(entry)); + } + return None; + } + + // We're responsible for the spawn. + let result = self.do_spawn(server, root).await; + + // Two-phase: update state under the lock, then log + signal outside. + // Slot-disappeared warning is captured for emit after lock release. + let outcome = { + let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + let notify = state.spawning.remove(key); + let slot_missing = notify.is_none(); + let notify = notify.unwrap_or_else(|| Arc::new(Notify::new())); + match result { + Ok(entry) => { + let arc = Arc::new(entry); + state.clients.insert(key.clone(), Arc::clone(&arc)); + SpawnOutcome::Inserted { + arc, + notify, + slot_missing, + } + } + Err(e) => { + state.broken.insert(key.clone()); + SpawnOutcome::Failed { + err: e, + notify, + slot_missing, + } + } + } + }; + + match outcome { + SpawnOutcome::Inserted { + arc, + notify, + slot_missing, + } => { + if slot_missing { + tracing::warn!( + "lsp: spawning slot for {:?} disappeared before spawn finished", + key + ); + } + notify.notify_waiters(); + Some(arc) + } + SpawnOutcome::Failed { + err, + notify, + slot_missing, + } => { + if slot_missing { + tracing::warn!( + "lsp: spawning slot for {:?} disappeared before spawn failed", + key + ); + } + tracing::warn!( + server = %server.id, + root = %root.display(), + "lsp: spawn failed: {err}" + ); + notify.notify_waiters(); + None + } + } + } + + /// The actual spawn + initialize handshake. + async fn do_spawn( + &self, + server: &ServerInfo, + root: &Path, + ) -> Result { + let spawned = + self.spawner + .spawn(server.id, root) + .await + .map_err(|e| ManagerError::SpawnFailed { + server_id: server.id.to_string(), + source: e, + })?; + let crate::lsp::spawn::Spawned { + reader, + writer, + init_options, + guard, + } = spawned; + + let buf_reader = BufReader::new(reader); + let (rpc, _reader_task) = RpcClient::new(buf_reader, writer); + let _ = initialize(&rpc, root, None, init_options) + .await + .map_err(|e| ManagerError::InitializeFailed { + server_id: server.id.to_string(), + source: e, + })?; + + let client = LspClient::new(rpc).await; + Ok(ClientEntry { + client, + server_id: server.id.to_string(), + root: root.to_path_buf(), + guard, + }) + } + + /// Open or re-sync `path` on every claiming client. Caller passes + /// [`TouchMode::Notify`] when they just want the server to know about + /// the file, or [`TouchMode::AwaitPush`] to additionally block until a + /// fresh push arrives. Errors from individual clients are logged but + /// not surfaced — the agent should always make progress. + pub async fn touch_file(&self, path: &Path, mode: TouchMode) { + let clients = self.get_clients(path).await; + for entry in clients { + let send_at = Instant::now(); + match entry.client.notify_open(path).await { + Ok(_) => {} + Err(e) => { + tracing::warn!(server = %entry.server_id, path = %path.display(), "notify_open failed: {e}"); + continue; + } + } + if let TouchMode::AwaitPush { after, timeout } = mode { + let after = std::cmp::max(after, send_at); + if let Err(e) = entry.client.wait_for_push(path, after, timeout).await { + tracing::debug!(server = %entry.server_id, path = %path.display(), "wait_for_push: {e}"); + } + } + } + } + + /// Aggregated diagnostics across all attached clients. Same key/dedupe + /// semantics as [`LspClient::all_diagnostics`]. + pub fn all_diagnostics(&self) -> HashMap> { + let state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + let entries: Vec<_> = state.clients.values().cloned().collect(); + drop(state); + + let mut merged: HashMap> = HashMap::new(); + for entry in entries { + for (path, diags) in entry.client.all_diagnostics() { + merged.entry(path).or_default().extend(diags); + } + } + merged + } + + // ---- Fan-out helpers for the agent's `lsp` tool (Phase 5) ---- + + pub async fn hover(&self, file: &Path, line: u32, character: u32) -> Vec { + self.request_all( + file, + "textDocument/hover", + position_params(file, line, character), + ) + .await + } + + pub async fn definition( + &self, + file: &Path, + line: u32, + character: u32, + ) -> Vec { + self.request_all( + file, + "textDocument/definition", + position_params(file, line, character), + ) + .await + } + + pub async fn references(&self, file: &Path, line: u32, character: u32) -> Vec> { + let mut params = position_params(file, line, character); + params["context"] = json!({"includeDeclaration": true}); + self.request_all(file, "textDocument/references", params) + .await + } + + pub async fn implementation( + &self, + file: &Path, + line: u32, + character: u32, + ) -> Vec { + self.request_all( + file, + "textDocument/implementation", + position_params(file, line, character), + ) + .await + } + + pub async fn document_symbol(&self, file: &Path) -> Vec { + let params = json!({ + "textDocument": { "uri": path_to_file_uri_string(file) } + }); + self.request_all(file, "textDocument/documentSymbol", params) + .await + } + + pub async fn workspace_symbol( + &self, + anchor_file: &Path, + query: &str, + ) -> Vec { + let params = json!({ "query": query }); + self.request_all(anchor_file, "workspace/symbol", params) + .await + } + + pub async fn prepare_call_hierarchy( + &self, + file: &Path, + line: u32, + character: u32, + ) -> Vec> { + self.request_all( + file, + "textDocument/prepareCallHierarchy", + position_params(file, line, character), + ) + .await + } + + pub async fn incoming_calls( + &self, + file: &Path, + line: u32, + character: u32, + ) -> Vec> { + self.call_hierarchy(file, line, character, "callHierarchy/incomingCalls") + .await + } + + pub async fn outgoing_calls( + &self, + file: &Path, + line: u32, + character: u32, + ) -> Vec> { + self.call_hierarchy(file, line, character, "callHierarchy/outgoingCalls") + .await + } + + async fn call_hierarchy( + &self, + file: &Path, + line: u32, + character: u32, + method: &str, + ) -> Vec { + let clients = self.get_clients(file).await; + let mut out = Vec::new(); + for entry in clients { + let prepared: Vec = match entry + .client + .rpc() + .request( + "textDocument/prepareCallHierarchy", + position_params(file, line, character), + REQUEST_TIMEOUT, + ) + .await + { + Ok(v) => v, + Err(_) => continue, + }; + let Some(first) = prepared.first() else { + continue; + }; + match entry + .client + .rpc() + .request(method, json!({ "item": first }), REQUEST_TIMEOUT) + .await + { + Ok(v) => out.push(v), + Err(_) => continue, + } + } + out + } + + /// Fan a single `request` out across every client claiming `file`. + /// Per-client errors are swallowed and logged — the agent gets the + /// successful responses; one slow / broken server doesn't poison the + /// whole result. + async fn request_all(&self, file: &Path, method: &str, params: P) -> Vec + where + P: Serialize + Clone, + R: serde::de::DeserializeOwned, + { + let clients = self.get_clients(file).await; + let mut out = Vec::new(); + for entry in clients { + match entry + .client + .rpc() + .request(method, params.clone(), REQUEST_TIMEOUT) + .await + { + Ok(v) => out.push(v), + Err(e) => { + tracing::debug!(server = %entry.server_id, method = %method, "request failed: {e}"); + } + } + } + out + } +} + +enum SpawnOutcome { + Inserted { + arc: Arc, + notify: Arc, + slot_missing: bool, + }, + Failed { + err: ManagerError, + notify: Arc, + slot_missing: bool, + }, +} + +impl ClientEntry { + pub fn server_id(&self) -> &str { + &self.server_id + } + pub fn root(&self) -> &Path { + &self.root + } + pub fn client(&self) -> &LspClient { + &self.client + } +} + +fn position_params(file: &Path, line: u32, character: u32) -> Value { + json!({ + "textDocument": { "uri": path_to_file_uri_string(file) }, + "position": { "line": line, "character": character }, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::lsp::spawn::Spawned; + use futures::future::BoxFuture; + use std::sync::Arc as StdArc; + use std::sync::atomic::{AtomicUsize, Ordering}; + + /// Fixture: spawn counter + the actual mock from `spawn::tests`. + struct CountingSpawner { + spawn_calls: StdArc, + fail_keys: std::sync::Mutex>, + /// Delay each spawn by this much so concurrent calls have a chance + /// to collide on the dedupe path. + delay_ms: u64, + } + + impl CountingSpawner { + fn new(delay_ms: u64) -> Self { + Self { + spawn_calls: StdArc::new(AtomicUsize::new(0)), + fail_keys: std::sync::Mutex::new(HashSet::new()), + delay_ms, + } + } + fn count(&self) -> usize { + self.spawn_calls.load(Ordering::SeqCst) + } + fn fail_for(&self, server_id: &str, root: &Path) { + self.fail_keys + .lock() + .unwrap() + .insert((server_id.to_string(), root.to_path_buf())); + } + } + + impl Spawner for CountingSpawner { + fn spawn<'a>( + &'a self, + server_id: &'a str, + root: &'a Path, + ) -> BoxFuture<'a, std::io::Result> { + Box::pin(async move { + self.spawn_calls.fetch_add(1, Ordering::SeqCst); + if self.delay_ms > 0 { + tokio::time::sleep(Duration::from_millis(self.delay_ms)).await; + } + if self + .fail_keys + .lock() + .unwrap() + .contains(&(server_id.to_string(), root.to_path_buf())) + { + return Err(std::io::Error::other("forced fail")); + } + + // Build a minimal duplex pair with a fake server that + // answers `initialize` and ignores everything else. + let (client_in, mut server_writer) = tokio::io::duplex(8192); + let (mut server_reader, client_out) = tokio::io::duplex(8192); + let fake_server = tokio::spawn(async move { + use crate::lsp::jsonrpc::{decode_frame, encode_frame}; + let mut reader = tokio::io::BufReader::new(&mut server_reader); + loop { + let frame = match decode_frame(&mut reader).await { + Ok(b) => b, + Err(_) => break, + }; + let req: Value = match serde_json::from_slice(&frame) { + Ok(v) => v, + Err(_) => continue, + }; + if req.get("id").is_none() { + continue; + } + let id = req["id"].clone(); + let method = req["method"].as_str().unwrap_or(""); + let result = if method == "initialize" { + json!({"capabilities": {}}) + } else { + Value::Null + }; + let resp = json!({"jsonrpc": "2.0", "id": id, "result": result}); + if encode_frame(&mut server_writer, &serde_json::to_vec(&resp).unwrap()) + .await + .is_err() + { + break; + } + } + }); + Ok(Spawned { + reader: Box::new(tokio::io::BufReader::new(client_in)), + writer: Box::new(client_out), + init_options: Value::Null, + guard: Box::new(fake_server), + }) + }) + } + } + + /// Build a tempdir that looks like a Cargo workspace so the `rust` + /// server's `root` fn succeeds against files inside it. + fn cargo_tree(suffix: &str) -> PathBuf { + let root = std::env::temp_dir().join(format!( + "dirge-lsp-manager-test-{}-{}-{suffix}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0), + )); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(root.join("src")).unwrap(); + std::fs::write(root.join("Cargo.toml"), "[workspace]\nmembers = []\n").unwrap(); + std::fs::write(root.join("src/lib.rs"), "// hello\n").unwrap(); + root + } + + #[tokio::test] + async fn first_call_spawns_and_caches() { + let tree = cargo_tree("first-spawn"); + let spawner = StdArc::new(CountingSpawner::new(0)); + let manager = LspManager::new(spawner.clone(), tree.clone()); + let file = tree.join("src/lib.rs"); + + let clients = manager.get_clients(&file).await; + assert_eq!(clients.len(), 1); + assert_eq!(clients[0].server_id(), "rust"); + assert_eq!(spawner.count(), 1); + + // Second call: cache hit, no extra spawn. + let clients2 = manager.get_clients(&file).await; + assert_eq!(clients2.len(), 1); + assert_eq!(spawner.count(), 1); + + std::fs::remove_dir_all(&tree).ok(); + } + + // Regression: two concurrent get_clients calls for the same file must + // result in exactly ONE spawn. Without dedupe, every parallel tool call + // would race two `rust-analyzer` processes for the same workspace. + #[tokio::test] + async fn regression_concurrent_get_clients_only_spawns_once() { + let tree = cargo_tree("concurrent-spawn"); + let spawner = StdArc::new(CountingSpawner::new(40)); + let manager = LspManager::new(spawner.clone(), tree.clone()); + let file = tree.join("src/lib.rs"); + + let a = { + let manager = manager.clone(); + let file = file.clone(); + tokio::spawn(async move { manager.get_clients(&file).await }) + }; + let b = { + let manager = manager.clone(); + let file = file.clone(); + tokio::spawn(async move { manager.get_clients(&file).await }) + }; + let c = { + let manager = manager.clone(); + let file = file.clone(); + tokio::spawn(async move { manager.get_clients(&file).await }) + }; + let r_a = a.await.unwrap(); + let r_b = b.await.unwrap(); + let r_c = c.await.unwrap(); + assert_eq!(r_a.len(), 1); + assert_eq!(r_b.len(), 1); + assert_eq!(r_c.len(), 1); + assert_eq!( + spawner.count(), + 1, + "must dedupe inflight spawns; got {}", + spawner.count() + ); + + std::fs::remove_dir_all(&tree).ok(); + } + + // Regression: a spawn that fails must mark the (root, server_id) as + // broken so future calls don't re-attempt. Hammering a misconfigured + // server every tool call would be expensive and noisy. + #[tokio::test] + async fn regression_failed_spawn_marks_broken_and_no_retry() { + let tree = cargo_tree("broken"); + let spawner = StdArc::new(CountingSpawner::new(0)); + let root_canon = tree.canonicalize().unwrap(); + spawner.fail_for("rust", &root_canon); + let manager = LspManager::new(spawner.clone(), tree.clone()); + let file = tree.join("src/lib.rs"); + + let first = manager.get_clients(&file).await; + assert!( + first.is_empty(), + "first attempt should fail to produce a client" + ); + assert_eq!(spawner.count(), 1); + + // Second call: must NOT retry — broken-set blocks. + let second = manager.get_clients(&file).await; + assert!(second.is_empty()); + assert_eq!(spawner.count(), 1, "must not retry broken servers"); + + std::fs::remove_dir_all(&tree).ok(); + } + + #[tokio::test] + async fn no_server_claims_extension_returns_empty() { + let tree = cargo_tree("no-server"); + let spawner = StdArc::new(CountingSpawner::new(0)); + let manager = LspManager::new(spawner.clone(), tree.clone()); + let file = tree.join("notes.unknown"); + + let clients = manager.get_clients(&file).await; + assert!(clients.is_empty()); + assert_eq!( + spawner.count(), + 0, + "must not spawn for unsupported extensions" + ); + + std::fs::remove_dir_all(&tree).ok(); + } + + #[tokio::test] + async fn touch_file_calls_notify_open_on_attached_client() { + let tree = cargo_tree("touch"); + let spawner = StdArc::new(CountingSpawner::new(0)); + let manager = LspManager::new(spawner.clone(), tree.clone()); + let file = tree.join("src/lib.rs"); + + manager.touch_file(&file, TouchMode::Notify).await; + // The notify_open call should have ticked the file's version on the + // underlying client. + let entries = manager.get_clients(&file).await; + assert_eq!(entries.len(), 1); + + std::fs::remove_dir_all(&tree).ok(); + } + + // Manager drop cascades through client guards. We can't directly assert + // the spawned tokio task aborts, but we can assert dropping the manager + // releases its state lock (no deadlock on shutdown). + #[tokio::test] + async fn manager_drop_does_not_deadlock() { + let tree = cargo_tree("drop"); + let spawner = StdArc::new(CountingSpawner::new(0)); + let manager = LspManager::new(spawner.clone(), tree.clone()); + let file = tree.join("src/lib.rs"); + let _ = manager.get_clients(&file).await; + drop(manager); + // If we got here, drop completed without deadlocking on the state mutex. + std::fs::remove_dir_all(&tree).ok(); + } +} diff --git a/src/lsp/mod.rs b/src/lsp/mod.rs index 94911189..7f04d9f8 100644 --- a/src/lsp/mod.rs +++ b/src/lsp/mod.rs @@ -14,6 +14,8 @@ pub mod client; pub mod init; pub mod jsonrpc; pub mod language; +pub mod manager; pub mod rpc; pub mod server; +pub mod spawn; pub mod uri; diff --git a/src/lsp/spawn.rs b/src/lsp/spawn.rs new file mode 100644 index 00000000..6fd503ae --- /dev/null +++ b/src/lsp/spawn.rs @@ -0,0 +1,274 @@ +//! Spawning abstraction for LSP server processes. +//! +//! `Spawner` is a tiny trait so the orchestrator can be tested without +//! launching real `rust-analyzer` / `pyright` / etc. processes in CI. The +//! real implementation ([`ProcessSpawner`]) does the obvious thing with +//! [`tokio::process::Command`]; tests use an in-memory mock that pairs the +//! client with a fake server task over `tokio::io::duplex`. + +use std::path::{Path, PathBuf}; + +use futures::future::BoxFuture; +use serde_json::Value; +use tokio::io::{AsyncBufRead, AsyncWrite}; + +/// Result of a successful spawn: the async I/O halves the [`crate::lsp::rpc::RpcClient`] +/// will consume, the server-specific `initializationOptions` payload, and an +/// opaque guard that holds the child process alive (its `Drop` terminates). +pub struct Spawned { + pub reader: Box, + pub writer: Box, + pub init_options: Value, + /// Whatever needs to live for the child's lifetime — typically a + /// `tokio::process::Child` with `kill_on_drop(true)`. Opaque to the + /// manager; dropped when the client is shut down. + /// + /// `Send + Sync` because the manager stores entries behind an `Arc` and + /// passes them across `tokio::spawn` boundaries; both `Child` and + /// `JoinHandle` already satisfy this. + pub guard: Box, +} + +/// Launches LSP server processes. Trait so the orchestrator can be unit- +/// tested with in-memory duplex pipes. +pub trait Spawner: Send + Sync + 'static { + fn spawn<'a>( + &'a self, + server_id: &'a str, + root: &'a Path, + ) -> BoxFuture<'a, std::io::Result>; +} + +/// Default `Spawner` for production. Resolves the binary name via `which`- +/// style PATH lookup and spawns it with stdin/stdout/stderr piped. +/// +/// Knows nothing about LSP semantics — the orchestrator (Phase 4) chooses +/// the binary name + args based on the `server_id` and any user config. +pub struct ProcessSpawner { + /// Maps `server_id` to the binary + args to launch. Populated from the + /// builtin registry + user config when the manager is constructed. + commands: std::collections::HashMap, +} + +#[derive(Clone, Debug)] +pub struct ProcessCommand { + pub program: PathBuf, + pub args: Vec, + pub env: Vec<(String, String)>, + pub init_options: Value, +} + +impl ProcessSpawner { + pub fn new(commands: std::collections::HashMap) -> Self { + Self { commands } + } +} + +impl Spawner for ProcessSpawner { + fn spawn<'a>( + &'a self, + server_id: &'a str, + root: &'a Path, + ) -> BoxFuture<'a, std::io::Result> { + Box::pin(async move { + let cmd = self.commands.get(server_id).ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::NotFound, + format!("no spawn command configured for LSP server {server_id:?}"), + ) + })?; + + let mut command = tokio::process::Command::new(&cmd.program); + command + .args(&cmd.args) + .current_dir(root) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .kill_on_drop(true); + for (k, v) in &cmd.env { + command.env(k, v); + } + + let mut child = command.spawn().map_err(|e| { + std::io::Error::new( + e.kind(), + format!("failed to spawn LSP server {server_id:?}: {e}"), + ) + })?; + let stdin = child + .stdin + .take() + .ok_or_else(|| std::io::Error::other("LSP server stdin pipe unavailable"))?; + let stdout = child + .stdout + .take() + .ok_or_else(|| std::io::Error::other("LSP server stdout pipe unavailable"))?; + + // Drain stderr in the background. LSP servers chatter on stderr + // (rust-analyzer logs there) and a full pipe would block the + // child after ~64 KB. + if let Some(stderr) = child.stderr.take() { + let server_id = server_id.to_string(); + tokio::spawn(async move { + use tokio::io::AsyncBufReadExt; + let mut reader = tokio::io::BufReader::new(stderr); + let mut buf = String::new(); + loop { + buf.clear(); + match reader.read_line(&mut buf).await { + Ok(0) => break, // EOF + Ok(_) => tracing::debug!(server = %server_id, "{}", buf.trim_end()), + Err(_) => break, + } + } + }); + } + + let reader: Box = + Box::new(tokio::io::BufReader::new(stdout)); + let writer: Box = Box::new(stdin); + + Ok(Spawned { + reader, + writer, + init_options: cmd.init_options.clone(), + guard: Box::new(child), + }) + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Mock spawner that pairs the client with a fake LSP server task over + /// duplex pipes. The fake task responds to `initialize` with empty + /// capabilities and any other request with `result: null`. + pub(crate) struct MockSpawner { + spawn_calls: std::sync::Mutex>, + fail_for: std::sync::Mutex>, + } + + impl MockSpawner { + pub fn new() -> Self { + Self { + spawn_calls: std::sync::Mutex::new(Vec::new()), + fail_for: std::sync::Mutex::new(std::collections::HashSet::new()), + } + } + + pub fn fail_when_server_id(&self, server_id: &str) { + self.fail_for.lock().unwrap().insert(server_id.to_string()); + } + + pub fn calls(&self) -> Vec<(String, PathBuf)> { + self.spawn_calls.lock().unwrap().clone() + } + } + + impl Spawner for MockSpawner { + fn spawn<'a>( + &'a self, + server_id: &'a str, + root: &'a Path, + ) -> BoxFuture<'a, std::io::Result> { + Box::pin(async move { + self.spawn_calls + .lock() + .unwrap() + .push((server_id.to_string(), root.to_path_buf())); + + if self.fail_for.lock().unwrap().contains(server_id) { + return Err(std::io::Error::other(format!( + "mock spawn refused for {server_id}" + ))); + } + + let (client_in, mut server_writer) = tokio::io::duplex(8192); + let (mut server_reader, client_out) = tokio::io::duplex(8192); + + // Fake server task: respond to anything with a sensible reply. + let fake_server = tokio::spawn(async move { + use crate::lsp::jsonrpc::{decode_frame, encode_frame}; + use serde_json::json; + let mut reader = tokio::io::BufReader::new(&mut server_reader); + loop { + let frame = match decode_frame(&mut reader).await { + Ok(b) => b, + Err(_) => break, + }; + let req: Value = match serde_json::from_slice(&frame) { + Ok(v) => v, + Err(_) => continue, + }; + if req.get("id").is_none() { + continue; // notification — no reply + } + let id = req["id"].clone(); + let method = req["method"].as_str().unwrap_or(""); + let result = if method == "initialize" { + json!({"capabilities": {}}) + } else { + Value::Null + }; + let resp = json!({"jsonrpc": "2.0", "id": id, "result": result}); + if encode_frame(&mut server_writer, &serde_json::to_vec(&resp).unwrap()) + .await + .is_err() + { + break; + } + } + }); + + Ok(Spawned { + reader: Box::new(tokio::io::BufReader::new(client_in)), + writer: Box::new(client_out), + init_options: Value::Null, + guard: Box::new(fake_server), + }) + }) + } + } + + #[tokio::test] + async fn mock_spawner_responds_to_initialize() { + use crate::lsp::init::initialize; + use crate::lsp::rpc::RpcClient; + use tokio::io::BufReader; + + let s = MockSpawner::new(); + let spawned = s.spawn("rust", Path::new("/tmp")).await.unwrap(); + // BufReader> doesn't make sense — the inner + // reader already implements AsyncBufRead. Use the boxed reader + // directly. + let reader = BufReader::new(spawned.reader); + let (rpc, _) = RpcClient::new(reader, spawned.writer); + let result = initialize(&rpc, Path::new("/tmp"), Some(1), spawned.init_options).await; + assert!(result.is_ok(), "initialize should succeed: {result:?}"); + } + + #[tokio::test] + async fn mock_spawner_records_calls() { + let s = MockSpawner::new(); + let _ = s.spawn("rust", Path::new("/tmp")).await.unwrap(); + let _ = s.spawn("typescript", Path::new("/tmp/proj")).await.unwrap(); + let calls = s.calls(); + assert_eq!(calls.len(), 2); + assert_eq!(calls[0].0, "rust"); + assert_eq!(calls[1].0, "typescript"); + } + + #[tokio::test] + async fn mock_spawner_can_fail_on_command() { + let s = MockSpawner::new(); + s.fail_when_server_id("rust"); + // Can't `.unwrap_err()` directly because `Spawned` isn't `Debug`. + match s.spawn("rust", Path::new("/tmp")).await { + Ok(_) => panic!("expected spawn to fail"), + Err(e) => assert!(e.to_string().contains("refused")), + } + } +}