supercode_harness/

lsp.rs

//! §2 module 28 `lsp` (COMPOSABLE-HARNESS-DESIGN.md line 478): "D1 LSP
//! diagnostics in edit path + query tool" — this module ships the
//! WEAKEST FORM that satisfies D1: server LIFECYCLE for a HANDFUL of
//! user-configured language servers, and diagnostics surfaced in the
//! edit/write TOOL RESULT via the shared D-5 write-path seam
//! ([`crate::tools::WriteObserver`], P5-9/P5-11).
//!
//! # Honest, deliberate gaps (design §4.4 "LSP-in-edit at fleet scale")
//! - **No auto-spawn/auto-download fleet.** opencode auto-provisions ~38
//!   language servers. This module only ever spawns a server the user
//!   EXPLICITLY configured under `[capabilities.lsp.servers.<name>]` — no
//!   network fetch, no bundled binaries, nothing runs that wasn't named in
//!   config.
//! - **No `/find/symbol` symbol-indexing query tool.** A real symbol index
//!   (workspace/symbol, textDocument/definition, ...) is D-4-sized work
//!   this module does not attempt — shipping a `lsp.query` tool that
//!   silently no-ops would be a worse outcome than not shipping it (build
//!   brief: "if you expose any query surface, it must work or not
//!   exist"), so none is exposed. `[capabilities.lsp]` carries exactly two
//!   real, wired knobs: `enabled` and `servers` (plus the bounds
//!   `max_diagnostics`/`timeout_secs`) — no `query`/`symbols` key is ever
//!   parsed, so there is no declared-but-dead knob for either gap.
//!
//! # Wire protocol — hand-rolled, no new dependency
//! LSP frames a JSON-RPC message behind a tiny HTTP-style header
//! (`Content-Length: N\r\n\r\n<N bytes of JSON>`). That framing is a dozen
//! lines over `tokio::io::AsyncBufReadExt`/`AsyncReadExt` — pulling in a
//! dedicated `lsp-types`/`lsp-server` crate for it would be the heavy,
//! over-built option for a module scoped to lifecycle + diagnostics over a
//! handful of servers (no symbol index, no code actions, no incremental
//! sync deltas — just `initialize`/`initialized`/`didOpen`/`didChange`/
//! `publishDiagnostics`/`shutdown`/`exit`), so `write_message`/
//! `read_message` below hand-roll it instead. `cargo deny check` has
//! nothing new to license-audit as a result.
//!
//! # Process lifecycle (no orphaned language servers, incl. grandchildren)
//! A server is a long-lived child process, spawned lazily
//! ([`LspManager::diagnostics_after_write`], on the first write to a file
//! extension it's configured for) and kept alive in [`LspManager`] for
//! reuse across writes. A real configured server (`rust-analyzer`,
//! `typescript-language-server`, `gopls`, ...) commonly spawns its OWN
//! persistent worker subprocesses (a proc-macro/build server, `tsserver`,
//! `go`, ...) — so `.kill_on_drop(true)`/`Child::start_kill` alone (which
//! only ever signal the ONE directly-tracked pid) are not enough; this is
//! the SAME grandchild-orphan class `crate::agent::kill_job_process_group`
//! was built to close for background shell jobs (P5-6), and the fix here
//! reuses that exact mechanism: `LspClient::spawn` puts the server in its
//! OWN process group (`Command::process_group(0)`, unix), and
//! `LspClient::kill` SIGKILLs the WHOLE group (`kill_process_group`),
//! not just the leader — `.kill_on_drop(true)` remains as a second,
//! independent backstop for the leader pid specifically. On non-unix
//! targets, no portable process-group primitive is wired up (same posture
//! as `kill_job_process_group`'s own `#[cfg(not(unix))]` arm) — this falls
//! back to the pre-fix direct-child-only kill, a documented residual, not
//! silently claimed fixed there.
//!
//! [`LspManager::kill_all_sync`] (this module's group-kill, above) is
//! called from `impl Drop for crate::Agent` — the ONLY production teardown
//! path today, provable/traceable rather than relying solely on
//! `kill_on_drop(true)`'s implicit runtime behavior. [`LspManager::shutdown_all`]
//! (a graceful LSP `shutdown`/`exit` handshake, letting a well-behaved
//! server reap its own children before this module force-kills the group)
//! is NOT wired into any automatic path — `Agent::run_loop` runs once PER
//! TURN, not once per session, so calling it there would tear down and
//! respawn a reused server every turn, defeating the "kept alive for reuse
//! across writes" design above; there is no separate session-level
//! clean-exit hook distinct from `Drop` in this codebase today. It remains
//! available as public API (exercised directly by this module's own tests)
//! for a caller that manages its own `Agent` lifecycle and wants to drain
//! gracefully before dropping it — but nothing calls it automatically, and
//! that is the honest, current state (not an aspirational claim about a
//! code path that doesn't exist).

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};

