use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use crate::error::{Error, Result};
pub const LSP_MAX_MESSAGE_BYTES: usize = 16 * 1024 * 1024;
pub const DEFAULT_LSP_MAX_DIAGNOSTICS: usize = 20;
pub const DEFAULT_LSP_TIMEOUT_SECS: u64 = 5;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LspServerSpec {
pub command: String,
pub args: Vec<String>,
pub extensions: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct DiagnosticEntry {
severity: &'static str,
line: u32,
character: u32,
message: String,
}
struct LspIo {
stdin: tokio::process::ChildStdin,
stdout: BufReader<tokio::process::ChildStdout>,
next_id: i64,
opened: HashMap<String, i64>,
initialized: bool,
}
#[derive(Debug)]
pub struct LspClient {
name: String,
child: std::sync::Mutex<tokio::process::Child>,
io: tokio::sync::Mutex<LspIo>,
}
impl std::fmt::Debug for LspIo {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("LspIo")
.field("next_id", &self.next_id)
.field("opened", &self.opened.keys().collect::<Vec<_>>())
.field("initialized", &self.initialized)
.finish()
}
}
async fn write_message(stdin: &mut tokio::process::ChildStdin, val: &Value) -> Result<()> {
let body = serde_json::to_vec(val).map_err(|e| Error::tool("lsp", format!("encode: {e}")))?;
let header = format!("Content-Length: {}\r\n\r\n", body.len());
stdin
.write_all(header.as_bytes())
.await
.map_err(|e| Error::tool("lsp", format!("write: {e}")))?;
stdin
.write_all(&body)
.await
.map_err(|e| Error::tool("lsp", format!("write: {e}")))?;
stdin
.flush()
.await
.map_err(|e| Error::tool("lsp", format!("flush: {e}")))?;
Ok(())
}
async fn read_message(stdout: &mut BufReader<tokio::process::ChildStdout>) -> Result<Value> {
let mut content_length: Option<usize> = None;
loop {
let mut line = String::new();
let n = stdout
.read_line(&mut line)
.await
.map_err(|e| Error::tool("lsp", format!("read: {e}")))?;
if n == 0 {
return Err(Error::tool("lsp", "server closed stdout (eof)"));
}
let trimmed = line.trim_end_matches(['\r', '\n']);
if trimmed.is_empty() {
break; }
if let Some(v) = trimmed.strip_prefix("Content-Length:") {
content_length = v.trim().parse().ok();
}
}
let len = content_length
.ok_or_else(|| Error::tool("lsp", "message missing Content-Length header"))?;
if len > LSP_MAX_MESSAGE_BYTES {
return Err(Error::tool(
"lsp",
format!("message too large ({len} bytes) — refusing to buffer"),
));
}
let mut buf = vec![0u8; len];
stdout
.read_exact(&mut buf)
.await
.map_err(|e| Error::tool("lsp", format!("read body: {e}")))?;
serde_json::from_slice(&buf).map_err(|e| Error::tool("lsp", format!("decode: {e}")))
}
fn path_to_uri(path: &Path) -> String {
let s = path.to_string_lossy().replace('\\', "/");
if let Some(stripped) = s.strip_prefix('/') {
format!("file:///{stripped}")
} else {
format!("file:///{s}")
}
}
fn language_id_for(path: &Path) -> &'static str {
match path
.extension()
.and_then(|e| e.to_str())
.unwrap_or_default()
.to_ascii_lowercase()
.as_str()
{
"rs" => "rust",
"py" => "python",
"js" | "mjs" | "cjs" => "javascript",
"jsx" => "javascriptreact",
"ts" | "mts" | "cts" => "typescript",
"tsx" => "typescriptreact",
"go" => "go",
"rb" => "ruby",
"java" => "java",
"c" | "h" => "c",
"cpp" | "cc" | "cxx" | "hpp" => "cpp",
"cs" => "csharp",
"json" => "json",
"toml" => "toml",
"yaml" | "yml" => "yaml",
"md" => "markdown",
"sh" | "bash" => "shellscript",
_ => "plaintext",
}
}
#[cfg(unix)]
pub(crate) fn kill_process_group(pid: u32) {
unsafe {
libc::kill(-(pid as libc::pid_t), libc::SIGKILL);
}
}
fn severity_label(sev: Option<i64>) -> &'static str {
match sev {
Some(1) => "error",
Some(2) => "warning",
Some(3) => "information",
Some(4) => "hint",
_ => "diagnostic",
}
}
fn is_publish_diagnostics_for(msg: &Value, uri: &str) -> bool {
msg.get("method").and_then(|m| m.as_str()) == Some("textDocument/publishDiagnostics")
&& msg
.get("params")
.and_then(|p| p.get("uri"))
.and_then(|u| u.as_str())
== Some(uri)
}
fn diagnostics_from_message(msg: &Value, cap: usize) -> (usize, Vec<DiagnosticEntry>) {
let Some(arr) = msg
.get("params")
.and_then(|p| p.get("diagnostics"))
.and_then(|d| d.as_array())
else {
return (0, Vec::new());
};
let total = arr.len();
let entries = arr
.iter()
.take(cap)
.map(|d| DiagnosticEntry {
severity: severity_label(d.get("severity").and_then(|s| s.as_i64())),
line: d
.get("range")
.and_then(|r| r.get("start"))
.and_then(|s| s.get("line"))
.and_then(|v| v.as_u64())
.unwrap_or(0) as u32,
character: d
.get("range")
.and_then(|r| r.get("start"))
.and_then(|s| s.get("character"))
.and_then(|v| v.as_u64())
.unwrap_or(0) as u32,
message: d
.get("message")
.and_then(|m| m.as_str())
.unwrap_or_default()
.to_string(),
})
.collect();
(total, entries)
}
const MAX_DIAGNOSTIC_MESSAGE_CHARS: usize = 400;
fn format_diagnostics(
server: &str,
path: &Path,
total: usize,
entries: &[DiagnosticEntry],
) -> String {
let mut out = format!(
"LSP diagnostics ({server}) for {}: {total} issue(s)",
path.display()
);
for d in entries {
let mut msg = d.message.clone();
if msg.chars().count() > MAX_DIAGNOSTIC_MESSAGE_CHARS {
msg = msg.chars().take(MAX_DIAGNOSTIC_MESSAGE_CHARS).collect();
msg.push('\u{2026}');
}
out.push_str(&format!(
"\n {}:{}: {}: {}",
d.line + 1,
d.character + 1,
d.severity,
msg
));
}
if total > entries.len() {
out.push_str(&format!(
"\n ... and {} more diagnostic(s) not shown",
total - entries.len()
));
}
out
}
impl LspClient {
async fn spawn(name: &str, spec: &LspServerSpec) -> Result<Arc<LspClient>> {
let mut cmd = tokio::process::Command::new(&spec.command);
cmd.args(&spec.args)
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::null())
.kill_on_drop(true);
#[cfg(unix)]
cmd.process_group(0);
let mut child = cmd
.spawn()
.map_err(|e| Error::tool("lsp", format!("spawn {}: {e}", spec.command)))?;
let stdin = child
.stdin
.take()
.ok_or_else(|| Error::tool("lsp", "no stdin"))?;
let stdout = BufReader::new(
child
.stdout
.take()
.ok_or_else(|| Error::tool("lsp", "no stdout"))?,
);
Ok(Arc::new(LspClient {
name: name.to_string(),
child: std::sync::Mutex::new(child),
io: tokio::sync::Mutex::new(LspIo {
stdin,
stdout,
next_id: 0,
opened: HashMap::new(),
initialized: false,
}),
}))
}
async fn open_or_change_and_diagnose(
&self,
root: &Path,
uri: &str,
text: &str,
language_id: &str,
timeout: Duration,
cap: usize,
) -> Result<(usize, Vec<DiagnosticEntry>)> {
let mut io = self.io.lock().await;
if !io.initialized {
let id = io.next_id;
io.next_id += 1;
let req = json!({
"jsonrpc": "2.0",
"id": id,
"method": "initialize",
"params": {
"processId": std::process::id(),
"rootUri": path_to_uri(root),
"capabilities": {},
}
});
write_message(&mut io.stdin, &req).await?;
let deadline = tokio::time::Instant::now() + timeout;
loop {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
return Err(Error::tool(
"lsp",
"timed out waiting for initialize response",
));
}
let msg = tokio::time::timeout(remaining, read_message(&mut io.stdout))
.await
.map_err(|_| {
Error::tool("lsp", "timed out waiting for initialize response")
})??;
if msg.get("id").and_then(|v| v.as_i64()) == Some(id) {
break; }
}
let notif = json!({"jsonrpc": "2.0", "method": "initialized", "params": {}});
write_message(&mut io.stdin, ¬if).await?;
io.initialized = true;
}
let msg = match io.opened.get(uri).copied() {
Some(version) => {
let next = version + 1;
io.opened.insert(uri.to_string(), next);
json!({
"jsonrpc": "2.0",
"method": "textDocument/didChange",
"params": {
"textDocument": {"uri": uri, "version": next},
"contentChanges": [{"text": text}],
}
})
}
None => {
io.opened.insert(uri.to_string(), 1);
json!({
"jsonrpc": "2.0",
"method": "textDocument/didOpen",
"params": {
"textDocument": {
"uri": uri,
"languageId": language_id,
"version": 1,
"text": text,
}
}
})
}
};
write_message(&mut io.stdin, &msg).await?;
let deadline = tokio::time::Instant::now() + timeout;
loop {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
return Err(Error::tool("lsp", "timed out waiting for diagnostics"));
}
let msg = tokio::time::timeout(remaining, read_message(&mut io.stdout))
.await
.map_err(|_| Error::tool("lsp", "timed out waiting for diagnostics"))??;
if is_publish_diagnostics_for(&msg, uri) {
return Ok(diagnostics_from_message(&msg, cap));
}
}
}
fn kill(&self) {
if let Ok(mut child) = self.child.lock() {
#[cfg(unix)]
if let Some(pid) = child.id() {
kill_process_group(pid);
}
let _ = child.start_kill();
let _ = child.try_wait();
}
}
async fn shutdown(&self) {
let graceful = async {
let mut io = self.io.lock().await;
if !io.initialized {
return; }
let id = io.next_id;
io.next_id += 1;
let req = json!({"jsonrpc": "2.0", "id": id, "method": "shutdown", "params": null});
if write_message(&mut io.stdin, &req).await.is_ok() {
let _ =
tokio::time::timeout(Duration::from_millis(500), read_message(&mut io.stdout))
.await;
}
let notif = json!({"jsonrpc": "2.0", "method": "exit", "params": null});
let _ = write_message(&mut io.stdin, ¬if).await;
};
let _ = tokio::time::timeout(Duration::from_secs(2), graceful).await;
self.kill();
}
}
#[derive(Debug)]
pub struct LspManager {
servers: std::sync::Mutex<HashMap<String, Arc<LspClient>>>,
specs: Vec<(String, LspServerSpec)>,
root: PathBuf,
timeout: Duration,
max_diagnostics: usize,
}
impl LspManager {
pub fn new(
root: PathBuf,
specs: Vec<(String, LspServerSpec)>,
timeout: Duration,
max_diagnostics: usize,
) -> Self {
LspManager {
servers: std::sync::Mutex::new(HashMap::new()),
specs,
root,
timeout,
max_diagnostics,
}
}
fn spec_for_extension(&self, path: &Path) -> Option<(String, LspServerSpec)> {
let ext = path
.extension()
.and_then(|e| e.to_str())?
.to_ascii_lowercase();
self.specs
.iter()
.find(|(_, s)| {
s.extensions
.iter()
.any(|e| e.trim_start_matches('.').to_ascii_lowercase() == ext)
})
.cloned()
}
async fn get_or_spawn(&self, name: &str, spec: &LspServerSpec) -> Result<Arc<LspClient>> {
if let Some(existing) = self.servers.lock().unwrap().get(name).cloned() {
return Ok(existing);
}
let client = LspClient::spawn(name, spec).await?;
let mut map = self.servers.lock().unwrap();
let winner = map.entry(name.to_string()).or_insert(client).clone();
Ok(winner)
}
pub async fn diagnostics_after_write(&self, path: &Path) -> Option<String> {
if !crate::safe_path::contained(&self.root, path) {
return None; }
let (name, spec) = self.spec_for_extension(path)?;
let client = match self.get_or_spawn(&name, &spec).await {
Ok(c) => c,
Err(e) => {
tracing::warn!(server = %name, "lsp: failed to spawn/reuse server: {e}");
return None;
}
};
let text = match tokio::fs::read_to_string(path).await {
Ok(t) => t,
Err(_) => return None, };
let uri = path_to_uri(path);
let language_id = language_id_for(path);
match client
.open_or_change_and_diagnose(
&self.root,
&uri,
&text,
language_id,
self.timeout,
self.max_diagnostics,
)
.await
{
Ok((total, entries)) if total > 0 => {
Some(format_diagnostics(&client.name, path, total, &entries))
}
Ok(_) => None, Err(e) => {
tracing::warn!(server = %name, "lsp: diagnostics unavailable: {e}");
None
}
}
}
pub fn kill_all_sync(&self) {
let drained: Vec<Arc<LspClient>> = self
.servers
.lock()
.map(|mut m| m.drain().map(|(_, c)| c).collect())
.unwrap_or_default();
for client in drained {
client.kill();
}
}
pub async fn shutdown_all(&self) {
let drained: Vec<Arc<LspClient>> = self
.servers
.lock()
.map(|mut m| m.drain().map(|(_, c)| c).collect())
.unwrap_or_default();
for client in drained {
client.shutdown().await;
}
}
pub fn running_server_count(&self) -> usize {
self.servers.lock().map(|m| m.len()).unwrap_or(0)
}
}
#[derive(Debug)]
pub struct LspDiagnosticsObserver {
manager: Arc<LspManager>,
}
impl LspDiagnosticsObserver {
pub fn new(manager: Arc<LspManager>) -> Self {
LspDiagnosticsObserver { manager }
}
}
#[async_trait::async_trait]
impl crate::tools::WriteObserver for LspDiagnosticsObserver {
async fn before_write(&self, _path: &Path) {}
async fn after_write(&self, path: &Path) -> Option<String> {
self.manager.diagnostics_after_write(path).await
}
}
pub fn manager_for_config(config: &crate::Config) -> Option<Arc<LspManager>> {
if !config.lsp_enabled {
return None;
}
if config.lsp_servers.is_empty() {
eprintln!(
"warning: [capabilities.lsp] is enabled but no servers are configured under \
[capabilities.lsp.servers.<name>] — no language server will ever be launched"
);
}
Some(Arc::new(LspManager::new(
config.cwd.clone(),
config.lsp_servers.clone(),
Duration::from_secs(config.lsp_timeout_secs.max(1)),
config.lsp_max_diagnostics.max(1),
)))
}
#[cfg(test)]
mod tests {
use super::*;
const STUB_SERVER_PY: &str = r#"
import sys, json
def read_message():
headers = {}
while True:
line = sys.stdin.buffer.readline()
if not line:
return None
line = line.decode("utf-8", "replace").rstrip("\r\n")
if line == "":
break
if ":" in line:
k, v = line.split(":", 1)
headers[k.strip()] = v.strip()
length = int(headers.get("Content-Length", "0"))
body = sys.stdin.buffer.read(length)
return json.loads(body.decode("utf-8"))
def write_message(obj):
body = json.dumps(obj).encode("utf-8")
sys.stdout.buffer.write(("Content-Length: %d\r\n\r\n" % len(body)).encode("utf-8"))
sys.stdout.buffer.write(body)
sys.stdout.buffer.flush()
def diagnostics_for(text):
if "TODO" in text:
return [{
"range": {"start": {"line": 0, "character": 0}, "end": {"line": 0, "character": 4}},
"severity": 1,
"message": "found a TODO marker",
}]
return []
while True:
msg = read_message()
if msg is None:
break
method = msg.get("method")
if method == "initialize":
write_message({"jsonrpc": "2.0", "id": msg["id"], "result": {"capabilities": {}}})
elif method == "initialized":
pass
elif method in ("textDocument/didOpen", "textDocument/didChange"):
params = msg["params"]
if method == "textDocument/didOpen":
uri = params["textDocument"]["uri"]
text = params["textDocument"]["text"]
else:
uri = params["textDocument"]["uri"]
text = params["contentChanges"][0]["text"]
write_message({
"jsonrpc": "2.0",
"method": "textDocument/publishDiagnostics",
"params": {"uri": uri, "diagnostics": diagnostics_for(text)},
})
elif method == "shutdown":
write_message({"jsonrpc": "2.0", "id": msg["id"], "result": None})
elif method == "exit":
break
"#;
const SILENT_SERVER_PY: &str = r#"
import sys, time
while True:
line = sys.stdin.buffer.readline()
if not line:
break
time.sleep(3600)
"#;
const WORKER_STUB_SERVER_PY: &str = r#"
import sys, json, subprocess
def read_message():
headers = {}
while True:
line = sys.stdin.buffer.readline()
if not line:
return None
line = line.decode("utf-8", "replace").rstrip("\r\n")
if line == "":
break
if ":" in line:
k, v = line.split(":", 1)
headers[k.strip()] = v.strip()
length = int(headers.get("Content-Length", "0"))
body = sys.stdin.buffer.read(length)
return json.loads(body.decode("utf-8"))
def write_message(obj):
body = json.dumps(obj).encode("utf-8")
sys.stdout.buffer.write(("Content-Length: %d\r\n\r\n" % len(body)).encode("utf-8"))
sys.stdout.buffer.write(body)
sys.stdout.buffer.flush()
worker = subprocess.Popen(["sleep", "3600"])
with open(sys.argv[1], "w") as f:
f.write(str(worker.pid))
f.flush()
while True:
msg = read_message()
if msg is None:
break
method = msg.get("method")
if method == "initialize":
write_message({"jsonrpc": "2.0", "id": msg["id"], "result": {"capabilities": {}}})
elif method == "initialized":
pass
elif method in ("textDocument/didOpen", "textDocument/didChange"):
uri = msg["params"]["textDocument"]["uri"]
write_message({
"jsonrpc": "2.0",
"method": "textDocument/publishDiagnostics",
"params": {"uri": uri, "diagnostics": []},
})
elif method == "shutdown":
write_message({"jsonrpc": "2.0", "id": msg["id"], "result": None})
elif method == "exit":
break
"#;
fn write_stub(dir: &Path, name: &str, source: &str) -> PathBuf {
let path = dir.join(name);
std::fs::write(&path, source).unwrap();
path
}
fn python() -> Option<String> {
for candidate in ["python3", "python"] {
if std::process::Command::new(candidate)
.arg("--version")
.output()
.is_ok()
{
return Some(candidate.to_string());
}
}
None
}
fn tmp(tag: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!(
"supercode-lsp-test-{tag}-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
std::fs::create_dir_all(&dir).unwrap();
dir
}
fn stub_spec(script: &Path) -> LspServerSpec {
LspServerSpec {
command: python().expect("python3/python required for lsp tests"),
args: vec![script.to_string_lossy().into_owned()],
extensions: vec![".rs".to_string()],
}
}
#[tokio::test]
async fn manager_for_config_is_none_when_disabled_default_off_byte_identity() {
let config = crate::Config::builder().model("m").build();
assert!(!config.lsp_enabled);
assert!(manager_for_config(&config).is_none());
}
#[tokio::test]
async fn diagnostics_after_write_surfaces_a_diagnostic_from_the_stub_server() {
let Some(_py) = python() else {
eprintln!("skipping: no python3/python on PATH");
return;
};
let project = tmp("diag-hit");
let script = write_stub(&project, "stub_lsp.py", STUB_SERVER_PY);
let manager = LspManager::new(
project.clone(),
vec![("stub".to_string(), stub_spec(&script))],
Duration::from_secs(5),
DEFAULT_LSP_MAX_DIAGNOSTICS,
);
let file = project.join("has_todo.rs");
std::fs::write(&file, "// TODO fix this\nfn main() {}\n").unwrap();
let result = manager.diagnostics_after_write(&file).await;
let text = result.expect("expected diagnostics for a file containing TODO");
assert!(text.contains("stub"), "names the server: {text}");
assert!(text.contains("found a TODO marker"), "{text}");
manager.shutdown_all().await;
std::fs::remove_dir_all(&project).ok();
}
#[tokio::test]
async fn diagnostics_after_write_is_none_for_a_clean_file() {
let Some(_py) = python() else {
eprintln!("skipping: no python3/python on PATH");
return;
};
let project = tmp("diag-clean");
let script = write_stub(&project, "stub_lsp.py", STUB_SERVER_PY);
let manager = LspManager::new(
project.clone(),
vec![("stub".to_string(), stub_spec(&script))],
Duration::from_secs(5),
DEFAULT_LSP_MAX_DIAGNOSTICS,
);
let file = project.join("clean.rs");
std::fs::write(&file, "fn main() {}\n").unwrap();
let result = manager.diagnostics_after_write(&file).await;
assert!(
result.is_none(),
"a clean file must not produce a diagnostics annotation: {result:?}"
);
manager.shutdown_all().await;
std::fs::remove_dir_all(&project).ok();
}
#[tokio::test]
async fn diagnostics_after_write_is_none_for_an_unconfigured_extension() {
let Some(_py) = python() else {
eprintln!("skipping: no python3/python on PATH");
return;
};
let project = tmp("diag-unconfigured");
let script = write_stub(&project, "stub_lsp.py", STUB_SERVER_PY);
let manager = LspManager::new(
project.clone(),
vec![("stub".to_string(), stub_spec(&script))], Duration::from_secs(5),
DEFAULT_LSP_MAX_DIAGNOSTICS,
);
let file = project.join("has_todo.py");
std::fs::write(&file, "# TODO\n").unwrap();
let result = manager.diagnostics_after_write(&file).await;
assert!(result.is_none());
assert_eq!(
manager.running_server_count(),
0,
"an unconfigured extension must never spawn anything"
);
std::fs::remove_dir_all(&project).ok();
}
#[tokio::test]
async fn a_second_write_to_the_same_server_reuses_the_running_process() {
let Some(_py) = python() else {
eprintln!("skipping: no python3/python on PATH");
return;
};
let project = tmp("diag-reuse");
let script = write_stub(&project, "stub_lsp.py", STUB_SERVER_PY);
let manager = LspManager::new(
project.clone(),
vec![("stub".to_string(), stub_spec(&script))],
Duration::from_secs(5),
DEFAULT_LSP_MAX_DIAGNOSTICS,
);
let file = project.join("f.rs");
std::fs::write(&file, "fn main() {}\n").unwrap();
manager.diagnostics_after_write(&file).await;
assert_eq!(manager.running_server_count(), 1);
std::fs::write(&file, "// TODO\nfn main() {}\n").unwrap();
manager.diagnostics_after_write(&file).await;
assert_eq!(
manager.running_server_count(),
1,
"a second write to the same extension must REUSE the already-running server, not spawn a second one"
);
manager.shutdown_all().await;
std::fs::remove_dir_all(&project).ok();
}
#[tokio::test]
async fn a_silent_server_degrades_to_none_within_the_timeout_bound() {
let Some(_py) = python() else {
eprintln!("skipping: no python3/python on PATH");
return;
};
let project = tmp("diag-silent");
let script = write_stub(&project, "silent_lsp.py", SILENT_SERVER_PY);
let manager = LspManager::new(
project.clone(),
vec![("silent".to_string(), stub_spec(&script))],
Duration::from_millis(500),
DEFAULT_LSP_MAX_DIAGNOSTICS,
);
let file = project.join("f.rs");
std::fs::write(&file, "fn main() {}\n").unwrap();
let started = std::time::Instant::now();
let result = tokio::time::timeout(
Duration::from_secs(10),
manager.diagnostics_after_write(&file),
)
.await
.expect("must not hang past the configured lsp timeout");
assert!(result.is_none());
assert!(
started.elapsed() < Duration::from_secs(5),
"took {:?}, expected to bail out near the 500ms configured timeout",
started.elapsed()
);
manager.kill_all_sync();
std::fs::remove_dir_all(&project).ok();
}
#[tokio::test]
async fn kill_all_sync_actually_terminates_the_os_process() {
let Some(_py) = python() else {
eprintln!("skipping: no python3/python on PATH");
return;
};
let project = tmp("diag-kill");
let script = write_stub(&project, "stub_lsp.py", STUB_SERVER_PY);
let manager = LspManager::new(
project.clone(),
vec![("stub".to_string(), stub_spec(&script))],
Duration::from_secs(5),
DEFAULT_LSP_MAX_DIAGNOSTICS,
);
let file = project.join("f.rs");
std::fs::write(&file, "fn main() {}\n").unwrap();
manager.diagnostics_after_write(&file).await;
assert_eq!(manager.running_server_count(), 1);
let pid = {
let servers = manager.servers.lock().unwrap();
let client = servers.values().next().unwrap();
let id = client.child.lock().unwrap().id();
id
};
let pid = pid.expect("spawned child must have a pid before it's reaped");
manager.kill_all_sync();
assert_eq!(
manager.running_server_count(),
0,
"kill_all_sync must drain the manager's bookkeeping"
);
let mut still_alive = true;
for _ in 0..50 {
let alive = unsafe { libc::kill(pid as libc::pid_t, 0) == 0 };
if !alive {
still_alive = false;
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
assert!(
!still_alive,
"pid {pid} must be dead after kill_all_sync (no orphaned language server)"
);
std::fs::remove_dir_all(&project).ok();
}
#[cfg(unix)]
#[tokio::test]
async fn kill_all_sync_reaps_grandchild_worker_processes() {
let Some(py) = python() else {
eprintln!("skipping: no python3/python on PATH");
return;
};
let project = tmp("diag-grandchild");
let script = write_stub(&project, "worker_stub_lsp.py", WORKER_STUB_SERVER_PY);
let pidfile = project.join("worker.pid");
let spec = LspServerSpec {
command: py,
args: vec![
script.to_string_lossy().into_owned(),
pidfile.to_string_lossy().into_owned(),
],
extensions: vec![".rs".to_string()],
};
let manager = LspManager::new(
project.clone(),
vec![("worker".to_string(), spec)],
Duration::from_secs(5),
DEFAULT_LSP_MAX_DIAGNOSTICS,
);
let file = project.join("f.rs");
std::fs::write(&file, "fn main() {}\n").unwrap();
manager.diagnostics_after_write(&file).await;
assert_eq!(manager.running_server_count(), 1);
let mut grandchild_pid: Option<i32> = None;
for _ in 0..100 {
if let Ok(s) = std::fs::read_to_string(&pidfile) {
if let Ok(pid) = s.trim().parse::<i32>() {
grandchild_pid = Some(pid);
break;
}
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
let grandchild_pid =
grandchild_pid.expect("stub server must have recorded its worker grandchild's pid");
assert!(
unsafe { libc::kill(grandchild_pid, 0) == 0 },
"grandchild worker pid {grandchild_pid} must be alive before kill_all_sync"
);
manager.kill_all_sync();
let mut still_alive = true;
for _ in 0..100 {
let alive = unsafe { libc::kill(grandchild_pid, 0) == 0 };
if !alive {
still_alive = false;
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
assert!(
!still_alive,
"grandchild worker pid {grandchild_pid} must be dead after kill_all_sync — \
it must not orphan (P5-11 review repro)"
);
std::fs::remove_dir_all(&project).ok();
}
#[tokio::test]
async fn diagnostics_after_write_refuses_a_path_outside_the_root() {
let Some(_py) = python() else {
eprintln!("skipping: no python3/python on PATH");
return;
};
let project = tmp("diag-outside-project");
let outside = tmp("diag-outside-elsewhere");
let script = write_stub(&project, "stub_lsp.py", STUB_SERVER_PY);
let manager = LspManager::new(
project.clone(),
vec![("stub".to_string(), stub_spec(&script))],
Duration::from_secs(5),
DEFAULT_LSP_MAX_DIAGNOSTICS,
);
let victim = outside.join("victim.rs");
std::fs::write(&victim, "// TODO\nfn main() {}\n").unwrap();
let result = manager.diagnostics_after_write(&victim).await;
assert!(result.is_none());
assert_eq!(manager.running_server_count(), 0);
std::fs::remove_dir_all(&project).ok();
std::fs::remove_dir_all(&outside).ok();
}
#[tokio::test]
async fn lsp_diagnostics_observer_before_write_is_a_true_noop() {
let project = tmp("obs-noop");
let manager = Arc::new(LspManager::new(
project.clone(),
vec![],
Duration::from_secs(5),
DEFAULT_LSP_MAX_DIAGNOSTICS,
));
let observer = LspDiagnosticsObserver::new(manager);
<LspDiagnosticsObserver as crate::tools::WriteObserver>::before_write(
&observer,
&project.join("f.rs"),
)
.await;
std::fs::remove_dir_all(&project).ok();
}
}