/// Hardening cap (same rationale as `crate::mcp::MCP_MAX_RESPONSE_BYTES`):
/// the largest single Content-Length-framed message this client will
/// buffer before treating the server as hostile/broken and erroring out —
/// bounds how much memory a misbehaving configured language server can
/// force this process to allocate for one message.
pub const LSP_MAX_MESSAGE_BYTES: usize = 16 * 1024 * 1024;

/// Default cap on the number of diagnostics rendered into a single tool
/// result (bounded-context requirement — build brief: "a flood mustn't
/// blow context"). Overridable via `[capabilities.lsp] max_diagnostics`.
pub const DEFAULT_LSP_MAX_DIAGNOSTICS: usize = 20;

/// Default wait for a configured server to publish diagnostics after a
/// `didOpen`/`didChange` before giving up gracefully (never blocking the
/// tool call indefinitely). Overridable via `[capabilities.lsp] timeout_secs`.
pub const DEFAULT_LSP_TIMEOUT_SECS: u64 = 5;

/// One `[capabilities.lsp.servers.<name>]` entry — a user-configured
/// language server this module is allowed to spawn. `command`/`args` are
/// config-borne code execution (D-10) — see `crate::configfile::sanitize_for_project`
/// / `crate::userconfig`'s project-strip, which refuses this table from an
/// untrusted project layer exactly like `hooks`/`mcp.servers`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LspServerSpec {
    /// The executable to spawn (searched on `PATH` like any `Command::new`).
    pub command: String,
    /// Extra arguments passed to `command`.
    pub args: Vec<String>,
    /// File extensions (with or without a leading `.`, matched case-
    /// insensitively) this server handles — a write to a matching path
    /// lazily spawns (or reuses) this server.
    pub extensions: Vec<String>,
}

/// One diagnostic surfaced from a server's `textDocument/publishDiagnostics`
/// notification — the small subset of the LSP `Diagnostic` shape this
/// module renders into a tool result (no `code`/`source`/`relatedInformation`
/// — weakest form).
#[derive(Debug, Clone, PartialEq, Eq)]
struct DiagnosticEntry {
    severity: &'static str,
    line: u32,
    character: u32,
    message: String,
}

/// Per-connection JSON-RPC-over-stdio state — bundled behind ONE
/// `tokio::sync::Mutex` (rather than separate locks for stdin/stdout) so a
/// full request/response (or notify+wait-for-push) cycle runs atomically:
/// two concurrent writes to files the SAME server handles can never
/// interleave their reads and steal each other's response/diagnostics.
struct LspIo {
    stdin: tokio::process::ChildStdin,
    stdout: BufReader<tokio::process::ChildStdout>,
    next_id: i64,
    /// uri -> last-sent document version (LSP full-text sync: `didOpen`
    /// sends version 1, every subsequent `didChange` increments it).
    opened: HashMap<String, i64>,
    /// Whether `initialize`/`initialized` has completed on this connection.
    initialized: bool,
}

/// A live connection to one configured language server — one spawned child
/// process, kept alive for reuse across writes to files it handles.
#[derive(Debug)]
pub struct LspClient {
    name: String,
    // Sync `Mutex` (not `tokio::sync::Mutex`): only ever touched via the
    // synchronous `start_kill`/`try_wait` (never awaited while held) — see
    // `crate::agent::kill_job_process_group`'s identical precedent, which
    // is exactly why this can be called from the non-async `impl Drop for
    // Agent::drop`.
    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()
    }
}

/// Write one JSON-RPC value with LSP's `Content-Length` header framing.
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(())
}

/// Read one JSON-RPC value framed with LSP's `Content-Length` header —
/// bounded by [`LSP_MAX_MESSAGE_BYTES`], and errors (rather than hangs) on
/// EOF (the server exited or closed its stdout).
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; // blank line ends the header block
        }
        if let Some(v) = trimmed.strip_prefix("Content-Length:") {
            content_length = v.trim().parse().ok();
        }
        // Any other header (e.g. `Content-Type:`) is read and ignored.
    }
    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}")))
}

/// `file://` URI for `path` — a minimal, deterministic encoding (no
/// percent-escaping beyond backslash normalization) sufficient for the
/// stdio-local servers this module targets; every server this module talks
/// to is a local child process reading the SAME literal path this process
/// resolved, so round-trip fidelity (not RFC 3986 completeness) is what
/// matters.
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}")
    }
}

/// Best-effort LSP `languageId` for `path`'s extension — covers the common
/// languages a configured server would plausibly handle; unrecognized
/// extensions fall back to `"plaintext"` (a server that cares can still use
/// the extension embedded in the uri).
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",
    }
}

/// SIGKILL an entire process group — the shared grandchild-orphan-fix
/// primitive, same mechanism as `crate::agent::kill_job_process_group`
/// (P5-6). `pid` must be a process-group LEADER's pid (i.e. the process was
/// spawned with `Command::process_group(0)`, making its pgid equal its own
/// pid) for `-(pid)` to address the whole group rather than just the
/// leader. Reused by `crate::formatters` for the same reason — see
/// `crate::formatters::run_formatter`'s timeout arm.
///
/// SAFETY: `libc::kill` with a negative pid is `killpg` — it only ever
/// sends a signal (never dereferences memory), so this is safe regardless
/// of whether the group is still alive; a group that already exited yields
/// `ESRCH`, a documented no-op, not an error worth surfacing.
#[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",
    }
}

/// `true` iff `msg` is a `textDocument/publishDiagnostics` notification for
/// `uri` specifically (a server may publish for OTHER files it opened as
/// part of project analysis — those are ignored, never mixed into this
/// write's result).
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)
}

/// Parse a `publishDiagnostics` notification's `diagnostics` array into
/// `(true_total_count, bounded_entries)` — `true_total_count` may exceed
/// `entries.len()` when the server reported more than `cap`, so the caller
/// can render an honest "N more not shown" instead of silently dropping
/// them.
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)
}

/// Max characters kept from a single diagnostic's message text — a
/// pathological server can't blow context with one enormous message
/// either, on top of the diagnostics-COUNT cap.
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 {
    /// Spawn `spec.command` and hold it open, uninitialized (the LSP
    /// `initialize` handshake happens lazily, under the SAME `io` lock as
    /// the first `didOpen`, in [`Self::open_or_change_and_diagnose`]).
    /// `.kill_on_drop(true)` mirrors `crate::mcp::McpClient::connect`'s own
    /// stdio precedent — reaps the server if this client is ever dropped
    /// without an explicit [`Self::kill`]/[`Self::shutdown`].
    ///
    /// `.process_group(0)` (unix) puts the server in its OWN new process
    /// group (pgid == its own pid) — see the module doc's "Process
    /// lifecycle" section and `crate::agent::kill_job_process_group` for
    /// why: it's what lets [`Self::kill`] SIGKILL the server's WORKER
    /// grandchildren (proc-macro/build servers, `tsserver`, `go`, ...) too,
    /// not just this one directly-tracked pid.
    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,
            }),
        }))
    }

    /// The full D1 cycle for one write: ensure `initialize`/`initialized`
    /// has happened once, send `didOpen` (first touch of this uri) or
    /// `didChange` (subsequent touches, full-text sync), then read
    /// messages until this uri's `publishDiagnostics` notification arrives
    /// or `timeout` elapses. Holds the `io` lock for the whole cycle —
    /// see [`LspIo`]'s doc comment for why that's required, not just
    /// convenient.
    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; // the initialize response — ignore any notification before it
                }
            }
            let notif = json!({"jsonrpc": "2.0", "method": "initialized", "params": {}});
            write_message(&mut io.stdin, &notif).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));
            }
            // Some other notification (or a server-initiated request this
            // weakest-form client doesn't answer) — ignore and keep
            // waiting for OUR diagnostics, bounded by `deadline` above.
        }
    }

    /// Real, synchronous OS-process kill — callable from the non-async
    /// `impl Drop for Agent` (same reasoning as
    /// `crate::agent::kill_job_process_group`). SIGKILLs the server's WHOLE
    /// process group (unix) — see [`Self::spawn`]'s `.process_group(0)` and
    /// `kill_process_group` — so worker grandchildren die too, not just
    /// this directly-tracked pid; falls back to the direct-child-only kill
    /// on non-unix. A no-op if the process already exited.
    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();
        }
    }

    /// Graceful shutdown: `shutdown` request, brief wait for its response,
    /// `exit` notification — then [`Self::kill`] unconditionally as a
    /// backstop (a server that ignores `exit` must not linger). Bounded to
    /// 2s total so a hung server can never block session teardown.
    async fn shutdown(&self) {
        let graceful = async {
            let mut io = self.io.lock().await;
            if !io.initialized {
                return; // never handshaked — nothing to shut down gracefully
            }
            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, &notif).await;
        };
        let _ = tokio::time::timeout(Duration::from_secs(2), graceful).await;
        self.kill();
    }
}

/// Session-scoped registry of configured language servers and the live
/// connections spawned so far — the [`crate::tools::WriteObserver`] this
/// module installs ([`LspDiagnosticsObserver`]) is a thin wrapper around a
/// shared `Arc<LspManager>`; `crate::agent::build_tool_context` keeps its
/// own `Arc` clone too, so `crate::Agent`'s `Drop` impl can reach
/// [`Self::kill_all_sync`] regardless of how many observer clones exist.
#[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 {
    /// Build a manager over `specs` (name -> server definition), rooted at
    /// `root` (used as the LSP `rootUri` and the containment floor every
    /// touched path is checked against via `crate::safe_path::contained`).
    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();
        // A concurrent write to another file this SAME server handles may
        // have raced this spawn and already inserted — keep whichever
        // landed first (the loser's freshly-spawned child is dropped here,
        // which reaps it via `kill_on_drop`, never leaked).
        let winner = map.entry(name.to_string()).or_insert(client).clone();
        Ok(winner)
    }

    /// D1: the write-path diagnostics hook — spawns (or reuses) the
    /// configured server for `path`'s extension, if any, sends
    /// `didOpen`/`didChange` with the file's CURRENT on-disk content (the
    /// caller — `LspDiagnosticsObserver::after_write` — only calls this
    /// once the write has completed, so this always reads the FINAL bytes,
    /// which for `[capabilities.formatters]` also on means the FORMATTED
    /// content, not the model's pre-format draft — see the observer
    /// ordering `crate::agent::build_tool_context` installs), and waits
    /// (bounded by `self.timeout`) for that file's diagnostics. Never
    /// fails the caller: every error (no configured server, spawn
    /// failure, protocol error, timeout) degrades to `None`, logged once
    /// via `tracing::warn!`.
    pub async fn diagnostics_after_write(&self, path: &Path) -> Option<String> {
        if !crate::safe_path::contained(&self.root, path) {
            return None; // out of this module's scope — never touch outside the project
        }
        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, // deleted/unreadable — nothing to diagnose
        };
        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, // clean file — no noise on every write
            Err(e) => {
                tracing::warn!(server = %name, "lsp: diagnostics unavailable: {e}");
                None
            }
        }
    }

    /// Real, synchronous, provable kill of every server this manager has
    /// spawned so far — called from `impl Drop for crate::Agent` (a
    /// non-async context, hence the sync signature). A no-op for any
    /// server already exited.
    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();
        }
    }

    /// Graceful async shutdown of every server this manager has spawned —
    /// `shutdown`/`exit` handshake per server, `kill_all_sync`-equivalent
    /// backstop applied per-client by `LspClient::shutdown` itself. Use
    /// this on a clean-exit path that can afford to `.await`; use
    /// [`Self::kill_all_sync`] from `Drop`.
    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;
        }
    }

    /// Test/observability hook: how many servers are currently live.
    pub fn running_server_count(&self) -> usize {
        self.servers.lock().map(|m| m.len()).unwrap_or(0)
    }
}

/// The [`crate::tools::WriteObserver`] `[capabilities.lsp]` installs —
/// `before_write` is a true no-op (LSP has nothing to capture before a
/// mutation); `after_write` delegates straight to
/// [`LspManager::diagnostics_after_write`].
#[derive(Debug)]
pub struct LspDiagnosticsObserver {
    manager: Arc<LspManager>,
}

impl LspDiagnosticsObserver {
    /// Wrap `manager` as a [`crate::tools::WriteObserver`].
    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
    }
}

/// Build the [`LspManager`] a fresh [`crate::Agent`] should install, given
/// a resolved [`crate::Config`] — called once, from
/// `crate::agent::build_tool_context`. `Config::lsp_enabled` is the ONE
/// gate: `false` (the default) returns `None` WITHOUT spawning anything —
/// the default-off byte-identity guarantee. `true` with an EMPTY
/// `Config::lsp_servers` still returns a (harmless, does-nothing) manager,
/// but warns once — an enabled module with no configured servers is very
/// likely a config mistake, not silent-by-design.
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),
    )))
}