aft/lsp/

manager.rs

use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use crossbeam_channel::{bounded, unbounded, Receiver, RecvTimeoutError, Sender, TrySendError};
use lsp_types::notification::{
    DidChangeTextDocument, DidChangeWatchedFiles, DidCloseTextDocument, DidOpenTextDocument,
};
use lsp_types::{
    DidChangeTextDocumentParams, DidChangeWatchedFilesParams, DidCloseTextDocumentParams,
    DidOpenTextDocumentParams, FileChangeType, FileEvent, TextDocumentContentChangeEvent,
    TextDocumentIdentifier, TextDocumentItem, VersionedTextDocumentIdentifier,
};

use crate::alert_state::AcceptedDiagnosticSnapshot;
use crate::config::Config;
use crate::lsp::child_registry::LspChildRegistry;
use crate::lsp::client::{LspClient, LspEvent, ReaderExitReap, ServerState};
use crate::lsp::diagnostics::{
    from_lsp_diagnostics, DiagnosticEntry, DiagnosticsStore, StoredDiagnostic,
};
use crate::lsp::document::DocumentStore;
use crate::lsp::position::{uri_for_path, uri_to_path};
use crate::lsp::pull_params::{
    AftDocumentDiagnosticParams, AftDocumentDiagnosticRequest, AftWorkspaceDiagnosticParams,
    AftWorkspaceDiagnosticRequest,
};
use crate::lsp::registry::{resolve_server_binary, servers_for_file, ServerDef, ServerKind};
use crate::lsp::roots::ServerKey;
use crate::lsp::LspError;
use crate::slog_error;
use crate::slog_info;

const STDERR_REASON_BYTES: usize = 2 * 1024;
/// The total grace period for draining every LSP client during process shutdown.
const LSP_SHUTDOWN_ALL_BUDGET: Duration = Duration::from_secs(5);

fn server_key_for_definition(
    def: &ServerDef,
    file_path: &Path,
    config: &Config,
) -> Option<ServerKey> {
    def.workspace_root_for_file_with_project_root(file_path, config.project_root.as_deref())
        .map(|root| ServerKey {
            kind: def.kind.clone(),
            root,
        })
}

fn server_key_sort(left: &ServerKey, right: &ServerKey) -> std::cmp::Ordering {
    left.kind
        .id_str()
        .cmp(right.kind.id_str())
        .then(left.root.cmp(&right.root))
}

/// Outcome of attempting to ensure a server is running for a single matching
/// `ServerDef`. Returned per matching server so the caller can report exactly
/// what happened to the user instead of collapsing all failures into "no
/// server".
#[derive(Debug, Clone)]
pub enum ServerAttemptResult {
    /// Server is running and ready to serve requests for this file.
    Ok { server_key: ServerKey },
    /// No workspace root was found by walking up from the file looking for
    /// any of the server's configured root markers.
    NoRootMarker { looked_for: Vec<String> },
    /// The server's binary could not be found on PATH (or override was
    /// missing/invalid).
    BinaryNotInstalled { binary: String },
    /// Binary was found but spawning or initializing the server failed.
    SpawnFailed { binary: String, reason: String },
}

/// One server's attempt to handle a file.
#[derive(Debug, Clone)]
pub struct ServerAttempt {
    /// Stable server identifier (kind ID, e.g. "pyright", "rust-analyzer").
    pub server_id: String,
    /// Server display name from the registry.
    pub server_name: String,
    pub result: ServerAttemptResult,
}

/// Aggregate outcome of `ensure_server_for_file_detailed`. Distinguishes:
/// - "No server registered for this file's extension" (`attempts.is_empty()`)
/// - "Servers registered but none could start" (`successful.is_empty()` but
///   `!attempts.is_empty()`)
/// - "At least one server is ready" (`!successful.is_empty()`)
#[derive(Debug, Clone, Default)]
pub struct EnsureServerOutcomes {
    /// Server keys that are now running and ready to serve requests.
    pub successful: Vec<ServerKey>,
    /// Per-server attempt records. Empty if no server is registered for the
    /// file's extension.
    pub attempts: Vec<ServerAttempt>,
}

/// Side-effect-free result of resolving the server kinds applicable to an
/// inspection root. The public record set is deliberately only `ServerKey`s;
/// the definitions retained internally are used later by the explicit start
/// step and never cause a process to be opened during resolution.
#[derive(Clone, Debug)]
pub struct ApplicableServerSnapshot {
    pub server_keys: Vec<ServerKey>,
    candidates: Vec<ApplicableServerCandidate>,
    producer_failures: Vec<ApplicableServerFailure>,
}

#[derive(Clone, Debug)]
struct ApplicableServerCandidate {
    key: ServerKey,
    definition: ServerDef,
    source_file: PathBuf,
}

#[derive(Clone, Debug)]
pub enum ApplicabilityResolutionError {
    RootUnreadable { root: PathBuf, reason: String },
    RequestDeadline { root: PathBuf },
}

#[derive(Clone, Debug)]
pub struct ApplicableServerFailure {
    pub server_key: ServerKey,
    pub result: ServerAttemptResult,
}

impl ApplicableServerFailure {
    pub fn reason(&self) -> String {
        self.result.failure_reason()
    }
}

#[derive(Clone, Debug, Default)]
pub struct ApplicableServerStartOutcomes {
    pub successful: Vec<ServerKey>,
    pub failures: Vec<ApplicableServerFailure>,
    /// The first producer whose startup could not complete before the inspect
    /// request's shared work deadline.
    pub deadline_exceeded: Option<ServerKey>,
}

impl ServerAttemptResult {
    pub fn failure_reason(&self) -> String {
        match self {
            Self::BinaryNotInstalled { binary } => format!("{binary} is unavailable"),
            Self::SpawnFailed { reason, .. } => reason.clone(),
            Self::NoRootMarker { looked_for } => {
                format!(
                    "no workspace root marker found (looked for {})",
                    looked_for.join(", ")
                )
            }
            Self::Ok { .. } => "server started successfully".to_string(),
        }
    }
}

impl EnsureServerOutcomes {
    /// True if no server in the registry matched this file's extension.
    pub fn no_server_registered(&self) -> bool {
        self.attempts.is_empty()
    }

    /// True when servers matched the file's extension but none actually apply
    /// to this project — i.e. nothing started and every attempt failed the root
    /// marker check (e.g. oxlint registered for `.ts` with no `.oxlintrc.json`).
    /// Distinct from `no_server_registered` (extension unsupported) and from a
    /// real outage (binary missing / spawn failed): a missing root marker is a
    /// filesystem fact that never changes mid-scan, so such a file will never
    /// produce diagnostics and must not be reported as "pending".
    pub fn only_inapplicable_root_markers(&self) -> bool {
        self.successful.is_empty()
            && !self.attempts.is_empty()
            && self
                .attempts
                .iter()
                .all(|attempt| matches!(attempt.result, ServerAttemptResult::NoRootMarker { .. }))
    }
}

/// Outcome of a post-edit diagnostics wait. Reports the per-server status
/// alongside the fresh diagnostics, so the response layer can build an
/// honest tri-state payload (`success: true` + `complete: bool` + named
/// gap fields per `crates/aft/src/protocol.rs`).
///
/// `diagnostics` only contains entries from servers that proved freshness.
/// `accepted_snapshots` preserves that same result per producer and document
/// version so alert consumers never infer a partition from flattened output.
/// Pre-edit cached entries and unversioned reports are NEVER included.
#[derive(Debug, Clone, Default)]
pub struct PostEditWaitOutcome {
    /// Complete, per-producer accepted snapshots. This includes a snapshot
    /// with an empty diagnostics list when a live server authoritatively found
    /// the edited document clean; pending, exited, warming, and unversioned
    /// producers do not appear here.
    pub accepted_snapshots: Vec<AcceptedDiagnosticSnapshot>,
    /// Authoritative diagnostics flattened only for the legacy response body.
    /// Consumers that preserve producer partitions must use
    /// `accepted_snapshots` instead.
    pub diagnostics: Vec<StoredDiagnostic>,
    /// Servers we expected to publish but didn't before the deadline.
    /// Reported to the agent via `pending_lsp_servers` so they understand
    /// the result is partial.
    pub pending_servers: Vec<ServerKey>,
    /// Servers whose process exited between notification and deadline.
    /// Reported separately so the agent knows the gap is unrecoverable
    /// without a server restart, not "wait longer."
    pub exited_servers: Vec<ServerKey>,
}

/// Pre-edit freshness snapshot for one server/file pair.
#[derive(Debug, Clone, Copy, Default)]
pub struct PreEditSnapshot {
    pub epoch: u64,
    pub document_version_at_capture: Option<i32>,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct StaleDiagnosticsMark {
    pub had_entries: bool,
    pub changed: bool,
}

pub fn post_edit_entry_is_fresh(
    entry: &DiagnosticEntry,
    target_version: i32,
    pre: PreEditSnapshot,
) -> bool {
    if entry.stale || entry.epoch <= pre.epoch {
        return false;
    }

    match entry.version {
        Some(version) => version >= target_version,
        // Unversioned publishDiagnostics payloads cannot prove which document
        // state they describe. Epoch advancement only proves arrival order; an
        // old analysis result can still arrive after our pre-snapshot. Treat as
        // pending/partial rather than fresh.
        None => false,
    }
}

impl PostEditWaitOutcome {
    /// True if every expected server reported a fresh result. False means
    /// the agent should treat the diagnostics as a partial picture.
    pub fn complete(&self) -> bool {
        self.pending_servers.is_empty() && self.exited_servers.is_empty()
    }
}

/// Per-server outcome of a `textDocument/diagnostic` (per-file pull) request.
#[derive(Debug, Clone)]
pub enum PullFileOutcome {
    /// Server returned a full report; diagnostics stored.
    Full { diagnostic_count: usize },
    /// Server returned `kind: "unchanged"` — cached diagnostics still valid.
    Unchanged,
    /// Server returned a partial-result token; we don't subscribe to streamed
    /// progress so the response is treated as a soft empty until the next pull.
    PartialNotSupported,
    /// Server doesn't advertise pull capability — caller should fall back to
    /// push diagnostics for this server.
    PullNotSupported,
    /// The pull request failed (timeout, server error, etc.).
    RequestFailed { reason: String },
}

/// Result of ensuring a document is open in every matching server.
#[derive(Debug, Clone, Default)]
pub struct EnsureFileOpenResult {
    pub server_keys: Vec<ServerKey>,
    /// Servers that received `textDocument/didOpen` during this call.
    pub newly_opened: Vec<ServerKey>,
}

impl EnsureFileOpenResult {
    pub fn is_empty(&self) -> bool {
        self.server_keys.is_empty()
    }
}

/// Result of `pull_file_diagnostics` for one matching server.
#[derive(Debug, Clone)]
pub struct PullFileResult {
    pub server_key: ServerKey,
    pub outcome: PullFileOutcome,
}

/// Result of `pull_workspace_diagnostics` for a single server.
#[derive(Debug, Clone)]
pub struct PullWorkspaceResult {
    pub server_key: ServerKey,
    /// Files for which a Full report was received and cached. Files that came
    /// back as `Unchanged` are NOT listed here because their cached entry was
    /// already authoritative.
    pub files_reported: Vec<PathBuf>,
    /// True if the server returned a full response within the timeout.
    pub complete: bool,
    /// True if we cancelled (request timed out before the server responded).
    pub cancelled: bool,
    /// True if the server advertised workspace pull support. When false, the
    /// other fields are empty and the caller should fall back to file-mode
    /// pull or to push semantics.
    pub supports_workspace: bool,
}

pub struct DrainedLspEvents {
    pub events: Vec<LspEvent>,
    pub diagnostics_changed: bool,
    /// Complete snapshots accepted from document-version-verified publishes
    /// emitted by already-live, quiescent producers. Runtime consumers must
    /// retain each producer's snapshot until they construct an
    /// `AcceptedObservation` that records those producer-specific results;
    /// flattening the snapshots would erase producer ownership.
    pub accepted_snapshots: Vec<AcceptedDiagnosticSnapshot>,
    pub has_more: bool,
}

/// State carried by an `AppContext` post-edit wait after it releases the
/// manager mutex. The raw receiver clone competes for each event exactly once;
/// the dedicated wake receiver covers the case where another drain path wins
/// that race and updates the manager before this waiter sees the raw event.
pub(crate) struct PostEditDiagnosticsWait {
    lookup_path: PathBuf,
    expected_versions: Vec<(ServerKey, i32)>,
    pre_snapshot: HashMap<ServerKey, PreEditSnapshot>,
    event_rx: Receiver<LspEvent>,
    wake_rx: Receiver<()>,
    waiter_id: u64,
    deadline: std::time::Instant,
    fresh: HashMap<ServerKey, Vec<StoredDiagnostic>>,
    exited: Vec<ServerKey>,
}

impl PostEditDiagnosticsWait {
    pub(crate) fn deadline_reached(&self) -> bool {
        std::time::Instant::now() >= self.deadline
    }

    pub(crate) fn next_event(&self) -> Option<LspEvent> {
        let remaining = self
            .deadline
            .saturating_duration_since(std::time::Instant::now());
        if remaining.is_zero() {
            return None;
        }

        crossbeam_channel::select! {
            recv(self.event_rx) -> event => event.ok(),
            recv(self.wake_rx) -> _ => None,
            default(remaining) => None,
        }
    }
}

impl IntoIterator for DrainedLspEvents {
    type Item = LspEvent;
    type IntoIter = std::vec::IntoIter<LspEvent>;

    fn into_iter(self) -> Self::IntoIter {
        self.events.into_iter()
    }
}

/// Result of one bounded all-client shutdown. The same counts are emitted in
/// the `lsp shutdown_all` summary log.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct LspShutdownAllOutcome {
    pub graceful: usize,
    pub forced: usize,
    pub elapsed: Duration,
}

pub struct LspManager {
    /// Active server instances, keyed by (ServerKind, workspace_root).
    clients: HashMap<ServerKey, LspClient>,
    /// Binary names for active server instances. Kept separate from
    /// `LspClient` so crash handling can report the installable binary name
    /// after a post-initialize process exit.
    server_binaries: HashMap<ServerKey, String>,
    /// Tracks opened documents and versions per active server.
    documents: HashMap<ServerKey, DocumentStore>,
    /// Stored publishDiagnostics payloads across all servers.
    diagnostics: DiagnosticsStore,
    /// Unified event channel — all server reader threads send here.
    event_tx: Sender<LspEvent>,
    event_rx: Receiver<LspEvent>,
    /// One-slot wake channels for post-edit waits that released the manager
    /// mutex. Drains notify them after handling an event, so a waiter cannot
    /// sleep through a diagnostics update consumed by another drain path.
    post_edit_waiters: HashMap<u64, Sender<()>>,
    next_post_edit_waiter_id: u64,
    /// Optional binary path overrides used by integration tests.
    binary_overrides: HashMap<ServerKind, PathBuf>,
    /// Last plugin-pushed LSP search paths. `None` keeps direct `LspManager`
    /// users compatible with the paths supplied on their `Config` argument.
    pushed_search_paths: Option<Vec<PathBuf>>,
    /// Extra env vars merged into every spawned LSP child. Used in tests to
    /// drive the fake server's behavioral variants (`AFT_FAKE_LSP_PULL=1`,
    /// `AFT_FAKE_LSP_WORKSPACE=1`, etc.). Production code does not set this.
    extra_env: HashMap<String, String>,
    /// Per-(kind,root) cache of spawn failures. Once a server fails to spawn
    /// for a workspace root, we remember why and skip subsequent attempts for
    /// the lifetime of this AFT process. Without this, every file open or
    /// didChange retries `spawn_server` and logs a fresh ERROR — visible as
    /// repeated `failed to spawn TypeScript Language Server: Could not find a
    /// valid TypeScript installation` lines per edit.
    ///
    /// Entries are NEVER evicted automatically. The expected recovery path is
    /// for the user to fix their environment (install the missing binary or
    /// add a `tsconfig.json` / `package.json` with the right dependency) and
    /// restart OpenCode/Pi, which spawns a fresh `aft` process with an empty
    /// cache. We deliberately don't auto-retry on file events: the failure
    /// modes we track here (binary not installed, init handshake failure)
    /// don't fix themselves at runtime.
    failed_spawns: HashMap<ServerKey, ServerAttemptResult>,
    /// Server/root pairs for which we already logged that watched-file
    /// notifications are skipped because the capability is absent.
    watched_file_skip_logged: HashSet<ServerKey>,
    /// The last watched-file routing decision, retained on Windows so a CI
    /// timeout can distinguish a skipped send from a delayed fake-server reply.
    #[cfg(windows)]
    last_watched_file_notification_trace: String,
    /// Tracks PIDs of spawned LSP child processes so the signal handler can
    /// kill them on SIGTERM/SIGINT before aft exits, preventing orphans.
    /// Defaults to empty; production wires this from `AppContext`.
    child_registry: LspChildRegistry,
}

impl LspManager {
    pub fn new() -> Self {
        let (event_tx, event_rx) = unbounded();
        Self {
            clients: HashMap::new(),
            server_binaries: HashMap::new(),
            documents: HashMap::new(),
            diagnostics: DiagnosticsStore::new(),
            event_tx,
            event_rx,
            post_edit_waiters: HashMap::new(),
            next_post_edit_waiter_id: 0,
            binary_overrides: HashMap::new(),
            pushed_search_paths: None,
            extra_env: HashMap::new(),
            failed_spawns: HashMap::new(),
            watched_file_skip_logged: HashSet::new(),
            #[cfg(windows)]
            last_watched_file_notification_trace: "no watched-file notification attempted"
                .to_string(),
            child_registry: LspChildRegistry::new(),
        }
    }

    /// Set the child-PID registry. Must be called before any servers spawn.
    pub fn set_child_registry(&mut self, registry: LspChildRegistry) {
        self.child_registry = registry;
    }

    /// For testing: set an extra environment variable that gets passed to
    /// every spawned LSP child process. Useful for driving fake-server
    /// behavioral variants in integration tests.
    pub fn set_extra_env(&mut self, key: &str, value: &str) {
        self.extra_env.insert(key.to_string(), value.to_string());
    }

    /// Count active LSP server instances.
    pub fn server_count(&self) -> usize {
        self.clients.len()
    }

    /// Estimate the per-server document metadata and diagnostics retained by
    /// the manager. LSP child-process memory is outside this process RSS and is
    /// not included.
    pub fn estimated_memory(&self) -> crate::memory::MemoryEstimate {
        let mut bytes = 0u64;
        let mut document_count = 0u64;
        for documents in self.documents.values() {
            let estimate = documents.estimated_memory();
            bytes = bytes.saturating_add(estimate.estimated_bytes.unwrap_or(0));
            document_count = document_count
                .saturating_add(estimate.counts.get("documents").copied().unwrap_or(0));
        }
        let diagnostics = self.diagnostics.estimated_memory();
        bytes = bytes.saturating_add(diagnostics.estimated_bytes.unwrap_or(0));
        crate::memory::MemoryEstimate::estimated(bytes)
            .count("servers", self.clients.len())
            .count("document_stores", self.documents.len())
            .count_u64("documents", document_count)
            .count_u64(
                "diagnostic_entries",
                diagnostics
                    .counts
                    .get("diagnostic_entries")
                    .copied()
                    .unwrap_or(0),
            )
            .count_u64(
                "diagnostics",
                diagnostics.counts.get("diagnostics").copied().unwrap_or(0),
            )
    }

    /// Apply the configured diagnostic LRU cap (the `lsp.diagnostic_cache_size`
    /// knob). 0 disables the cap. Called at construction so the documented
    /// config field actually takes effect instead of always using the default.
    pub fn set_diagnostic_capacity(&mut self, capacity: usize) {
        self.diagnostics.set_capacity(capacity);
    }

    /// For testing: override the binary for a server kind.
    pub fn override_binary(&mut self, kind: ServerKind, binary_path: PathBuf) {
        self.binary_overrides.insert(kind, binary_path);
    }

    /// Replace the plugin-managed binary search paths used by future lazy starts.
    /// Existing servers stay live; failed starts are forgotten when the paths
    /// change so a newly installed binary is retried without restarting AFT.
    pub fn set_search_paths(&mut self, paths: Vec<PathBuf>) -> bool {
        if self.pushed_search_paths.as_ref() == Some(&paths) {
            return false;
        }
        self.pushed_search_paths = Some(paths);
        self.clear_failed_spawns();
        true
    }

    /// Resolve every configured server applicable to a root without starting it.
    pub fn resolve_applicable_servers_for_root(
        &self,
        project_root: &Path,
        config: &Config,
    ) -> Result<ApplicableServerSnapshot, ApplicabilityResolutionError> {
        self.resolve_applicable_servers(project_root, None, config, None)
    }

    /// Resolve inspect producers within the request deadline. A scoped inspect
    /// may start rust-analyzer only for the Cargo workspace that owns each scope;
    /// an unrelated nested workspace must not gain a process merely because it
    /// lives elsewhere under the configured project root.
    pub fn resolve_applicable_servers_for_inspect(
        &self,
        project_root: &Path,
        scope_roots: Option<&[PathBuf]>,
        config: &Config,
        deadline: std::time::Instant,
    ) -> Result<ApplicableServerSnapshot, ApplicabilityResolutionError> {
        self.resolve_applicable_servers(project_root, scope_roots, config, Some(deadline))
    }

    fn resolve_applicable_servers(
        &self,
        project_root: &Path,
        scope_roots: Option<&[PathBuf]>,
        config: &Config,
        deadline: Option<std::time::Instant>,
    ) -> Result<ApplicableServerSnapshot, ApplicabilityResolutionError> {
        if !project_root.is_dir() {
            return Err(ApplicabilityResolutionError::RootUnreadable {
                root: project_root.to_path_buf(),
                reason: "project root is not a directory".to_string(),
            });
        }

        let mut candidates = HashMap::<ServerKey, ApplicableServerCandidate>::new();
        let mut producer_failures = HashMap::<ServerKey, ApplicableServerFailure>::new();
        // Prevent a disappearing child mount from making ReadDir::drop abort on ENXIO.
        let walker = ignore::WalkBuilder::new(project_root)
            .same_file_system(true)
            .standard_filters(true)
            .add_custom_ignore_filename(".aftignore")
            .filter_entry(|entry| {
                !matches!(
                    entry.file_name().to_string_lossy().as_ref(),
                    ".git" | "node_modules" | "target" | "dist" | "build" | ".next" | ".turbo"
                )
            })
            .build();

        for entry in walker {
            if deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) {
                return Err(ApplicabilityResolutionError::RequestDeadline {
                    root: project_root.to_path_buf(),
                });
            }
            let entry = entry.map_err(|error| ApplicabilityResolutionError::RootUnreadable {
                root: project_root.to_path_buf(),
                reason: error.to_string(),
            })?;
            if !entry
                .file_type()
                .is_some_and(|file_type| file_type.is_file())
            {
                continue;
            }
            let file = entry.path();
            for definition in servers_for_file(file, config) {
                let Some(key) = server_key_for_definition(&definition, file, config) else {
                    continue;
                };
                if candidates.contains_key(&key) || producer_failures.contains_key(&key) {
                    continue;
                }
                if let Some(result) = self.failed_spawns.get(&key) {
                    producer_failures.insert(
                        key.clone(),
                        ApplicableServerFailure {
                            server_key: key,
                            result: result.clone(),
                        },
                    );
                    continue;
                }
                if self.resolve_binary(&definition, &key.root, config).is_err() {
                    producer_failures.insert(
                        key.clone(),
                        ApplicableServerFailure {
                            server_key: key,
                            result: ServerAttemptResult::BinaryNotInstalled {
                                binary: definition.binary.clone(),
                            },
                        },
                    );
                    continue;
                }
                candidates.insert(
                    key.clone(),
                    ApplicableServerCandidate {
                        key,
                        definition,
                        source_file: file.to_path_buf(),
                    },
                );
            }
        }

        if let Some(scope_roots) = scope_roots.filter(|roots| !roots.is_empty()) {
            let rust_roots = candidates
                .keys()
                .chain(producer_failures.keys())
                .filter(|key| key.kind == ServerKind::Rust)
                .map(|key| key.root.clone())
                .collect::<HashSet<_>>();
            let mut owners = HashSet::new();
            for scope_root in scope_roots {
                if let Some(owner) = rust_roots
                    .iter()
                    .filter(|root| scope_root.starts_with(root))
                    .max_by_key(|root| root.components().count())
                {
                    owners.insert(owner.clone());
                }
            }
            candidates.retain(|key, _| key.kind != ServerKind::Rust || owners.contains(&key.root));
            producer_failures
                .retain(|key, _| key.kind != ServerKind::Rust || owners.contains(&key.root));
        }

        let mut candidates = candidates.into_values().collect::<Vec<_>>();
        candidates.sort_by(|left, right| server_key_sort(&left.key, &right.key));
        let mut producer_failures = producer_failures.into_values().collect::<Vec<_>>();
        producer_failures
            .sort_by(|left, right| server_key_sort(&left.server_key, &right.server_key));
        let mut server_keys = candidates
            .iter()
            .map(|candidate| candidate.key.clone())
            .chain(
                producer_failures
                    .iter()
                    .map(|failure| failure.server_key.clone()),
            )
            .collect::<Vec<_>>();
        server_keys.sort_by(server_key_sort);
        Ok(ApplicableServerSnapshot {
            server_keys,
            candidates,
            producer_failures,
        })
    }

    /// Start exactly the servers selected by a prior applicability snapshot.
    ///
    /// No document is opened here. That boundary makes a late configuration or
    /// filesystem change apply to the next inspection rather than starting an
    /// unrecorded server in the current call.
    pub fn start_applicable_servers(
        &mut self,
        snapshot: &ApplicableServerSnapshot,
        config: &Config,
    ) -> ApplicableServerStartOutcomes {
        self.start_applicable_servers_inner(snapshot, config, None)
    }

    /// Start one inspect producer without extending the request's shared deadline.
    /// Calling this per key lets the inspect phase log begin a phase only while
    /// request budget remains to attempt that producer.
    pub fn start_applicable_server_until(
        &mut self,
        snapshot: &ApplicableServerSnapshot,
        server: &ServerKey,
        config: &Config,
        deadline: std::time::Instant,
    ) -> ApplicableServerStartOutcomes {
        let single = ApplicableServerSnapshot {
            server_keys: vec![server.clone()],
            candidates: snapshot
                .candidates
                .iter()
                .filter(|candidate| candidate.key == *server)
                .cloned()
                .collect(),
            producer_failures: snapshot
                .producer_failures
                .iter()
                .filter(|failure| failure.server_key == *server)
                .cloned()
                .collect(),
        };
        self.start_applicable_servers_inner(&single, config, Some(deadline))
    }

    fn start_applicable_servers_inner(
        &mut self,
        snapshot: &ApplicableServerSnapshot,
        config: &Config,
        deadline: Option<std::time::Instant>,
    ) -> ApplicableServerStartOutcomes {
        let mut outcomes = ApplicableServerStartOutcomes {
            failures: snapshot.producer_failures.clone(),
            ..ApplicableServerStartOutcomes::default()
        };
        for candidate in &snapshot.candidates {
            let initialize_timeout = deadline
                .map(|deadline| deadline.saturating_duration_since(std::time::Instant::now()));
            if initialize_timeout.is_some_and(|remaining| remaining.is_zero()) {
                outcomes.deadline_exceeded = Some(candidate.key.clone());
                break;
            }
            if self.clients.contains_key(&candidate.key) {
                outcomes.successful.push(candidate.key.clone());
                continue;
            }
            match self.spawn_server_with_timeout(
                &candidate.definition,
                &candidate.key.root,
                &candidate.source_file,
                config,
                initialize_timeout,
            ) {
                Ok(client) => {
                    self.clients.insert(candidate.key.clone(), client);
                    self.server_binaries
                        .insert(candidate.key.clone(), candidate.definition.binary.clone());
                    self.documents.entry(candidate.key.clone()).or_default();
                    outcomes.successful.push(candidate.key.clone());
                }
                Err(error) => {
                    if deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) {
                        // A request-owned timeout says nothing durable about the
                        // producer; a later call with a fresh budget may retry it.
                        outcomes.deadline_exceeded = Some(candidate.key.clone());
                        break;
                    }
                    let result = classify_spawn_error(&candidate.definition.binary, &error);
                    self.failed_spawns
                        .insert(candidate.key.clone(), result.clone());
                    outcomes.failures.push(ApplicableServerFailure {
                        server_key: candidate.key.clone(),
                        result,
                    });
                }
            }
        }
        outcomes
    }

    /// Ensure a server is running for the given file. Spawns if needed.
    /// Returns the active server keys for the file, or an empty vec if none match.
    ///
    /// This is the lightweight wrapper around [`ensure_server_for_file_detailed`]
    /// that drops failure context. Prefer the detailed variant in command
    /// handlers that need to surface honest error messages to the agent.
    pub fn ensure_server_for_file(&mut self, file_path: &Path, config: &Config) -> Vec<ServerKey> {
        self.ensure_server_for_file_detailed(file_path, config)
            .successful
    }

    fn running_server_keys_for_file(&self, file_path: &Path, config: &Config) -> Vec<ServerKey> {
        servers_for_file(file_path, config)
            .into_iter()
            .filter_map(|def| server_key_for_definition(&def, file_path, config))
            .filter(|key| self.clients.contains_key(key))
            .collect()
    }

    /// Return whether a navigation request would need to initialize an applicable
    /// server. Missing binaries, inapplicable root markers, and cached spawn
    /// failures stay on the synchronous path because they cannot incur a
    /// handshake wait.
    pub fn navigation_requires_deferred_execution(
        &self,
        file_path: &Path,
        config: &Config,
    ) -> bool {
        let Ok(canonical_path) = canonicalize_for_lsp(file_path) else {
            return false;
        };
        servers_for_file(&canonical_path, config)
            .into_iter()
            .filter_map(|definition| {
                let key = server_key_for_definition(&definition, &canonical_path, config)?;
                Some((definition, key))
            })
            .any(|(definition, key)| {
                if let Some(client) = self.clients.get(&key) {
                    return client.state() != ServerState::Ready;
                }
                !self.failed_spawns.contains_key(&key)
                    && self.resolve_binary(&definition, &key.root, config).is_ok()
            })
    }

    /// Detailed version of [`ensure_server_for_file`] that records every
    /// matching server's outcome (`Ok` / `NoRootMarker` / `BinaryNotInstalled`
    /// / `SpawnFailed`).
    ///
    /// Use this when the caller wants to honestly report _why_ a file has no
    /// active server (e.g., to surface "bash-language-server not on PATH" to
    /// the agent instead of silently returning `total: 0`).
    pub fn ensure_server_for_file_detailed(
        &mut self,
        file_path: &Path,
        config: &Config,
    ) -> EnsureServerOutcomes {
        let defs = servers_for_file(file_path, config);
        let mut outcomes = EnsureServerOutcomes::default();

        for def in defs {
            let server_id = def.kind.id_str().to_string();
            let server_name = def.name.to_string();

            let Some(key) = server_key_for_definition(&def, file_path, config) else {
                outcomes.attempts.push(ServerAttempt {
                    server_id,
                    server_name,
                    result: ServerAttemptResult::NoRootMarker {
                        looked_for: def.root_markers.iter().map(|s| s.to_string()).collect(),
                    },
                });
                continue;
            };

            if !self.clients.contains_key(&key) {
                // If we already tried and failed to spawn this server for this
                // root, return the cached classification without retrying or
                // re-logging. This prevents per-edit ERROR spam when the user's
                // environment is missing a dependency the LSP needs (the
                // typescript-language-server "Could not find a valid TypeScript
                // installation" case is the canonical example).
                if let Some(cached) = self.failed_spawns.get(&key) {
                    outcomes.attempts.push(ServerAttempt {
                        server_id,
                        server_name,
                        result: cached.clone(),
                    });
                    continue;
                }

                // A spurious ServerExited can drop the client without killing
                // the child. Reap any still-tracked processes for this pair
                // before spawning a replacement, otherwise leaked servers
                // accumulate on a live worktree.
                self.reap_unreferenced_children_for(&key);

                match self.spawn_server(&def, &key.root, file_path, config) {
                    Ok(client) => {
                        self.clients.insert(key.clone(), client);
                        self.server_binaries.insert(key.clone(), def.binary.clone());
                        self.documents.entry(key.clone()).or_default();
                    }
                    Err(err) => {
                        slog_error!("failed to spawn {}: {}", def.name, err);
                        let result = classify_spawn_error(&def.binary, &err);
                        // Remember the failure so subsequent file events skip
                        // this (kind, root) pair instead of producing a fresh
                        // spawn attempt + ERROR log per request.
                        self.failed_spawns.insert(key.clone(), result.clone());
                        outcomes.attempts.push(ServerAttempt {
                            server_id,
                            server_name,
                            result,
                        });
                        continue;
                    }
                }
            }

            outcomes.attempts.push(ServerAttempt {
                server_id,
                server_name,
                result: ServerAttemptResult::Ok {
                    server_key: key.clone(),
                },
            });
            outcomes.successful.push(key);
        }

        outcomes
    }

    /// Ensure a server is running using the default LSP registry.
    /// Kept for integration tests that exercise built-in server helpers directly.
    pub fn ensure_server_for_file_default(&mut self, file_path: &Path) -> Vec<ServerKey> {
        self.ensure_server_for_file(file_path, &Config::default())
    }
    /// Ensure that servers are running for the file and that the document is open
    /// in each server's DocumentStore. Reads file content from disk if not already open.
    /// The result identifies which servers were already tracking the document and which
    /// received `textDocument/didOpen` during this call.
    pub fn ensure_file_open(
        &mut self,
        file_path: &Path,
        config: &Config,
    ) -> Result<EnsureFileOpenResult, LspError> {
        let canonical_path = canonicalize_for_lsp(file_path)?;
        let server_keys = self.ensure_server_for_file(&canonical_path, config);
        if server_keys.is_empty() {
            return Ok(EnsureFileOpenResult::default());
        }

        let uri = uri_for_path(&canonical_path)?;
        let language_id = language_id_for_extension(
            canonical_path
                .extension()
                .and_then(|ext| ext.to_str())
                .unwrap_or_default(),
        )
        .to_string();
        let needs_content = server_keys.iter().any(|key| {
            !self
                .documents
                .get(key)
                .is_some_and(|store| store.is_open(&canonical_path))
        });
        let initial_content = needs_content
            .then(|| std::fs::read_to_string(&canonical_path).map_err(LspError::Io))
            .transpose()?;
        let mut newly_opened = Vec::new();

        for key in &server_keys {
            let already_open = self
                .documents
                .get(key)
                .is_some_and(|store| store.is_open(&canonical_path));

            if !already_open {
                let content = initial_content
                    .as_ref()
                    .expect("content is loaded when any server needs didOpen");
                let (send_result, sent) = if let Some(client) = self.clients.get_mut(key) {
                    (
                        client.send_notification::<DidOpenTextDocument>(
                            DidOpenTextDocumentParams {
                                text_document: TextDocumentItem::new(
                                    uri.clone(),
                                    language_id.clone(),
                                    0,
                                    content.clone(),
                                ),
                            },
                        ),
                        true,
                    )
                } else {
                    (Ok(()), false)
                };
                if let Err(err) = send_result {
                    let _ = self.close_file_for_servers(&canonical_path, &newly_opened);
                    return Err(err);
                }
                if sent {
                    log_did_open_sent(key, &canonical_path, &language_id);
                }
                self.documents
                    .entry(key.clone())
                    .or_default()
                    .open(canonical_path.clone());
                newly_opened.push(key.clone());
                continue;
            }

            // Document is already open. Check disk drift — if the file has
            // been modified outside the AFT pipeline (other tool, manual
            // edit, sibling session) we MUST send a didChange before any
            // pull-diagnostic / hover query, otherwise the LSP server
            // returns results computed from stale in-memory content.
            //
            // Without this, ensure_file_open would skip an already-open file
            // without checking whether its disk content changed, leaving the
            // server's in-memory copy stale.
            let drifted = self
                .documents
                .get(key)
                .is_some_and(|store| store.is_stale_on_disk(&canonical_path));
            if drifted {
                let content = match std::fs::read_to_string(&canonical_path) {
                    Ok(content) => content,
                    Err(err) => {
                        let _ = self.close_file_for_servers(&canonical_path, &newly_opened);
                        return Err(LspError::Io(err));
                    }
                };
                let next_version = self
                    .documents
                    .get(key)
                    .and_then(|store| store.version(&canonical_path))
                    .map(|v| v + 1)
                    .unwrap_or(1);
                let send_result = if let Some(client) = self.clients.get_mut(key) {
                    client.send_notification::<DidChangeTextDocument>(DidChangeTextDocumentParams {
                        text_document: VersionedTextDocumentIdentifier::new(
                            uri.clone(),
                            next_version,
                        ),
                        content_changes: vec![TextDocumentContentChangeEvent {
                            range: None,
                            range_length: None,
                            text: content,
                        }],
                    })
                } else {
                    Ok(())
                };
                if let Err(err) = send_result {
                    let _ = self.close_file_for_servers(&canonical_path, &newly_opened);
                    return Err(err);
                }
                if let Some(store) = self.documents.get_mut(key) {
                    store.bump_version(&canonical_path);
                }
            }
        }

        Ok(EnsureFileOpenResult {
            server_keys,
            newly_opened,
        })
    }

    pub fn ensure_file_open_default(
        &mut self,
        file_path: &Path,
    ) -> Result<EnsureFileOpenResult, LspError> {
        self.ensure_file_open(file_path, &Config::default())
    }

    /// Notify relevant LSP servers that a file has been written/changed.
    /// This is the main hook called after every file write in AFT.
    ///
    /// If the file's server isn't running yet, starts it (lazy spawn).
    /// If the file isn't open in LSP yet, sends didOpen. Otherwise sends didChange.
    pub fn notify_file_changed(
        &mut self,
        file_path: &Path,
        content: &str,
        config: &Config,
    ) -> Result<(), LspError> {
        self.notify_file_changed_versioned(file_path, content, config)
            .map(|_| ())
    }

    /// Like `notify_file_changed`, but returns the target document version
    /// per server so the post-edit waiter can match `publishDiagnostics`
    /// against the exact version that this notification carried.
    ///
    /// Returns: `Vec<(ServerKey, target_version)>`. `target_version` is the
    /// `version` field on the `VersionedTextDocumentIdentifier` we just sent
    /// (post-bump). For freshly-opened documents (`didOpen`) the version is
    /// `0`. Servers that do not echo a document version cannot prove a report
    /// describes this edit, so post-edit observation leaves them pending rather
    /// than accepting a report merely because it arrived after the edit.
    pub fn notify_file_changed_versioned(
        &mut self,
        file_path: &Path,
        content: &str,
        config: &Config,
    ) -> Result<Vec<(ServerKey, i32)>, LspError> {
        let canonical_path = canonicalize_for_lsp(file_path)?;
        let server_keys = self.ensure_server_for_file(&canonical_path, config);
        self.notify_file_changed_for_server_keys(canonical_path, content, server_keys)
    }

    /// Notify only LSP servers that are already running for this file.
    ///
    /// Post-write notifications are best-effort and must not make a mutation
    /// wait for cold server startup. Explicit LSP requests use
    /// [`Self::notify_file_changed_versioned`] and retain lazy startup.
    pub fn notify_file_changed_if_running(
        &mut self,
        file_path: &Path,
        content: &str,
        config: &Config,
    ) -> Result<(), LspError> {
        self.notify_file_changed_if_running_versioned(file_path, content, config)
            .map(|_| ())
    }

    /// Notify only already-live servers and retain the document versions that
    /// prove a subsequent diagnostics report belongs to this post-edit wait.
    /// Unlike `notify_file_changed_versioned`, this path never starts a server.
    pub fn notify_file_changed_if_running_versioned(
        &mut self,
        file_path: &Path,
        content: &str,
        config: &Config,
    ) -> Result<Vec<(ServerKey, i32)>, LspError> {
        let canonical_path = canonicalize_for_lsp(file_path)?;
        let server_keys = self.running_server_keys_for_file(&canonical_path, config);
        self.notify_file_changed_for_server_keys(canonical_path, content, server_keys)
    }

    fn notify_file_changed_for_server_keys(
        &mut self,
        canonical_path: PathBuf,
        content: &str,
        server_keys: Vec<ServerKey>,
    ) -> Result<Vec<(ServerKey, i32)>, LspError> {
        if server_keys.is_empty() {
            return Ok(Vec::new());
        }

        let uri = uri_for_path(&canonical_path)?;
        let language_id = language_id_for_extension(
            canonical_path
                .extension()
                .and_then(|ext| ext.to_str())
                .unwrap_or_default(),
        )
        .to_string();

        let mut versions: Vec<(ServerKey, i32)> = Vec::with_capacity(server_keys.len());

        for key in server_keys {
            let current_version = self
                .documents
                .get(&key)
                .and_then(|store| store.version(&canonical_path));

            if let Some(version) = current_version {
                let next_version = version + 1;
                if let Some(client) = self.clients.get_mut(&key) {
                    client.send_notification::<DidChangeTextDocument>(
                        DidChangeTextDocumentParams {
                            text_document: VersionedTextDocumentIdentifier::new(
                                uri.clone(),
                                next_version,
                            ),
                            content_changes: vec![TextDocumentContentChangeEvent {
                                range: None,
                                range_length: None,
                                text: content.to_string(),
                            }],
                        },
                    )?;
                }
                if let Some(store) = self.documents.get_mut(&key) {
                    store.bump_version(&canonical_path);
                }
                versions.push((key, next_version));
                continue;
            }

            if let Some(client) = self.clients.get_mut(&key) {
                client.send_notification::<DidOpenTextDocument>(DidOpenTextDocumentParams {
                    text_document: TextDocumentItem::new(
                        uri.clone(),
                        language_id.clone(),
                        0,
                        content.to_string(),
                    ),
                })?;
                log_did_open_sent(&key, &canonical_path, &language_id);
            }
            self.documents
                .entry(key.clone())
                .or_default()
                .open(canonical_path.clone());
            // didOpen carries version 0 — that's the version the server
            // will echo on its first publishDiagnostics for this document.
            versions.push((key, 0));
        }

        Ok(versions)
    }

    pub fn notify_file_changed_default(
        &mut self,
        file_path: &Path,
        content: &str,
    ) -> Result<(), LspError> {
        self.notify_file_changed(file_path, content, &Config::default())
    }

    /// Notify every active server whose workspace contains at least one changed
    /// path that watched files changed. This is intentionally workspace-scoped
    /// rather than extension-scoped: configuration edits such as `package.json`
    /// or `tsconfig.json` affect a server's project graph even though those
    /// files may not be documents handled by the server itself.
    pub fn notify_files_watched_changed(
        &mut self,
        paths: &[(PathBuf, FileChangeType)],
        _config: &Config,
    ) -> Result<(), LspError> {
        #[cfg(windows)]
        let mut trace = vec![format!(
            "input_paths={paths:?}; active_keys={:?}",
            self.clients.keys().collect::<Vec<_>>()
        )];

        if paths.is_empty() {
            #[cfg(windows)]
            {
                trace.push("outcome=no-input-paths".to_string());
                self.last_watched_file_notification_trace = trace.join("\n");
            }
            return Ok(());
        }

        let mut canonical_events = Vec::with_capacity(paths.len());
        for (path, typ) in paths {
            let canonical_path = resolve_for_lsp_uri(path);
            canonical_events.push((canonical_path, *typ));
        }
        #[cfg(windows)]
        trace.push(format!("resolved_events={canonical_events:?}"));

        let keys: Vec<ServerKey> = self.clients.keys().cloned().collect();
        #[cfg(windows)]
        if keys.is_empty() {
            trace.push("outcome=no-active-client".to_string());
        }
        for key in keys {
            let mut changes = Vec::new();
            for (path, typ) in &canonical_events {
                if !path.starts_with(&key.root) {
                    continue;
                }
                changes.push(FileEvent::new(uri_for_path(path)?, *typ));
            }

            if changes.is_empty() {
                #[cfg(windows)]
                trace.push(format!("key={key:?}; outcome=outside-root"));
                continue;
            }

            if let Some(client) = self.clients.get_mut(&key) {
                // Send when the server either advertised initialize-time
                // watched-file support or dynamically registered a watcher.
                // The dynamic client capability we send during initialize only
                // permits runtime registration; it is tracked separately via
                // `has_watched_file_registration()`.
                let supports_static_watched_files = client.supports_watched_files();
                let has_dynamic_registration = client.has_watched_file_registration();
                if !(supports_static_watched_files || has_dynamic_registration) {
                    #[cfg(windows)]
                    trace.push(format!(
                        "key={key:?}; changes={changes:?}; outcome=unsupported; static={supports_static_watched_files}; dynamic={has_dynamic_registration}"
                    ));
                    if self.watched_file_skip_logged.insert(key.clone()) {
                        log::debug!(
                            "skipping didChangeWatchedFiles for {:?} (not supported or registered)",
                            key
                        );
                    }
                    continue;
                }
                #[cfg(windows)]
                trace.push(format!("key={key:?}; changes={changes:?}; action=send"));
                let send_result = client.send_notification::<DidChangeWatchedFiles>(
                    DidChangeWatchedFilesParams { changes },
                );
                #[cfg(windows)]
                trace.push(format!(
                    "key={key:?}; outcome={}",
                    if send_result.is_ok() {
                        "sent"
                    } else {
                        "send-error"
                    }
                ));
                if let Err(error) = send_result {
                    #[cfg(windows)]
                    {
                        self.last_watched_file_notification_trace = trace.join("\n");
                    }
                    return Err(error);
                }
            }
        }

        #[cfg(windows)]
        {
            self.last_watched_file_notification_trace = trace.join("\n");
        }
        Ok(())
    }

    /// Close a document in all servers that have it open.
    pub fn notify_file_closed(&mut self, file_path: &Path) -> Result<(), LspError> {
        let canonical_path = canonicalize_for_lsp(file_path)?;
        let keys = self
            .documents
            .iter()
            .filter(|(_, store)| store.is_open(&canonical_path))
            .map(|(key, _)| key.clone())
            .collect::<Vec<_>>();
        self.close_file_for_servers(&canonical_path, &keys)
    }

    /// Close a document only in the specified servers.
    ///
    /// Scoped inspection uses this to release documents it opened without
    /// disturbing pre-existing editor documents in other server stores.
    pub(crate) fn close_file_for_servers(
        &mut self,
        file_path: &Path,
        server_keys: &[ServerKey],
    ) -> Result<(), LspError> {
        let canonical_path = canonicalize_for_lsp(file_path)?;
        let uri = uri_for_path(&canonical_path)?;
        let mut first_error = None;

        for key in server_keys {
            let was_open = self
                .documents
                .get(key)
                .is_some_and(|store| store.is_open(&canonical_path));
            if !was_open {
                continue;
            }

            if let Some(client) = self.clients.get_mut(key) {
                if let Err(err) =
                    client.send_notification::<DidCloseTextDocument>(DidCloseTextDocumentParams {
                        text_document: TextDocumentIdentifier::new(uri.clone()),
                    })
                {
                    if first_error.is_none() {
                        first_error = Some(err);
                    }
                }
            }

            if let Some(store) = self.documents.get_mut(key) {
                store.close(&canonical_path);
            }
            self.diagnostics.clear_for_server_file(key, &canonical_path);
        }

        match first_error {
            Some(err) => Err(err),
            None => Ok(()),
        }
    }

    /// Get an active client for a file path, if one exists.
    pub fn client_for_file(&self, file_path: &Path, config: &Config) -> Option<&LspClient> {
        let key = self.server_key_for_file(file_path, config)?;
        self.clients.get(&key)
    }

    pub fn client_for_file_default(&self, file_path: &Path) -> Option<&LspClient> {
        self.client_for_file(file_path, &Config::default())
    }

    /// Get a mutable active client for a file path, if one exists.
    pub fn client_for_file_mut(
        &mut self,
        file_path: &Path,
        config: &Config,
    ) -> Option<&mut LspClient> {
        let key = self.server_key_for_file(file_path, config)?;
        self.clients.get_mut(&key)
    }

    pub fn client_for_file_mut_default(&mut self, file_path: &Path) -> Option<&mut LspClient> {
        self.client_for_file_mut(file_path, &Config::default())
    }

    /// Number of tracked server clients.
    pub fn active_client_count(&self) -> usize {
        self.clients.len()
    }

    /// Drain all pending LSP events. Call from the main loop.
    pub fn drain_events(&mut self) -> DrainedLspEvents {
        self.drain_events_bounded(usize::MAX)
    }

    /// Whether LSP events are waiting to be drained. Cheap channel peek for
    /// the maintenance scheduler's skip probe.
    pub fn has_pending_events(&self) -> bool {
        !self.event_rx.is_empty()
    }

    pub fn drain_events_bounded(&mut self, max_events: usize) -> DrainedLspEvents {
        let mut events = Vec::new();
        let mut diagnostics_changed = false;
        let mut accepted_snapshots = Vec::new();
        while events.len() < max_events {
            let Ok(event) = self.event_rx.try_recv() else {
                break;
            };
            if self.handle_event(&event).is_some() {
                diagnostics_changed = true;
            }
            if let Some(snapshot) = self.accepted_live_publish_snapshot(&event) {
                accepted_snapshots.push(snapshot);
            }
            events.push(event);
        }
        let has_more = events.len() >= max_events && !self.event_rx.is_empty();
        DrainedLspEvents {
            events,
            diagnostics_changed,
            accepted_snapshots,
            has_more,
        }
    }

    /// Wait for diagnostics to arrive for a specific file until a timeout expires.
    pub fn wait_for_diagnostics(
        &mut self,
        file_path: &Path,
        config: &Config,
        timeout: std::time::Duration,
    ) -> Vec<StoredDiagnostic> {
        let deadline = std::time::Instant::now() + timeout;
        self.wait_for_file_diagnostics(file_path, config, deadline)
    }

    pub fn wait_for_diagnostics_default(
        &mut self,
        file_path: &Path,
        timeout: std::time::Duration,
    ) -> Vec<StoredDiagnostic> {
        self.wait_for_diagnostics(file_path, &Config::default(), timeout)
    }

    /// Test-only accessor for the diagnostics store. Used by integration
    /// tests that need to inspect per-server entries (e.g., to verify that
    /// `ServerKey::root` is populated correctly, not the empty path that
    /// the legacy `publish_with_kind` path produced).
    #[doc(hidden)]
    pub fn diagnostics_store_for_test(&self) -> &DiagnosticsStore {
        &self.diagnostics
    }

    #[doc(hidden)]
    pub fn diagnostics_store_mut_for_test(&mut self) -> &mut DiagnosticsStore {
        &mut self.diagnostics
    }

    #[doc(hidden)]
    pub fn post_edit_outcome_for_entry_for_test(
        key: ServerKey,
        entry: &DiagnosticEntry,
        target_version: i32,
        pre: PreEditSnapshot,
    ) -> PostEditWaitOutcome {
        Self::post_edit_outcome_for_entry(key, entry, target_version, pre)
    }

    fn post_edit_outcome_for_entry(
        key: ServerKey,
        entry: &DiagnosticEntry,
        target_version: i32,
        pre: PreEditSnapshot,
    ) -> PostEditWaitOutcome {
        let mut fresh = HashMap::new();
        if let Some(diagnostics) =
            Self::authoritative_post_edit_diagnostics(entry, target_version, pre)
        {
            fresh.insert(key.clone(), diagnostics);
        }
        Self::post_edit_outcome(vec![(key, target_version)], fresh, Vec::new())
    }

    fn authoritative_post_edit_diagnostics(
        entry: &DiagnosticEntry,
        target_version: i32,
        pre: PreEditSnapshot,
    ) -> Option<Vec<StoredDiagnostic>> {
        (!entry.provisional && post_edit_entry_is_fresh(entry, target_version, pre))
            .then(|| entry.diagnostics.clone())
    }

    #[doc(hidden)]
    pub fn enqueue_event_for_test(&self, event: LspEvent) {
        self.event_tx
            .send(event)
            .expect("LSP event receiver should remain connected");
    }

    // Used only by the Unix-gated child-spawning test modules.
    #[cfg(all(test, unix))]
    pub(crate) fn event_sender_for_test(&self) -> Sender<LspEvent> {
        self.event_tx.clone()
    }

    // Used only by the Unix-gated child-spawning test modules.
    #[cfg(all(test, unix))]
    pub(crate) fn insert_client_for_test(&mut self, client: LspClient) {
        let key = ServerKey {
            kind: client.kind(),
            root: client.root().to_path_buf(),
        };
        self.clients.insert(key, client);
    }

    #[doc(hidden)]
    pub fn pending_event_count_for_test(&self) -> usize {
        self.event_rx.len()
    }

    #[doc(hidden)]
    pub fn document_is_open_for_test(&self, file_path: &Path) -> bool {
        canonicalize_for_lsp(file_path).is_ok_and(|canonical_path| {
            self.documents
                .values()
                .any(|store| store.is_open(&canonical_path))
        })
    }

    /// Error/warning counts across the entire warm diagnostics set (all files
    /// any server has published for this session). Powers the agent status bar;
    /// reads the continuously-drained store with no extra LSP round-trip.
    pub fn warm_error_warning_counts(&self) -> (usize, usize) {
        self.diagnostics.error_warning_counts()
    }

    pub fn warm_error_warning_counts_with_provisional(&self) -> ((usize, usize), bool) {
        self.diagnostics.error_warning_counts_with_provisional()
    }

    pub fn diagnostics_generation(&self) -> u64 {
        self.diagnostics.generation()
    }

    /// Status-bar error/warning counts with a per-file `keep` predicate and
    /// cross-server dedup applied (see
    /// [`DiagnosticsStore::filtered_error_warning_counts`]). The caller supplies
    /// the project-root + tsconfig-membership policy via `keep`.
    pub fn filtered_error_warning_counts(
        &self,
        keep: impl FnMut(&std::path::Path) -> bool,
    ) -> (usize, usize) {
        self.diagnostics.filtered_error_warning_counts(keep)
    }

    /// Status-bar counts plus whether any kept diagnostics came from a server
    /// that is still warming. The readiness flag is needed to retain the last
    /// authoritative E/W values while provisional reports replace old entries.
    pub fn filtered_error_warning_counts_with_provisional(
        &self,
        keep: impl FnMut(&std::path::Path) -> bool,
    ) -> ((usize, usize), bool) {
        self.diagnostics
            .filtered_error_warning_counts_with_provisional(keep)
    }

    /// Active rust-analyzer instances that have not yet reported quiescence.
    /// Other server kinds are intentionally absent because they do not use the
    /// rust-analyzer readiness extension.
    pub fn provisional_server_keys(&self) -> Vec<ServerKey> {
        self.clients
            .iter()
            .filter(|(_, client)| client.diagnostics_are_provisional())
            .map(|(key, _)| key.clone())
            .collect()
    }

    /// Snapshot the current per-server epoch for every entry that exists
    /// for `file_path`. Servers without an entry yet (never published)
    /// are absent from the map; for those, `pre = 0` (any first publish
    /// will be considered fresh under the epoch-fallback rule).
    pub fn snapshot_diagnostic_epochs(&self, file_path: &Path) -> HashMap<ServerKey, u64> {
        let lookup_path = normalize_lookup_path(file_path);
        self.diagnostics
            .entries_for_file(&lookup_path)
            .into_iter()
            .map(|(key, entry)| (key.clone(), entry.epoch))
            .collect()
    }

    /// Snapshot the current diagnostic epoch and document version for every
    /// active server relevant to `file_path` before a post-edit notification.
    pub fn snapshot_pre_edit_state(&self, file_path: &Path) -> HashMap<ServerKey, PreEditSnapshot> {
        let lookup_path = normalize_lookup_path(file_path);
        let mut snapshots: HashMap<ServerKey, PreEditSnapshot> = self
            .diagnostics
            .entries_for_file(&lookup_path)
            .into_iter()
            .map(|(key, entry)| {
                (
                    key.clone(),
                    PreEditSnapshot {
                        epoch: entry.epoch,
                        document_version_at_capture: None,
                    },
                )
            })
            .collect();

        for (key, store) in &self.documents {
            if let Some(version) = store.version(&lookup_path) {
                snapshots
                    .entry(key.clone())
                    .or_default()
                    .document_version_at_capture = Some(version);
            }
        }

        snapshots
    }

    /// True when the current diagnostic entry for `server_key` can be tied to
    /// that server's current in-memory document version for `file_path`.
    ///
    /// File-mode `lsp_diagnostics` uses this for push-only fallback after it
    /// has synced/opened the document. Versioned publishes are accepted when
    /// they match the current document version; unversioned publishes are not
    /// accepted as fresh because epoch/wall-clock ordering alone is racy.
    pub fn diagnostic_entry_is_fresh_for_document(
        &self,
        file_path: &Path,
        server_key: &ServerKey,
        pre: PreEditSnapshot,
    ) -> bool {
        let lookup_path = normalize_lookup_path(file_path);
        let Some(entry) = self
            .diagnostics
            .entries_for_file(&lookup_path)
            .into_iter()
            .find_map(|(key, entry)| if key == server_key { Some(entry) } else { None })
        else {
            return false;
        };

        if entry.stale {
            return false;
        }

        let target_version = self
            .documents
            .get(server_key)
            .and_then(|store| store.version(&lookup_path))
            .or(pre.document_version_at_capture)
            .unwrap_or(0);

        matches!(entry.version, Some(version) if version >= target_version)
    }

    /// Prepare a post-edit wait and subscribe it before the manager mutex is
    /// released. The subscription closes the race between the state snapshot
    /// and a concurrent event drain.
    pub(crate) fn start_post_edit_diagnostics_wait(
        &mut self,
        file_path: &Path,
        expected_versions: &[(ServerKey, i32)],
        pre_snapshot: &HashMap<ServerKey, PreEditSnapshot>,
        timeout: std::time::Duration,
    ) -> PostEditDiagnosticsWait {
        let lookup_path = normalize_lookup_path(file_path);

        // Events sent after didChange may already be queued. Handle them before
        // parking, while freshness checks still reject pre-edit publications.
        let _ = self.drain_events_for_file(&lookup_path);

        let waiter_id = self.next_post_edit_waiter_id;
        self.next_post_edit_waiter_id = self.next_post_edit_waiter_id.wrapping_add(1);
        let (wake_tx, wake_rx) = bounded(1);
        self.post_edit_waiters.insert(waiter_id, wake_tx);

        PostEditDiagnosticsWait {
            lookup_path,
            expected_versions: expected_versions.to_vec(),
            pre_snapshot: pre_snapshot.clone(),
            event_rx: self.event_rx.clone(),
            wake_rx,
            waiter_id,
            deadline: std::time::Instant::now() + timeout,
            fresh: HashMap::new(),
            exited: Vec::new(),
        }
    }

    pub(crate) fn poll_post_edit_diagnostics_wait(
        &mut self,
        wait: &mut PostEditDiagnosticsWait,
        event: Option<LspEvent>,
    ) -> bool {
        if let Some(event) = event {
            self.handle_event(&event);
        }

        for (key, target_version) in &wait.expected_versions {
            if wait.fresh.contains_key(key) || wait.exited.contains(key) {
                continue;
            }
            if !self.clients.contains_key(key) {
                wait.exited.push(key.clone());
                continue;
            }
            if let Some(entry) = self
                .diagnostics
                .entries_for_file(&wait.lookup_path)
                .into_iter()
                .find_map(|(stored_key, entry)| (stored_key == key).then_some(entry))
            {
                let pre = wait.pre_snapshot.get(key).copied().unwrap_or_default();
                if let Some(diagnostics) =
                    Self::authoritative_post_edit_diagnostics(entry, *target_version, pre)
                {
                    wait.fresh.insert(key.clone(), diagnostics);
                }
            }
        }

        wait.fresh.len() + wait.exited.len() == wait.expected_versions.len()
    }

    pub(crate) fn finish_post_edit_diagnostics_wait(
        &mut self,
        wait: PostEditDiagnosticsWait,
    ) -> PostEditWaitOutcome {
        self.post_edit_waiters.remove(&wait.waiter_id);
        Self::post_edit_outcome(wait.expected_versions, wait.fresh, wait.exited)
    }

    /// Wait for fresh per-server diagnostics matching the just-sent document
    /// version. Cached pre-edit entries and provisional warming reports remain
    /// pending; a server exit is reported separately.
    ///
    /// `AppContext` uses the prepare/poll/finish methods directly so channel
    /// waiting happens without its manager mutex. This convenience method keeps
    /// the same behavior for standalone manager callers.
    pub fn wait_for_post_edit_diagnostics(
        &mut self,
        file_path: &Path,
        // `config` is intentionally accepted (matches sibling wait APIs and
        // future-proofs us if freshness rules need it). Currently unused
        // because expected_versions/pre_snapshot fully determine behavior.
        _config: &Config,
        expected_versions: &[(ServerKey, i32)],
        pre_snapshot: &HashMap<ServerKey, PreEditSnapshot>,
        timeout: std::time::Duration,
    ) -> PostEditWaitOutcome {
        let mut wait = self.start_post_edit_diagnostics_wait(
            file_path,
            expected_versions,
            pre_snapshot,
            timeout,
        );
        let mut complete = self.poll_post_edit_diagnostics_wait(&mut wait, None);

        while !complete && !wait.deadline_reached() {
            let event = wait.next_event();
            complete = self.poll_post_edit_diagnostics_wait(&mut wait, event);
        }

        self.finish_post_edit_diagnostics_wait(wait)
    }

    fn post_edit_outcome(
        mut expected: Vec<(ServerKey, i32)>,
        mut fresh: HashMap<ServerKey, Vec<StoredDiagnostic>>,
        exited: Vec<ServerKey>,
    ) -> PostEditWaitOutcome {
        expected.sort_by(|(left, _), (right, _)| server_key_sort(left, right));

        let mut accepted_snapshots = Vec::new();
        let mut pending_servers = Vec::new();
        for (server_key, document_version) in expected {
            if let Some(diagnostics) = fresh.remove(&server_key) {
                accepted_snapshots.push(AcceptedDiagnosticSnapshot::new(
                    server_key,
                    document_version,
                    diagnostics,
                ));
            } else if !exited.contains(&server_key) {
                pending_servers.push(server_key);
            }
        }

        let mut diagnostics = accepted_snapshots
            .iter()
            .flat_map(|snapshot| snapshot.diagnostics.iter().cloned())
            .collect::<Vec<_>>();
        diagnostics.sort_by(|left, right| {
            left.file
                .cmp(&right.file)
                .then(left.line.cmp(&right.line))
                .then(left.column.cmp(&right.column))
                .then(left.message.cmp(&right.message))
        });

        PostEditWaitOutcome {
            accepted_snapshots,
            diagnostics,
            pending_servers,
            exited_servers: exited,
        }
    }

    /// Wait for diagnostics to arrive for a specific file until a deadline.
    ///
    /// Drains already-queued events first, then blocks on the shared event
    /// channel only until either `publishDiagnostics` arrives for this file or
    /// the deadline is reached.
    pub fn wait_for_file_diagnostics(
        &mut self,
        file_path: &Path,
        config: &Config,
        deadline: std::time::Instant,
    ) -> Vec<StoredDiagnostic> {
        let lookup_path = normalize_lookup_path(file_path);

        if self.server_key_for_file(&lookup_path, config).is_none() {
            return Vec::new();
        }

        loop {
            if self.drain_events_for_file(&lookup_path) {
                break;
            }

            let now = std::time::Instant::now();
            if now >= deadline {
                break;
            }

            let timeout = deadline.saturating_duration_since(now);
            match self.event_rx.recv_timeout(timeout) {
                Ok(event) => {
                    if matches!(
                        self.handle_event(&event),
                        Some(ref published_file) if published_file.as_path() == lookup_path.as_path()
                    ) {
                        break;
                    }
                }
                Err(RecvTimeoutError::Timeout) | Err(RecvTimeoutError::Disconnected) => break,
            }
        }

        self.get_diagnostics_for_file(&lookup_path)
            .into_iter()
            .cloned()
            .collect()
    }

    /// Default timeout for `textDocument/diagnostic` (per-file pull). Servers
    /// usually respond in under 1s for files they've already analyzed; we
    /// allow up to 10s before falling back to push semantics. Currently
    /// surfaced via [`Self::pull_file_timeout`] for callers that want to
    /// override the wait via the `wait_ms` knob.
    pub const PULL_FILE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);

    /// Public accessor so command handlers can reuse the documented default.
    pub fn pull_file_timeout() -> std::time::Duration {
        Self::PULL_FILE_TIMEOUT
    }

    /// Default timeout for `workspace/diagnostic`. The LSP spec allows the
    /// server to hold this open indefinitely; we cap at 10s and report
    /// `complete: false` to the agent rather than hanging the bridge.
    const PULL_WORKSPACE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);

    /// Issue a `textDocument/diagnostic` (LSP 3.17 per-file pull) request to
    /// every server that supports pull diagnostics for the given file.
    ///
    /// Returns the per-server outcome. If a server reports `kind: "unchanged"`,
    /// the cached entry's diagnostics are surfaced (deterministic re-use of
    /// the previous response). If a server doesn't advertise pull capability,
    /// it's skipped here — the caller should fall back to push for those.
    ///
    /// Side effects: results are stored in `DiagnosticsStore` so directory-mode
    /// queries can aggregate them later.
    pub fn pull_file_diagnostics(
        &mut self,
        file_path: &Path,
        config: &Config,
    ) -> Result<Vec<PullFileResult>, LspError> {
        self.pull_file_diagnostics_inner(file_path, config, None)
    }

    /// Pull diagnostics within a caller-owned wait budget. This is used after
    /// edits because a pull-capable server may disable push diagnostics after
    /// seeing the client's LSP 3.17 diagnostic capability.
    pub fn pull_file_diagnostics_with_timeout(
        &mut self,
        file_path: &Path,
        config: &Config,
        timeout: Duration,
    ) -> Result<Vec<PullFileResult>, LspError> {
        self.pull_file_diagnostics_inner(file_path, config, Some(Instant::now() + timeout))
    }

    fn pull_file_diagnostics_inner(
        &mut self,
        file_path: &Path,
        config: &Config,
        deadline: Option<Instant>,
    ) -> Result<Vec<PullFileResult>, LspError> {
        let canonical_path = canonicalize_for_lsp(file_path)?;
        // Make sure servers are running and the document is open with fresh
        // content (handles disk-drift via DocumentStore::is_stale_on_disk).
        let opened = self.ensure_file_open(&canonical_path, config)?;
        if opened.server_keys.is_empty() {
            return Ok(Vec::new());
        }

        let uri = uri_for_path(&canonical_path)?;
        let mut results = Vec::with_capacity(opened.server_keys.len());

        for key in opened.server_keys {
            let supports_pull = self
                .clients
                .get(&key)
                .and_then(|c| c.diagnostic_capabilities())
                .is_some_and(|caps| caps.pull_diagnostics);

            if !supports_pull {
                results.push(PullFileResult {
                    server_key: key.clone(),
                    outcome: PullFileOutcome::PullNotSupported,
                });
                continue;
            }

            // Look up previous resultId for incremental requests.
            let previous_result_id = self
                .diagnostics
                .entries_for_file(&canonical_path)
                .into_iter()
                .find(|(k, _)| **k == key)
                .and_then(|(_, entry)| entry.result_id.clone());

            let identifier = self
                .clients
                .get(&key)
                .and_then(|c| c.diagnostic_capabilities())
                .and_then(|caps| caps.identifier.clone());

            let params = AftDocumentDiagnosticParams {
                text_document: lsp_types::TextDocumentIdentifier { uri: uri.clone() },
                identifier,
                previous_result_id,
                work_done_progress_params: Default::default(),
                partial_result_params: Default::default(),
            };

            let request_timeout = deadline
                .map(|deadline| {
                    deadline
                        .saturating_duration_since(Instant::now())
                        .min(Self::PULL_FILE_TIMEOUT)
                })
                .unwrap_or(Self::PULL_FILE_TIMEOUT);
            let document_version = self
                .documents
                .get(&key)
                .and_then(|store| store.version(&canonical_path));
            slog_info!(
                "lsp_protocol server={} root={} method=textDocument/diagnostic event=sent file={} document_version={}",
                key.kind.id_str(),
                key.root.display(),
                canonical_path.display(),
                document_version
                    .map(|version| version.to_string())
                    .unwrap_or_else(|| "none".to_string())
            );
            let outcome = match self.send_pull_request(&key, params, request_timeout) {
                Ok(report) => {
                    let report_kind = match &report {
                        lsp_types::DocumentDiagnosticReportResult::Report(
                            lsp_types::DocumentDiagnosticReport::Full(_),
                        ) => "full",
                        lsp_types::DocumentDiagnosticReportResult::Report(
                            lsp_types::DocumentDiagnosticReport::Unchanged(_),
                        ) => "unchanged",
                        lsp_types::DocumentDiagnosticReportResult::Partial(_) => "partial",
                    };
                    slog_info!(
                        "lsp_protocol server={} root={} method=textDocument/diagnostic event=received file={} report_kind={}",
                        key.kind.id_str(),
                        key.root.display(),
                        canonical_path.display(),
                        report_kind
                    );
                    if matches!(
                        &report,
                        lsp_types::DocumentDiagnosticReportResult::Report(
                            lsp_types::DocumentDiagnosticReport::Full(_)
                        )
                    ) {
                        // The server may publish diagnostics for didOpen before
                        // returning a full pull response. Apply those older events
                        // first so the full report remains authoritative. An
                        // unchanged response must inspect only a previous pull cache.
                        self.drain_events();
                    }
                    self.ingest_document_report(&key, &canonical_path, document_version, report)
                }
                Err(err) => {
                    slog_info!(
                        "lsp_protocol server={} root={} method=textDocument/diagnostic event=failed file={} error={}",
                        key.kind.id_str(),
                        key.root.display(),
                        canonical_path.display(),
                        err
                    );
                    if let Some(result) = self.cache_post_initialize_exit(&key, &err) {
                        PullFileOutcome::RequestFailed {
                            reason: server_attempt_result_reason(&result),
                        }
                    } else if recoverable_pull_rejection(&err)
                        && self.clients.get(&key).is_some_and(|client| {
                            matches!(
                                client.state(),
                                ServerState::Ready | ServerState::Initializing
                            )
                        })
                    {
                        PullFileOutcome::RequestFailed {
                            reason: format!("pull_rejected_push_fallback: {err}"),
                        }
                    } else {
                        PullFileOutcome::RequestFailed {
                            reason: err.to_string(),
                        }
                    }
                }
            };

            results.push(PullFileResult {
                server_key: key,
                outcome,
            });
        }

        Ok(results)
    }

    /// Issue a `workspace/diagnostic` request to a specific server. Cancels
    /// internally if `timeout` elapses before the server responds. Cached
    /// entries from the response are stored so directory-mode queries pick
    /// them up.
    pub fn pull_workspace_diagnostics(
        &mut self,
        server_key: &ServerKey,
        timeout: Option<std::time::Duration>,
    ) -> Result<PullWorkspaceResult, LspError> {
        let timeout = timeout.unwrap_or(Self::PULL_WORKSPACE_TIMEOUT);

        let supports_workspace = self
            .clients
            .get(server_key)
            .and_then(|c| c.diagnostic_capabilities())
            .is_some_and(|caps| caps.workspace_diagnostics);

        if !supports_workspace {
            return Ok(PullWorkspaceResult {
                server_key: server_key.clone(),
                files_reported: Vec::new(),
                complete: false,
                cancelled: false,
                supports_workspace: false,
            });
        }

        let identifier = self
            .clients
            .get(server_key)
            .and_then(|c| c.diagnostic_capabilities())
            .and_then(|caps| caps.identifier.clone());

        let params = AftWorkspaceDiagnosticParams {
            identifier,
            previous_result_ids: Vec::new(),
            work_done_progress_params: Default::default(),
            partial_result_params: Default::default(),
        };

        let result = match self
            .clients
            .get_mut(server_key)
            .ok_or_else(|| LspError::ServerNotReady("server not found".into()))?
            .send_request_with_timeout::<AftWorkspaceDiagnosticRequest>(params, timeout)
        {
            Ok(result) => result,
            Err(LspError::Timeout(_)) => {
                return Ok(PullWorkspaceResult {
                    server_key: server_key.clone(),
                    files_reported: Vec::new(),
                    complete: false,
                    cancelled: true,
                    supports_workspace: true,
                });
            }
            Err(err) => {
                if let Some(result) = self.cache_post_initialize_exit(server_key, &err) {
                    return Err(LspError::ServerNotReady(server_attempt_result_reason(
                        &result,
                    )));
                }
                return Err(err);
            }
        };

        // Extract the items list. Partial responses are not a complete
        // workspace view, but the partial payload can still contain useful
        // document reports; ingest those while surfacing complete=false.
        let (items, complete) = match result {
            lsp_types::WorkspaceDiagnosticReportResult::Report(report) => (report.items, true),
            lsp_types::WorkspaceDiagnosticReportResult::Partial(partial) => (partial.items, false),
        };

        // Ingest each file report into the diagnostics store.
        let mut files_reported = Vec::with_capacity(items.len());
        for item in items {
            match item {
                lsp_types::WorkspaceDocumentDiagnosticReport::Full(full) => {
                    if let Some(file) = uri_to_path(&full.uri) {
                        let stored = from_lsp_diagnostics(
                            file.clone(),
                            full.full_document_diagnostic_report.items.clone(),
                        );
                        self.diagnostics.publish_with_result_id(
                            server_key.clone(),
                            file.clone(),
                            stored,
                            full.full_document_diagnostic_report.result_id.clone(),
                        );
                        files_reported.push(file);
                    }
                }
                lsp_types::WorkspaceDocumentDiagnosticReport::Unchanged(_unchanged) => {
                    // "Unchanged" means the previously cached report is still
                    // valid. We left it in place; nothing to do.
                }
            }
        }

        Ok(PullWorkspaceResult {
            server_key: server_key.clone(),
            files_reported,
            complete,
            cancelled: false,
            supports_workspace: true,
        })
    }

    fn cache_post_initialize_exit(
        &mut self,
        key: &ServerKey,
        err: &LspError,
    ) -> Option<ServerAttemptResult> {
        let binary = self
            .server_binaries
            .get(key)
            .cloned()
            .unwrap_or_else(|| key.kind.id_str().to_string());
        let (status, stderr_tail) = {
            let client = self.clients.get_mut(key)?;
            let mut status = client.child_exit_status();
            for _ in 0..10 {
                if status.is_some() {
                    break;
                }
                std::thread::sleep(std::time::Duration::from_millis(10));
                status = client.child_exit_status();
            }
            let status = status?;
            wait_for_stderr_tail(client);
            (status, client.stderr_tail())
        };
        let reason = format_post_initialize_exit_reason(&binary, status, &stderr_tail, err);
        let result = ServerAttemptResult::SpawnFailed { binary, reason };
        self.clients.remove(key);
        self.server_binaries.remove(key);
        self.documents.remove(key);
        self.diagnostics.clear_for_server(key);
        self.failed_spawns.insert(key.clone(), result.clone());
        Some(result)
    }

    /// Issue the per-file diagnostic request and return the report.
    fn send_pull_request(
        &mut self,
        key: &ServerKey,
        params: AftDocumentDiagnosticParams,
        timeout: Duration,
    ) -> Result<lsp_types::DocumentDiagnosticReportResult, LspError> {
        let client = self
            .clients
            .get_mut(key)
            .ok_or_else(|| LspError::ServerNotReady("server not found".into()))?;
        // The caller caps this below the global 30s request timeout so a stalled
        // pull server cannot consume the entire inspect or post-edit budget.
        client.send_request_with_timeout::<AftDocumentDiagnosticRequest>(params, timeout)
    }

    /// Store the result of a per-file pull request and return a structured
    /// outcome the caller can inspect.
    fn ingest_document_report(
        &mut self,
        key: &ServerKey,
        canonical_path: &Path,
        document_version: Option<i32>,
        result: lsp_types::DocumentDiagnosticReportResult,
    ) -> PullFileOutcome {
        let report = match result {
            lsp_types::DocumentDiagnosticReportResult::Report(report) => report,
            lsp_types::DocumentDiagnosticReportResult::Partial(_) => {
                // Partial results stream in via $/progress notifications which
                // we don't currently subscribe to. Treat as a soft-empty
                // success — the next pull will get the full version.
                return PullFileOutcome::PartialNotSupported;
            }
        };

        match report {
            lsp_types::DocumentDiagnosticReport::Full(full) => {
                let result_id = full.full_document_diagnostic_report.result_id.clone();
                let stored = from_lsp_diagnostics(
                    canonical_path.to_path_buf(),
                    full.full_document_diagnostic_report.items.clone(),
                );
                let count = stored.len();
                let provisional = self
                    .clients
                    .get(key)
                    .is_some_and(|client| client.diagnostics_are_provisional());
                self.diagnostics.publish_full_with_provisional(
                    key.clone(),
                    canonical_path.to_path_buf(),
                    stored,
                    result_id,
                    document_version,
                    provisional,
                );
                PullFileOutcome::Full {
                    diagnostic_count: count,
                }
            }
            lsp_types::DocumentDiagnosticReport::Unchanged(_unchanged) => {
                // The server says the previous resultId is still valid for the
                // current document. That is only usable if we already have a
                // report for this exact server/file; an initial `unchanged`
                // response cannot prove freshness. A stale watcher entry is
                // acceptable here because the pull response itself proves the
                // cached diagnostics still describe the now-synced file.
                if self
                    .diagnostics
                    .has_report_for_server_file(key, canonical_path)
                {
                    if let Some(version) = document_version {
                        self.diagnostics.confirm_for_server_file_version(
                            key,
                            canonical_path,
                            version,
                        );
                    } else {
                        self.diagnostics
                            .mark_fresh_for_server_file(key, canonical_path);
                    }
                    let authoritative = self
                        .clients
                        .get(key)
                        .map_or(true, |client| !client.diagnostics_are_provisional());
                    if authoritative {
                        self.diagnostics
                            .clear_provisional_for_server_file(key, canonical_path);
                    }
                    PullFileOutcome::Unchanged
                } else {
                    PullFileOutcome::RequestFailed {
                        reason: "no_cache_for_unchanged".to_string(),
                    }
                }
            }
        }
    }

    /// Drain every client and clear spawn/document/diagnostic state without
    /// waiting on child processes. The caller owns graceful shutdown so the
    /// manager lock is not held across a Shutdown handshake.
    pub fn take_all_clients(&mut self) -> Vec<(ServerKey, LspClient)> {
        let clients: Vec<_> = self.clients.drain().collect();
        self.server_binaries.clear();
        self.documents.clear();
        self.diagnostics = DiagnosticsStore::new();
        clients
    }

    /// Shut down every server concurrently within one shared grace period.
    pub fn shutdown_all(&mut self) -> LspShutdownAllOutcome {
        let clients = self.take_all_clients();
        Self::shutdown_all_clients(
            clients,
            self.child_registry.clone(),
            LSP_SHUTDOWN_ALL_BUDGET,
        )
    }

    fn shutdown_all_clients(
        clients: Vec<(ServerKey, LspClient)>,
        child_registry: LspChildRegistry,
        budget: Duration,
    ) -> LspShutdownAllOutcome {
        let started = Instant::now();
        let mut pending_pids = clients
            .iter()
            .map(|(_, client)| client.child_pid())
            .collect::<HashSet<_>>();
        let (result_tx, result_rx) = unbounded();

        for (key, mut client) in clients {
            let pid = client.child_pid();
            let result_tx = result_tx.clone();
            std::thread::spawn(move || {
                let result = client.shutdown();
                // Do the drop before reporting completion so the registry cannot
                // briefly report a reaped client as still live after this method
                // has returned its shutdown summary.
                drop(client);
                let _ = result_tx.send((key, pid, result));
            });
        }
        drop(result_tx);

        let deadline = started + budget;
        let mut outcome = LspShutdownAllOutcome::default();
        while !pending_pids.is_empty() {
            let remaining = deadline.saturating_duration_since(Instant::now());
            match result_rx.recv_timeout(remaining) {
                Ok((key, pid, result)) => {
                    if !pending_pids.remove(&pid) {
                        continue;
                    }
                    match result {
                        Ok(()) => outcome.graceful += 1,
                        Err(err) => {
                            outcome.forced += 1;
                            slog_error!("error shutting down {:?}: {}", key, err);
                        }
                    }
                }
                Err(RecvTimeoutError::Timeout | RecvTimeoutError::Disconnected) => break,
            }
        }

        if !pending_pids.is_empty() {
            let pids = pending_pids.into_iter().collect::<Vec<_>>();
            outcome.forced += pids.len();
            // The shared registry owns process-group termination, so timed-out
            // clients are reaped even though their graceful worker is still
            // blocked in its per-client Shutdown request.
            child_registry.reap_pids(&pids);
        }

        outcome.elapsed = started.elapsed();
        slog_info!(
            "lsp shutdown_all: graceful={} forced={} elapsed_ms={}",
            outcome.graceful,
            outcome.forced,
            outcome.elapsed.as_millis()
        );
        outcome
    }

    /// Shut down taken clients on a detached thread named `aft-lsp-idle-reap`.
    /// Idle reapers use this so a hung Shutdown handshake cannot stall the
    /// daemon loop or hold the manager mutex.
    pub(crate) fn spawn_idle_lsp_reap(clients: Vec<(ServerKey, LspClient)>) {
        if clients.is_empty() {
            return;
        }
        let spawn_result = std::thread::Builder::new()
            .name("aft-lsp-idle-reap".into())
            .spawn(move || {
                for (key, mut client) in clients {
                    if let Err(err) = client.shutdown_for_idle_reap() {
                        slog_error!("error shutting down {:?}: {}", key, err);
                    }
                }
            });
        if let Err(err) = spawn_result {
            slog_error!("failed to spawn idle LSP reap thread: {err}");
        }
    }

    /// Check if any server is active.
    pub fn has_active_servers(&self) -> bool {
        self.clients
            .values()
            .any(|client| client.state() == ServerState::Ready)
    }

    /// Active server keys (running clients). Used by `lsp_diagnostics`
    /// directory mode to know which servers to ask for workspace pull.
    pub fn active_server_keys(&self) -> Vec<ServerKey> {
        self.clients.keys().cloned().collect()
    }

    /// Return the last watched-file routing decision for Windows CI timeout
    /// diagnostics. The trace records whether a matching client was absent,
    /// outside the changed path's root, unsupported, or successfully written.
    #[cfg(windows)]
    #[doc(hidden)]
    pub fn watched_file_notification_trace_for_test(&self) -> &str {
        &self.last_watched_file_notification_trace
    }

    pub fn get_diagnostics_for_file(&self, file: &Path) -> Vec<&StoredDiagnostic> {
        let normalized = normalize_lookup_path(file);
        self.diagnostics.for_file(&normalized)
    }

    pub fn get_diagnostics_for_file_with_provisional(
        &self,
        file: &Path,
    ) -> Vec<(&StoredDiagnostic, bool)> {
        let normalized = normalize_lookup_path(file);
        self.diagnostics.for_file_with_provisional(&normalized)
    }

    /// Drop all cached diagnostics for a file across every server. Called when a
    /// file is deleted/renamed away so its diagnostics don't linger in the warm
    /// set (no server republishes for a vanished path), inflating the
    /// error/warning counts in the status bar and `aft_inspect`.
    ///
    /// The store key is the canonical path from publish time, but a deleted file
    /// can no longer be canonicalized directly (`canonicalize` needs the file to
    /// exist). We therefore try several equivalent forms: the raw path, the
    /// canonicalize-or-fallback form, and — crucially — a reconstruction that
    /// canonicalizes the still-present parent directory and rejoins the file
    /// name, which reproduces the publish-time key even across `/var`↔
    /// `/private/var`-style symlink aliasing. Returns true if anything was
    /// removed.
    /// Forget all cached spawn FAILURES so the next file event retries them.
    /// Called on `configure`: a configure means something changed (the user may
    /// have just installed the missing language server, or fixed PATH / a
    /// version pin), so a previously-failed (kind, root) pair deserves a fresh
    /// attempt instead of being skipped until a full restart. Bounded: configure
    /// is not a per-request hot path, so this cannot cause a spawn storm.
    /// Returns the number of cleared entries.
    pub fn clear_failed_spawns(&mut self) -> usize {
        let n = self.failed_spawns.len();
        self.failed_spawns.clear();
        n
    }

    #[cfg(test)]
    pub(crate) fn insert_failed_spawn_for_test(&mut self) {
        let key = ServerKey {
            kind: crate::lsp::registry::ServerKind::Rust,
            root: std::path::PathBuf::from("/tmp/test-root"),
        };
        self.failed_spawns.insert(
            key,
            ServerAttemptResult::SpawnFailed {
                binary: "rust-analyzer".to_string(),
                reason: "test".to_string(),
            },
        );
    }

    pub fn clear_diagnostics_for_file(&mut self, file: &Path) -> bool {
        diagnostic_path_candidates(file)
            .into_iter()
            .fold(false, |removed, candidate| {
                removed | self.diagnostics.clear_for_file(&candidate)
            })
    }

    /// Mark cached diagnostics for this file stale after a watcher-observed
    /// external edit. The same path aliases as deletion are checked so canonical
    /// publish keys are found even when the watcher reports a symlinked path.
    pub fn mark_diagnostics_stale_for_file(&mut self, file: &Path) -> StaleDiagnosticsMark {
        let mut result = StaleDiagnosticsMark::default();
        for candidate in diagnostic_path_candidates(file) {
            let (had_entries, changed) = self.diagnostics.mark_stale_for_file(&candidate);
            result.had_entries |= had_entries;
            result.changed |= changed;
        }
        result
    }

    pub fn get_diagnostics_for_directory(&self, dir: &Path) -> Vec<&StoredDiagnostic> {
        let normalized = normalize_lookup_path(dir);
        self.diagnostics.for_directory(&normalized)
    }

    pub fn get_diagnostics_for_directory_with_provisional(
        &self,
        dir: &Path,
    ) -> Vec<(&StoredDiagnostic, bool)> {
        let normalized = normalize_lookup_path(dir);
        self.diagnostics.for_directory_with_provisional(&normalized)
    }

    pub fn get_all_diagnostics(&self) -> Vec<&StoredDiagnostic> {
        self.diagnostics.all()
    }

    pub fn get_all_diagnostics_with_provisional(&self) -> Vec<(&StoredDiagnostic, bool)> {
        self.diagnostics.all_with_provisional()
    }

    /// True if any LSP server has a current diagnostic report, including an
    /// empty report that proves a checked-clean file. This lets callers avoid
    /// treating an empty flattened diagnostic list as trustworthy when no server
    /// has actually run or every report was marked stale after an external edit.
    pub fn has_any_diagnostic_reports(&self) -> bool {
        self.diagnostics.has_any_fresh_report()
    }

    /// True if any server has a current report for this file, including an
    /// empty checked-clean report. Watcher-stale reports are excluded because
    /// they predate an external edit.
    pub fn has_diagnostic_report_for_file(&self, file: &Path) -> bool {
        let normalized = normalize_lookup_path(file);
        self.diagnostics.has_any_fresh_report_for_file(&normalized)
    }

    /// True if this exact server/file pair has a current diagnostic report,
    /// including an empty checked-clean report. Watcher-stale reports are
    /// excluded because they predate an external edit.
    pub fn has_diagnostic_report_for_server_file(&self, server: &ServerKey, file: &Path) -> bool {
        let normalized = normalize_lookup_path(file);
        self.diagnostics
            .has_fresh_report_for_server_file(server, &normalized)
    }

    /// True if any server has an authoritative report for this file: neither
    /// watcher-stale nor warming-provisional, including an empty checked-clean
    /// report. Needed for per-file diagnostic authority: a global "some server
    /// reported" bit cannot prove that a specific file was analyzed.
    pub fn has_authoritative_report_for_file(&self, file: &Path) -> bool {
        let normalized = normalize_lookup_path(file);
        self.diagnostics
            .has_authoritative_report_for_file(&normalized)
    }

    /// True if this server instance holds any authoritative report: an entry
    /// that is neither watcher-stale nor warming-provisional, including an
    /// empty checked-clean report. Needed by the blocking quiescence wait, which
    /// settles per producer rather than per file.
    pub fn has_authoritative_report_for_server(&self, server: &ServerKey) -> bool {
        self.diagnostics.has_authoritative_report_for_server(server)
    }

    /// True if this server instance is still warming: its reports stay
    /// provisional until it declares quiescence. A server the manager does not
    /// know (never started, or exited since) is not warming and cannot block a
    /// quiescence wait.
    pub fn server_is_warming(&self, server: &ServerKey) -> bool {
        self.clients
            .get(server)
            .is_some_and(|client| client.diagnostics_are_provisional())
    }

    /// A producer has settled when it holds a current authoritative report or
    /// has stopped warming. The blocking inspect wait and the unscoped
    /// diagnostics freshness gate both use this predicate so a complete
    /// producer set cannot be judged incomplete by a second, stricter check.
    pub fn producer_has_settled(&self, server: &ServerKey) -> bool {
        self.has_authoritative_report_for_server(server) || !self.server_is_warming(server)
    }

    /// True when every expected producer has settled. Empty input is vacuously
    /// true; callers that mean "no producer was started" must not treat that
    /// as a fresh diagnostics collection.
    pub fn producers_settled(&self, expected: &[ServerKey]) -> bool {
        expected
            .iter()
            .all(|server| self.producer_has_settled(server))
    }

    fn drain_events_for_file(&mut self, file_path: &Path) -> bool {
        let mut saw_file_diagnostics = false;
        while let Ok(event) = self.event_rx.try_recv() {
            if matches!(
                self.handle_event(&event),
                Some(ref published_file) if published_file.as_path() == file_path
            ) {
                saw_file_diagnostics = true;
            }
        }
        saw_file_diagnostics
    }

    fn accepted_live_publish_snapshot(
        &self,
        event: &LspEvent,
    ) -> Option<AcceptedDiagnosticSnapshot> {
        let LspEvent::Notification {
            server_kind,
            root,
            method,
            params: Some(params),
        } = event
        else {
            return None;
        };
        if method != "textDocument/publishDiagnostics" {
            return None;
        }

        let publish_params =
            serde_json::from_value::<lsp_types::PublishDiagnosticsParams>(params.clone()).ok()?;
        let file = uri_to_path(&publish_params.uri)?;
        let server_key = ServerKey {
            kind: server_kind.clone(),
            root: root.clone(),
        };
        if self.live_publish_drop_reason(&server_key, &file).is_some() {
            return None;
        }
        let document_version = self.documents.get(&server_key)?.version(&file)?;
        let entry = self
            .diagnostics
            .entries_for_file(&file)
            .into_iter()
            .find_map(|(stored_key, entry)| (stored_key == &server_key).then_some(entry))?;

        Some(AcceptedDiagnosticSnapshot::new(
            server_key,
            document_version,
            entry.diagnostics.clone(),
        ))
    }

    fn live_publish_drop_reason(&self, key: &ServerKey, file: &Path) -> Option<&'static str> {
        let Some(client) = self.clients.get(key) else {
            return Some("server-not-live");
        };
        if client.state() != ServerState::Ready {
            return Some("server-not-ready");
        }
        if client.diagnostics_are_provisional() {
            return Some("producer-warming");
        }
        let Some(document_version) = self
            .documents
            .get(key)
            .and_then(|documents| documents.version(file))
        else {
            return Some("document-not-open");
        };
        let Some(entry) = self
            .diagnostics
            .entries_for_file(file)
            .into_iter()
            .find_map(|(stored_key, entry)| (stored_key == key).then_some(entry))
        else {
            return Some("report-not-stored");
        };
        if entry.stale {
            return Some("report-stale");
        }
        if entry.provisional {
            return Some("report-provisional");
        }
        match entry.version {
            None => Some("missing-version"),
            Some(version) if version != document_version => Some("version-mismatch"),
            Some(_) => None,
        }
    }

    fn handle_event(&mut self, event: &LspEvent) -> Option<PathBuf> {
        let published_file = match event {
            LspEvent::Notification {
                server_kind,
                root,
                method,
                params: Some(params),
            } if method == "textDocument/publishDiagnostics" => {
                self.handle_publish_diagnostics(server_kind.clone(), root.clone(), params)
            }
            LspEvent::Notification {
                server_kind,
                root,
                method,
                params: Some(params),
            } if method == "experimental/serverStatus" => {
                self.handle_server_status(server_kind.clone(), root.clone(), params);
                None
            }
            LspEvent::ServerExited {
                server_kind,
                root,
                reason,
            } => {
                let key = ServerKey {
                    kind: server_kind.clone(),
                    root: root.clone(),
                };
                if let Some(mut client) = self.clients.remove(&key) {
                    match client.reap_after_reader_exit(reason) {
                        ReaderExitReap::AlreadyExited(status) => {
                            slog_info!(
                                "exited {:?} {}: exit status {status} ({reason})",
                                server_kind,
                                root.display()
                            );
                        }
                        ReaderExitReap::KilledWhileAlive => {
                            slog_info!(
                                "exited {:?} {}: reader ended while child alive: {reason}",
                                server_kind,
                                root.display()
                            );
                        }
                    }
                }
                self.server_binaries.remove(&key);
                self.documents.remove(&key);
                self.diagnostics.clear_for_server(&key);
                None
            }
            _ => None,
        };
        self.wake_post_edit_waiters();
        published_file
    }

    fn wake_post_edit_waiters(&mut self) {
        Self::wake_waiters(&mut self.post_edit_waiters);
    }

    fn wake_waiters(waiters: &mut HashMap<u64, Sender<()>>) {
        waiters.retain(|_, sender| match sender.try_send(()) {
            Ok(()) | Err(TrySendError::Full(())) => true,
            Err(TrySendError::Disconnected(())) => false,
        });
    }

    fn handle_publish_diagnostics(
        &mut self,
        server: ServerKind,
        root: PathBuf,
        params: &serde_json::Value,
    ) -> Option<PathBuf> {
        let publish_params = match serde_json::from_value::<lsp_types::PublishDiagnosticsParams>(
            params.clone(),
        ) {
            Ok(params) => params,
            Err(err) => {
                slog_info!(
                    "lsp_protocol server={} root={} method=textDocument/publishDiagnostics event=dropped-because-invalid-params error={}",
                    server.id_str(),
                    root.display(),
                    err
                );
                return None;
            }
        };
        let Some(file) = uri_to_path(&publish_params.uri) else {
            slog_info!(
                "lsp_protocol server={} root={} method=textDocument/publishDiagnostics event=dropped-because-invalid-uri uri={:?}",
                server.id_str(),
                root.display(),
                publish_params.uri
            );
            return None;
        };
        let diagnostic_count = publish_params.diagnostics.len();
        let version = publish_params.version;
        slog_info!(
            "lsp_protocol server={} root={} method=textDocument/publishDiagnostics event=received file={} version={} diagnostics={}",
            server.id_str(),
            root.display(),
            file.display(),
            version
                .map(|version| version.to_string())
                .unwrap_or_else(|| "none".to_string()),
            diagnostic_count
        );
        let stored = from_lsp_diagnostics(file.clone(), publish_params.diagnostics);
        // Store with the real ServerKey and the published document version
        // so observation sources can accept only reports that prove which
        // in-memory document state they describe. The earlier
        // `publish_with_kind` path silently dropped both facts.
        let key = ServerKey { kind: server, root };
        let provisional = self
            .clients
            .get(&key)
            .is_some_and(|client| client.diagnostics_are_provisional());
        self.diagnostics.publish_full_with_provisional(
            key.clone(),
            file.clone(),
            stored,
            None,
            version,
            provisional,
        );
        if let Some(reason) = self.live_publish_drop_reason(&key, &file) {
            slog_info!(
                "lsp_protocol server={} root={} method=textDocument/publishDiagnostics event=dropped-because-{} file={} version={}",
                key.kind.id_str(),
                key.root.display(),
                reason,
                file.display(),
                version
                    .map(|version| version.to_string())
                    .unwrap_or_else(|| "none".to_string())
            );
        }
        Some(file)
    }

    fn handle_server_status(
        &mut self,
        server: ServerKind,
        root: PathBuf,
        params: &serde_json::Value,
    ) {
        if !matches!(&server, ServerKind::Rust)
            || params.get("quiescent").and_then(serde_json::Value::as_bool) != Some(true)
        {
            return;
        }

        let key = ServerKey { kind: server, root };
        let became_quiescent = self
            .clients
            .get_mut(&key)
            .is_some_and(|client| client.set_rust_analyzer_quiescent(true));
        if became_quiescent {
            self.diagnostics.promote_provisional_for_server(&key);
        }
    }

    fn reap_unreferenced_children_for(&self, key: &ServerKey) {
        let live_pids = self
            .clients
            .values()
            .map(LspClient::child_pid)
            .collect::<HashSet<_>>();
        let orphans = self
            .child_registry
            .pids_for_server(&key.root, &key.kind)
            .into_iter()
            .filter(|pid| !live_pids.contains(pid))
            .collect::<Vec<_>>();
        if !orphans.is_empty() {
            self.child_registry.reap_pids(&orphans);
        }
    }

    fn spawn_server(
        &self,
        def: &ServerDef,
        root: &Path,
        source_file: &Path,
        config: &Config,
    ) -> Result<LspClient, LspError> {
        self.spawn_server_with_timeout(def, root, source_file, config, None)
    }

    fn spawn_server_with_timeout(
        &self,
        def: &ServerDef,
        root: &Path,
        source_file: &Path,
        config: &Config,
        initialize_timeout: Option<std::time::Duration>,
    ) -> Result<LspClient, LspError> {
        let initialization_options =
            initialization_options_for_spawn(def, source_file, root, config)?;
        let binary = self.resolve_binary(def, root, config)?;

        // Merge the server-defined env with our test-injected env.
        // `extra_env` is empty in production; tests use it to drive fake
        // server variants (AFT_FAKE_LSP_PULL=1, etc.).
        let mut merged_env = def.env.clone();
        for (key, value) in &self.extra_env {
            merged_env.insert(key.clone(), value.clone());
        }

        // A server may use a nested language workspace, but the reclaim marker
        // belongs to the configured session project. Only register that broader
        // root when it contains the server root; otherwise retain the server root
        // so an unrelated configuration path cannot reap this child.
        let reclaim_root = config
            .project_root
            .as_deref()
            .map(crate::inspect::job::canonicalize_normalized)
            .filter(|project_root| root.starts_with(project_root))
            .unwrap_or_else(|| root.to_path_buf());

        let mut client = LspClient::spawn_with_reclaim_root(
            def.kind.clone(),
            root.to_path_buf(),
            &binary,
            &def.args,
            &merged_env,
            self.event_tx.clone(),
            self.child_registry.clone(),
            Some(&reclaim_root),
        )?;
        let initialize = match initialize_timeout {
            Some(timeout) => client.initialize_with_timeout(root, initialization_options, timeout),
            None => client.initialize(root, initialization_options),
        };
        if let Err(err) = initialize {
            wait_for_stderr_tail(&mut client);
            let stderr_tail = client.stderr_tail();
            let reason = if client.child_exited() || !stderr_tail.is_empty() {
                format_initialize_failure_reason(&def.binary, &stderr_tail, &err)
            } else {
                format!("server failed during initialize: {err}")
            };
            return Err(LspError::ServerNotReady(reason));
        }
        Ok(client)
    }

    fn resolve_binary(
        &self,
        def: &ServerDef,
        root: &Path,
        config: &Config,
    ) -> Result<PathBuf, LspError> {
        if let Some(path) = self.binary_overrides.get(&def.kind) {
            if path.exists() {
                return Ok(path.clone());
            }
            return Err(LspError::NotFound(format!(
                "override binary for {:?} not found: {}",
                def.kind,
                path.display()
            )));
        }

        if let Some(path) = env_binary_override(&def.kind) {
            if path.exists() {
                return Ok(path);
            }
            return Err(LspError::NotFound(format!(
                "environment override binary for {:?} not found: {}",
                def.kind,
                path.display()
            )));
        }

        let mut pushed_config;
        let resolution_config = if let Some(paths) = self.pushed_search_paths.as_ref() {
            pushed_config = config.clone();
            pushed_config.lsp_paths_extra.clone_from(paths);
            &pushed_config
        } else {
            config
        };
        resolve_server_binary(def, Some(root), resolution_config).ok_or_else(|| {
            let searched = if matches!(def.kind, ServerKind::Python | ServerKind::Ty) {
                "the workspace virtualenv, node_modules/.bin, lsp_paths_extra, or PATH"
            } else {
                "node_modules/.bin, lsp_paths_extra, or PATH"
            };
            LspError::NotFound(format!(
                "language server binary '{}' not found in {searched}",
                def.binary,
            ))
        })
    }

    fn server_key_for_file(&self, file_path: &Path, config: &Config) -> Option<ServerKey> {
        for def in servers_for_file(file_path, config) {
            let key = server_key_for_definition(&def, file_path, config)?;
            if self.clients.contains_key(&key) {
                return Some(key);
            }
        }
        None
    }
}

impl Default for LspManager {
    fn default() -> Self {
        Self::new()
    }
}

const ASTRO_TSDK_UNAVAILABLE: &str = "astro-ls requires a project TypeScript install; none found";

fn initialization_options_for_spawn(
    def: &ServerDef,
    source_file: &Path,
    server_root: &Path,
    config: &Config,
) -> Result<Option<serde_json::Value>, LspError> {
    if def.kind != ServerKind::Astro {
        return Ok(def.initialization_options.clone());
    }

    if def
        .initialization_options
        .as_ref()
        .and_then(|options| options.pointer("/typescript/tsdk"))
        .and_then(serde_json::Value::as_str)
        .is_some_and(|tsdk| !tsdk.is_empty())
    {
        return Ok(def.initialization_options.clone());
    }

    // astro-ls does not discover TypeScript itself. Resolve the nearest project
    // installation for the triggering .astro file without walking above the
    // configured project, so monorepo members use their own TypeScript SDK.
    let project_root = config.project_root.as_deref().unwrap_or(server_root);
    let boundary = if source_file.starts_with(project_root) {
        project_root
    } else {
        server_root
    };
    let tsdk = find_project_typescript_sdk(source_file, boundary)
        .ok_or_else(|| LspError::ServerNotReady(ASTRO_TSDK_UNAVAILABLE.to_string()))?;
    let mut options = serde_json::json!({
        "typescript": {
            "tsdk": tsdk.to_string_lossy(),
        }
    });
    if let Some(configured) = def.initialization_options.clone() {
        merge_json_override(&mut options, configured);
    }
    Ok(Some(options))
}

fn find_project_typescript_sdk(source_file: &Path, project_root: &Path) -> Option<PathBuf> {
    let mut directory = source_file.parent()?;
    loop {
        let lib = directory
            .join("node_modules")
            .join("typescript")
            .join("lib");
        if lib.join("tsserverlibrary.js").is_file() || lib.join("typescript.js").is_file() {
            return Some(lib);
        }
        if directory == project_root {
            return None;
        }
        let parent = directory.parent()?;
        if !parent.starts_with(project_root) {
            return None;
        }
        directory = parent;
    }
}

fn merge_json_override(base: &mut serde_json::Value, override_value: serde_json::Value) {
    match (base, override_value) {
        (serde_json::Value::Object(base), serde_json::Value::Object(override_fields)) => {
            for (key, value) in override_fields {
                if let Some(existing) = base.get_mut(&key) {
                    merge_json_override(existing, value);
                } else {
                    base.insert(key, value);
                }
            }
        }
        (base, value) => *base = value,
    }
}

fn wait_for_stderr_tail(client: &mut LspClient) {
    for _ in 0..10 {
        if !client.stderr_tail().is_empty() {
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(10));
    }
}

fn recoverable_pull_rejection(err: &LspError) -> bool {
    matches!(
        err,
        LspError::ServerError {
            code: -32601 | -32602,
            ..
        }
    )
}

fn server_attempt_result_reason(result: &ServerAttemptResult) -> String {
    match result {
        ServerAttemptResult::SpawnFailed { binary, reason } => {
            format!("spawn_failed: {binary} ({reason})")
        }
        ServerAttemptResult::BinaryNotInstalled { binary } => {
            format!("binary_not_installed: {binary}")
        }
        ServerAttemptResult::NoRootMarker { looked_for } => {
            format!("no_root_marker (looked for: {})", looked_for.join(", "))
        }
        ServerAttemptResult::Ok { .. } => "ok".to_string(),
    }
}

fn format_stderr_tail_for_reason(stderr_tail: &str) -> String {
    truncate_stderr_tail_for_reason(stderr_tail)
        .lines()
        .map(|line| format!("  {line}"))
        .collect::<Vec<_>>()
        .join("\n")
}

fn truncate_stderr_tail_for_reason(stderr_tail: &str) -> String {
    if stderr_tail.len() <= STDERR_REASON_BYTES {
        return stderr_tail.to_string();
    }

    let ellipsis = "...";
    let target_len = STDERR_REASON_BYTES.saturating_sub(ellipsis.len());
    let mut start = stderr_tail.len() - target_len;
    while start < stderr_tail.len() && !stderr_tail.is_char_boundary(start) {
        start += 1;
    }
    format!("{ellipsis}{}", &stderr_tail[start..])
}

fn format_initialize_failure_reason(binary: &str, stderr_tail: &str, err: &LspError) -> String {
    let mut reason = format!("server crashed during initialize: {err}");
    if !stderr_tail.is_empty() {
        reason.push_str("; stderr (last 64 lines):\n");
        reason.push_str(&format_stderr_tail_for_reason(stderr_tail));
        reason.push_str("\n\n");
        reason.push_str(&failure_hint(binary, stderr_tail));
    }
    reason
}

fn format_post_initialize_exit_reason(
    binary: &str,
    status: std::process::ExitStatus,
    stderr_tail: &str,
    err: &LspError,
) -> String {
    let code = status
        .code()
        .map(|c| c.to_string())
        .unwrap_or_else(|| "signal/unknown".to_string());
    let mut reason = format!("server exited after initialize (code {code}): {err}");
    if !stderr_tail.is_empty() {
        reason.push_str("; stderr (last 64 lines):\n");
        reason.push_str(&format_stderr_tail_for_reason(stderr_tail));
        reason.push_str("\n\n");
        reason.push_str(&failure_hint(binary, stderr_tail));
    }
    reason
}

fn failure_hint(binary: &str, stderr_tail: &str) -> String {
    if stderr_tail.contains("MODULE_NOT_FOUND") || stderr_tail.contains("Cannot find module") {
        let package_manager = infer_package_manager(stderr_tail);
        format!(
            "Your package-manager shim resolves to a missing file. Try reinstalling: {package_manager} install -g {binary} --force. Common cause: hard-link breakage from fs migration or store prune."
        )
    } else if let Some(component) = rustup_missing_component(stderr_tail) {
        // The binary on PATH is rustup's proxy shim, but the toolchain
        // component isn't installed, so rustup rejects the dispatch with
        // "Unknown binary '<name>' in ... toolchain". The actionable fix is to
        // add the component, not anything about the binary itself.
        format!("'{component}' is a rustup proxy but the component is not installed. Install it: rustup component add {component}")
    } else {
        format!("Hint: see stderr above for '{binary}' failure details.")
    }
}

/// Detect the rustup "proxy shim without installed component" failure and
/// return the component name to add. rustup prints
/// `error: Unknown binary '<name>' in official toolchain '<triple>'` when a
/// `~/.cargo/bin/<name>` proxy is on PATH but the component was never installed
/// (the canonical case is `rust-analyzer`, which ships as an opt-in component).
fn rustup_missing_component(stderr_tail: &str) -> Option<String> {
    let marker = "Unknown binary '";
    let start = stderr_tail.find(marker)? + marker.len();
    let rest = &stderr_tail[start..];
    let end = rest.find('\'')?;
    let name = &rest[..end];
    // Only treat it as a rustup-component issue when the toolchain phrasing is
    // present, so an unrelated "Unknown binary" message doesn't mislead.
    if name.is_empty() || !stderr_tail.contains("toolchain") {
        return None;
    }
    Some(name.to_string())
}

fn infer_package_manager(stderr_tail: &str) -> &'static str {
    let lower = stderr_tail.to_ascii_lowercase();
    if lower.contains(".pnpm/") || lower.contains(".pnpm\\") || lower.contains("/pnpm/") {
        "pnpm"
    } else if lower.contains(".yarn/")
        || lower.contains(".yarn\\")
        || lower.contains("/yarn/")
        || lower.contains("yarn")
    {
        "yarn"
    } else {
        "npm"
    }
}

fn canonicalize_for_lsp(file_path: &Path) -> Result<PathBuf, LspError> {
    // The whole LSP subsystem must agree on ONE canonical form. Workspace
    // roots are normalized (verbatim prefix stripped on Windows) because
    // CreateProcess rejects verbatim cwds; document and watched-file paths
    // are compared against those roots with starts_with, so a bare
    // fs::canonicalize here would produce verbatim paths on Windows that
    // never match any client root.
    std::fs::canonicalize(file_path)
        .map(|canonical| crate::inspect::job::normalize_path(&canonical))
        .map_err(LspError::from)
}

fn resolve_for_lsp_uri(file_path: &Path) -> PathBuf {
    // Same normalized form as canonicalize_for_lsp and the client roots;
    // see the comment there.
    if let Ok(path) = std::fs::canonicalize(file_path) {
        return crate::inspect::job::normalize_path(&path);
    }

    let mut existing = file_path.to_path_buf();
    let mut missing = Vec::new();
    while !existing.exists() {
        let Some(name) = existing.file_name() else {
            break;
        };
        missing.push(name.to_owned());
        let Some(parent) = existing.parent() else {
            break;
        };
        existing = parent.to_path_buf();
    }

    let mut resolved = std::fs::canonicalize(&existing)
        .map(|canonical| crate::inspect::job::normalize_path(&canonical))
        .unwrap_or(existing);
    for segment in missing.into_iter().rev() {
        resolved.push(segment);
    }
    resolved
}

fn language_id_for_extension(ext: &str) -> &'static str {
    match ext {
        "ts" => "typescript",
        "tsx" => "typescriptreact",
        "js" | "mjs" | "cjs" => "javascript",
        "jsx" => "javascriptreact",
        "py" | "pyi" => "python",
        "rs" => "rust",
        "go" => "go",
        "html" | "htm" => "html",
        "md" | "markdown" | "mdx" | "mkd" | "mkdn" | "mdown" | "mdwn" | "qmd" | "rmd" => "markdown",
        _ => "plaintext",
    }
}

fn log_did_open_sent(key: &ServerKey, file: &Path, language_id: &str) {
    slog_info!(
        "lsp_protocol server={} root={} method=textDocument/didOpen event=sent file={} language_id={} version=0",
        key.kind.id_str(),
        key.root.display(),
        file.display(),
        language_id
    );
}

fn normalize_lookup_path(path: &Path) -> PathBuf {
    // Normalized like every other LSP-subsystem path (see canonicalize_for_lsp):
    // store keys and lookups must share one canonical form or Windows verbatim
    // spellings silently miss.
    std::fs::canonicalize(path)
        .map(|canonical| crate::inspect::job::normalize_path(&canonical))
        .unwrap_or_else(|_| path.to_path_buf())
}

fn diagnostic_path_candidates(file: &Path) -> Vec<PathBuf> {
    let mut candidates = Vec::with_capacity(4);
    let mut add = |candidate: PathBuf| {
        if !candidates.iter().any(|existing| existing == &candidate) {
            candidates.push(candidate);
        }
    };

    // Existing files normalize through the same URI path used when publishing
    // diagnostics. A deleted file cannot be canonicalized, so retain the
    // watcher spelling as a candidate too.
    add(file.to_path_buf());
    add(normalize_lookup_path(file));

    // The parent survives a file deletion. Rebuild both forms used by the
    // diagnostics store: raw filesystem canonicalization (legacy/direct
    // publishers) and its non-verbatim normalized spelling (LSP URI events).
    if let (Some(parent), Some(name)) = (file.parent(), file.file_name()) {
        if let Ok(canonical_parent) = std::fs::canonicalize(parent) {
            let reconstructed = canonical_parent.join(name);
            add(reconstructed.clone());
            add(crate::inspect::job::normalize_path(&reconstructed));
        }
    }

    candidates
}

/// Classify an error returned by `spawn_server` into a structured
/// `ServerAttemptResult`. The two interesting cases for callers are:
/// - `BinaryNotInstalled` — the server's binary couldn't be resolved on PATH
///   or via override. The agent can be told "install bash-language-server".
/// - `SpawnFailed` — binary was found but spawning/initializing failed
///   (permissions, missing runtime, server crashed during initialize, etc.).
fn classify_spawn_error(binary: &str, err: &LspError) -> ServerAttemptResult {
    match err {
        // resolve_binary returns NotFound for both missing override paths and
        // missing PATH binaries. The "override missing" case is rare in
        // practice (only set in tests / env vars); we report all NotFound as
        // BinaryNotInstalled so the user sees an actionable install hint.
        LspError::NotFound(_) => ServerAttemptResult::BinaryNotInstalled {
            binary: binary.to_string(),
        },
        other => ServerAttemptResult::SpawnFailed {
            binary: binary.to_string(),
            reason: other.to_string(),
        },
    }
}

fn env_binary_override(kind: &ServerKind) -> Option<PathBuf> {
    env_binary_override_from(kind, |key| std::env::var_os(key))
}

fn env_binary_override_from(
    kind: &ServerKind,
    lookup: impl FnOnce(&str) -> Option<std::ffi::OsString>,
) -> Option<PathBuf> {
    let id = kind.id_str();
    let suffix: String = id
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() {
                ch.to_ascii_uppercase()
            } else {
                '_'
            }
        })
        .collect();
    let key = format!("AFT_LSP_{suffix}_BINARY");
    lookup(&key)
        .filter(|value| !value.is_empty())
        .map(PathBuf::from)
}

#[cfg(test)]
mod env_binary_override_tests {
    use super::*;

    #[test]
    fn empty_lsp_binary_override_is_unset_without_mutating_the_process_environment() {
        let kind = ServerKind::TypeScript;
        assert_eq!(
            env_binary_override_from(&kind, |key| {
                assert_eq!(key, "AFT_LSP_TYPESCRIPT_BINARY");
                Some(std::ffi::OsString::new())
            }),
            None
        );
        assert_eq!(
            env_binary_override_from(&kind, |_| Some(std::ffi::OsString::from("/bin/lsp"))),
            Some(PathBuf::from("/bin/lsp"))
        );
    }
}

#[cfg(test)]
mod language_id_tests {
    use super::language_id_for_extension;

    #[test]
    fn markdown_extensions_use_markdown_language_id() {
        for extension in [
            "md", "markdown", "mdx", "mkd", "mkdn", "mdown", "mdwn", "qmd", "rmd",
        ] {
            assert_eq!(language_id_for_extension(extension), "markdown");
        }
    }
}

#[cfg(all(test, windows))]
mod windows_server_key_tests {
    use std::fs;
    use std::os::windows::ffi::OsStrExt;

    use super::{canonicalize_for_lsp, server_key_for_definition};
    use crate::config::{Config, UserServerDef};
    use crate::lsp::registry::servers_for_file;

    #[test]
    fn normalized_and_verbatim_inputs_produce_identical_server_key_material() {
        let temp_dir = tempfile::tempdir().expect("tempdir");
        let root = temp_dir.path().join("workspace");
        let source = root.join("src").join("main.customts");
        fs::create_dir_all(source.parent().expect("source parent")).expect("create source dir");
        fs::write(root.join("custom-root.json"), "{}\n").expect("write root marker");
        fs::write(&source, "export const value = 1;\n").expect("write source");

        let config = Config {
            project_root: Some(root),
            lsp_servers: vec![UserServerDef {
                id: "custom-ts".to_string(),
                extensions: vec!["customts".to_string()],
                binary: "custom-ts-lsp".to_string(),
                args: Vec::new(),
                root_markers: vec!["custom-root.json".to_string()],
                env: Default::default(),
                initialization_options: None,
                disabled: false,
            }],
            ..Config::default()
        };

        let normalized_input = canonicalize_for_lsp(&source).expect("normalized source path");
        let bare_canonical_input = fs::canonicalize(&source).expect("canonical source path");
        let key_for = |path: &std::path::Path| {
            let def = servers_for_file(path, &config)
                .into_iter()
                .find(|def| def.kind.id_str() == "custom-ts")
                .expect("custom server definition");
            server_key_for_definition(&def, path, &config).expect("custom server root")
        };

        let key_material = |key: &crate::lsp::roots::ServerKey| {
            let root_bytes = key
                .root
                .as_os_str()
                .encode_wide()
                .flat_map(u16::to_le_bytes)
                .collect::<Vec<_>>();
            (key.kind.id_str().to_string(), root_bytes)
        };
        let ensure_key = key_for(&normalized_input);
        let running_lookup_key = key_for(&bare_canonical_input);

        assert_eq!(key_material(&ensure_key), key_material(&running_lookup_key));
    }
}

#[cfg(test)]
mod failure_hint_tests {
    use super::{failure_hint, rustup_missing_component};

    #[test]
    fn detects_rustup_proxy_without_component() {
        // The exact rustup stderr for a proxy shim whose component is missing.
        let stderr = "error: Unknown binary 'rust-analyzer' in official toolchain 'stable-aarch64-apple-darwin'.";
        assert_eq!(
            rustup_missing_component(stderr).as_deref(),
            Some("rust-analyzer")
        );
        let hint = failure_hint("rust-analyzer", stderr);
        assert!(
            hint.contains("rustup component add rust-analyzer"),
            "expected actionable rustup hint, got: {hint}"
        );
    }

    #[test]
    fn ignores_unknown_binary_without_toolchain_phrasing() {
        // "Unknown binary" without the rustup toolchain phrasing must not be
        // misattributed to a rustup component issue.
        let stderr = "fatal: Unknown binary 'foo' was requested by the linker.";
        assert_eq!(rustup_missing_component(stderr), None);
        assert!(failure_hint("foo", stderr).starts_with("Hint: see stderr"));
    }

    #[test]
    fn npm_module_not_found_still_wins() {
        // The existing package-manager-shim case is unaffected.
        let stderr = "Error: Cannot find module '/x/typescript-language-server/lib/cli.mjs'";
        let hint = failure_hint("typescript-language-server", stderr);
        assert!(hint.contains("install -g"), "got: {hint}");
    }
}

#[cfg(test)]
mod diagnostic_capacity_tests {
    use std::fs;

    use super::LspManager;
    use crate::config::Config;

    // The lsp.diagnostic_cache_size config knob must actually take effect:
    // set_diagnostic_capacity (called at AppContext construction with the config
    // value) propagates the cap to the underlying DiagnosticsStore. Before this
    // wiring the field was parsed but never applied (always the hardcoded 5000).
    #[test]
    fn set_diagnostic_capacity_propagates_to_store() {
        let mut manager = LspManager::new();
        manager.set_diagnostic_capacity(7);
        assert_eq!(manager.diagnostics_store_for_test().capacity_for_test(), 7);
        manager.set_diagnostic_capacity(0); // 0 = unbounded
        assert_eq!(manager.diagnostics_store_for_test().capacity_for_test(), 0);
    }

    // configure clears cached spawn failures so a just-installed server retries
    // without a full restart.
    #[test]
    fn clear_failed_spawns_empties_the_cache() {
        let mut manager = LspManager::new();
        assert_eq!(manager.clear_failed_spawns(), 0);
        manager.insert_failed_spawn_for_test();
        assert_eq!(manager.clear_failed_spawns(), 1);
        assert_eq!(manager.clear_failed_spawns(), 0);
    }

    #[test]
    fn pushed_search_paths_make_new_binary_visible_to_stale_config() {
        let root = tempfile::tempdir().unwrap();
        let bin_dir = tempfile::tempdir().unwrap();
        let binary_name = "aft-test-pushed-lsp";
        let binary = bin_dir.path().join(binary_name);
        fs::write(&binary, "#!/bin/sh\nexit 0\n").unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            fs::set_permissions(&binary, fs::Permissions::from_mode(0o755)).unwrap();
        }

        let config = Config {
            project_root: Some(root.path().to_path_buf()),
            lsp_servers: vec![crate::config::UserServerDef {
                id: "pushed-search-path-test".to_string(),
                extensions: vec!["pushedpath".to_string()],
                binary: binary_name.to_string(),
                args: Vec::new(),
                root_markers: Vec::new(),
                env: Default::default(),
                initialization_options: None,
                disabled: false,
            }],
            ..Config::default()
        };
        let file = root.path().join("sample.pushedpath");
        fs::write(&file, "test\n").unwrap();
        let definition = crate::lsp::registry::servers_for_file(&file, &config)
            .into_iter()
            .find(|definition| definition.kind.id_str() == "pushed-search-path-test")
            .unwrap();
        let mut manager = LspManager::new();

        assert!(manager
            .resolve_binary(&definition, root.path(), &config)
            .is_err());
        assert!(manager.set_search_paths(vec![bin_dir.path().to_path_buf()]));
        assert_eq!(
            manager
                .resolve_binary(&definition, root.path(), &config)
                .unwrap(),
            binary
        );
    }

    #[test]
    fn post_write_notification_does_not_start_a_cold_server() {
        let dir = tempfile::tempdir().unwrap();
        let file = dir.path().join("main.ts");
        fs::write(dir.path().join("package.json"), "{}").unwrap();
        fs::write(&file, "export const value = 1;\n").unwrap();

        let mut manager = LspManager::new();
        manager
            .notify_file_changed_if_running(&file, "export const value = 1;\n", &Config::default())
            .unwrap();
        assert!(manager.clients.is_empty());
    }
}

#[cfg(test)]
mod post_edit_waiter_tests {
    use std::collections::HashMap;
    use std::path::PathBuf;
    use std::time::{Duration, Instant};

    use super::LspManager;
    use crate::lsp::client::LspEvent;
    use crate::lsp::registry::ServerKind;

    #[test]
    fn draining_an_event_wakes_registered_post_edit_waiter() {
        let mut manager = LspManager::new();
        let mut wait = manager.start_post_edit_diagnostics_wait(
            PathBuf::from("/workspace/src/main.rs").as_path(),
            &[],
            &HashMap::new(),
            Duration::from_secs(2),
        );
        manager.enqueue_event_for_test(LspEvent::Notification {
            server_kind: ServerKind::Rust,
            root: PathBuf::from("/workspace"),
            method: "custom/drainedElsewhere".to_string(),
            params: None,
        });

        assert_eq!(manager.drain_events().events.len(), 1);
        let started = Instant::now();
        assert!(wait.next_event().is_none());
        assert!(
            started.elapsed() < Duration::from_millis(250),
            "a competing drain did not wake the parked post-edit waiter"
        );
        let _ = manager.poll_post_edit_diagnostics_wait(&mut wait, None);
        let _ = manager.finish_post_edit_diagnostics_wait(wait);
    }
}

#[cfg(test)]
mod clear_diagnostics_tests {
    use std::path::PathBuf;

    use super::LspManager;
    use crate::lsp::client::LspEvent;
    use crate::lsp::diagnostics::{DiagnosticSeverity, StoredDiagnostic};
    use crate::lsp::position::uri_for_path;
    use crate::lsp::registry::ServerKind;
    use crate::lsp::roots::ServerKey;

    fn err_diag(file: &PathBuf) -> StoredDiagnostic {
        StoredDiagnostic {
            file: file.clone(),
            line: 1,
            column: 1,
            end_line: 1,
            end_column: 2,
            severity: DiagnosticSeverity::Error,
            message: "boom".into(),
            code: None,
            source: None,
        }
    }

    // A just-deleted file can no longer be canonicalized directly, but its
    // store key was the canonical path from publish time. The manager must
    // reconstruct that key via the still-present parent dir so symlink-aliased
    // roots (macOS /var -> /private/var) still match and the diagnostic clears.
    #[test]
    fn clear_diagnostics_for_deleted_file_matches_canonical_key() {
        let dir = tempfile::tempdir().unwrap();
        // Canonicalize the parent the way publish time would have.
        let canonical_dir = std::fs::canonicalize(dir.path()).unwrap();
        let canonical_file = canonical_dir.join("gone.ts");
        // Write then remove the file so its parent exists but the file does not,
        // mirroring the post-delete state the watcher observes.
        std::fs::write(&canonical_file, "x").unwrap();

        let mut manager = LspManager::new();
        let key = ServerKey {
            kind: ServerKind::TypeScript,
            root: canonical_dir.clone(),
        };
        manager.diagnostics_store_mut_for_test().publish(
            key,
            canonical_file.clone(),
            vec![err_diag(&canonical_file)],
        );
        assert_eq!(manager.warm_error_warning_counts(), (1, 0));

        std::fs::remove_file(&canonical_file).unwrap();

        // Clear by the NON-canonical path the watcher might hand us (the raw
        // tempdir path, which on macOS differs from the canonical /private form).
        let watcher_path = dir.path().join("gone.ts");
        let removed = manager.clear_diagnostics_for_file(&watcher_path);

        assert!(removed, "expected the deleted file's diagnostic to clear");
        assert_eq!(manager.warm_error_warning_counts(), (0, 0));
    }

    #[cfg(windows)]
    #[test]
    fn clear_diagnostics_for_deleted_file_matches_normalized_publish_key() {
        let dir = tempfile::tempdir().unwrap();
        let file = dir.path().join("normalized-gone.ts");
        std::fs::write(&file, "x").unwrap();
        let normalized_file = crate::inspect::job::canonicalize_normalized(&file);

        let mut manager = LspManager::new();
        let key = ServerKey {
            kind: ServerKind::TypeScript,
            root: normalized_file.parent().unwrap().to_path_buf(),
        };
        manager.diagnostics_store_mut_for_test().publish(
            key,
            normalized_file.clone(),
            vec![err_diag(&normalized_file)],
        );
        std::fs::remove_file(&file).unwrap();

        assert!(manager.clear_diagnostics_for_file(&file));
        assert_eq!(manager.warm_error_warning_counts(), (0, 0));
    }

    #[cfg(windows)]
    #[test]
    fn stale_diagnostics_for_deleted_file_matches_normalized_publish_key() {
        let dir = tempfile::tempdir().unwrap();
        let file = dir.path().join("normalized-stale.ts");
        std::fs::write(&file, "x").unwrap();
        let normalized_file = crate::inspect::job::canonicalize_normalized(&file);

        let mut manager = LspManager::new();
        let key = ServerKey {
            kind: ServerKind::TypeScript,
            root: normalized_file.parent().unwrap().to_path_buf(),
        };
        manager.diagnostics_store_mut_for_test().publish(
            key,
            normalized_file.clone(),
            vec![err_diag(&normalized_file)],
        );
        std::fs::remove_file(&file).unwrap();

        let result = manager.mark_diagnostics_stale_for_file(&file);
        assert!(result.had_entries);
        assert!(result.changed);
        assert_eq!(manager.warm_error_warning_counts(), (0, 0));
    }

    #[test]
    fn clear_diagnostics_for_unknown_file_is_noop() {
        let mut manager = LspManager::new();
        assert!(!manager.clear_diagnostics_for_file(&PathBuf::from("/nope/missing.ts")));
        assert_eq!(manager.warm_error_warning_counts(), (0, 0));
    }

    #[test]
    fn drain_events_reports_publish_diagnostics_updates() {
        let dir = tempfile::tempdir().unwrap();
        let root = std::fs::canonicalize(dir.path()).unwrap();
        let file = root.join("main.ts");
        std::fs::write(&file, "const x: number = 'nope';").unwrap();

        let mut manager = LspManager::new();
        let diagnostic = lsp_types::Diagnostic {
            range: lsp_types::Range {
                start: lsp_types::Position {
                    line: 0,
                    character: 0,
                },
                end: lsp_types::Position {
                    line: 0,
                    character: 1,
                },
            },
            severity: Some(lsp_types::DiagnosticSeverity::ERROR),
            code: None,
            code_description: None,
            source: Some("test".into()),
            message: "boom".into(),
            related_information: None,
            tags: None,
            data: None,
        };
        let params = serde_json::to_value(lsp_types::PublishDiagnosticsParams {
            uri: uri_for_path(&file).unwrap(),
            diagnostics: vec![diagnostic],
            version: Some(1),
        })
        .unwrap();
        manager
            .event_tx
            .send(LspEvent::Notification {
                server_kind: ServerKind::TypeScript,
                root,
                method: "textDocument/publishDiagnostics".into(),
                params: Some(params),
            })
            .unwrap();

        let drained = manager.drain_events();

        assert!(drained.diagnostics_changed);
        assert_eq!(drained.events.len(), 1);
        assert_eq!(manager.warm_error_warning_counts(), (1, 0));
    }
}

#[cfg(test)]
mod inspect_path_tests {
    use super::LspManager;
    use crate::config::{Config, UserServerDef};
    use crate::lsp::registry::ServerKind;

    #[test]
    fn applicability_resolution_does_not_start_or_open_a_server() {
        let temp_dir = tempfile::tempdir().expect("tempdir");
        let root = temp_dir.path().join("project");
        std::fs::create_dir_all(&root).expect("project root");
        std::fs::write(root.join("inspect-root.json"), "{}\n").expect("root marker");
        std::fs::write(root.join("input.inspectlang"), "value\n").expect("source file");

        let config = Config {
            project_root: Some(root.clone()),
            lsp_servers: vec![UserServerDef {
                id: "inspect-test".to_string(),
                extensions: vec!["inspectlang".to_string()],
                binary: "inspect-test-lsp".to_string(),
                args: Vec::new(),
                root_markers: vec!["inspect-root.json".to_string()],
                env: Default::default(),
                initialization_options: None,
                disabled: false,
            }],
            ..Config::default()
        };
        let mut manager = LspManager::new();
        manager.override_binary(
            ServerKind::Custom("inspect-test".into()),
            std::env::current_exe().expect("current executable"),
        );

        let snapshot = manager
            .resolve_applicable_servers_for_root(&root, &config)
            .expect("resolution succeeds without spawning");
        assert_eq!(snapshot.server_keys.len(), 1);
        assert_eq!(snapshot.server_keys[0].kind.id_str(), "inspect-test");
        assert_eq!(manager.server_count(), 0);
        assert!(!manager.document_is_open_for_test(&root.join("input.inspectlang")));
    }

    #[test]
    fn applicability_resolution_preserves_an_empty_snapshot() {
        let temp_dir = tempfile::tempdir().expect("tempdir");
        let root = temp_dir.path().join("project");
        std::fs::create_dir_all(&root).expect("project root");
        std::fs::write(root.join("notes.txt"), "plain text\n").expect("fixture file");

        let snapshot = LspManager::new()
            .resolve_applicable_servers_for_root(&root, &Config::default())
            .expect("an empty applicability set is valid");

        assert!(snapshot.server_keys.is_empty());
        assert!(snapshot.candidates.is_empty());
    }
}

// Every test in this module spawns a real child process, so the module is Unix-only.
#[cfg(all(test, unix))]
mod server_exit_reap_tests {
    use std::collections::HashMap;
    use std::path::Path;
    use std::process::{Child, Command};
    use std::thread;
    use std::time::{Duration, Instant};

    use super::LspManager;
    use crate::config::Config;
    use crate::lsp::child_registry::LspChildRegistry;
    use crate::lsp::client::{LspClient, LspEvent, ServerExitReason};
    use crate::lsp::registry::ServerKind;

    #[cfg(unix)]
    fn spawn_session_sleep() -> Child {
        use std::os::unix::process::CommandExt;

        let mut command = Command::new("sh");
        command.args(["-c", "exec sleep 60"]);
        unsafe {
            command.pre_exec(|| {
                if libc::setsid() == -1 {
                    return Err(std::io::Error::last_os_error());
                }
                Ok(())
            });
        }
        command.spawn().expect("spawn session-leader sleep")
    }

    #[cfg(unix)]
    fn wait_until_dead(pid: u32, timeout: Duration) -> bool {
        let started = Instant::now();
        while started.elapsed() < timeout {
            if !crate::bash_background::process::is_process_alive(pid) {
                return true;
            }
            thread::sleep(Duration::from_millis(20));
        }
        false
    }

    #[cfg(unix)]
    fn spawn_malformed_then_sleep_client(
        event_tx: crossbeam_channel::Sender<LspEvent>,
        registry: LspChildRegistry,
        root: std::path::PathBuf,
    ) -> LspClient {
        LspClient::spawn(
            ServerKind::TypeScript,
            root,
            Path::new("sh"),
            &[
                "-c".to_string(),
                "printf 'Content-Length: 3\r\n\r\n{{{'; exec sleep 60".to_string(),
            ],
            &HashMap::new(),
            event_tx,
            registry,
        )
        .expect("spawn malformed-then-sleep stand-in")
    }

    #[cfg(unix)]
    #[test]
    fn server_exited_handler_kills_live_child_and_untracks() {
        let registry = LspChildRegistry::new();
        let mut manager = LspManager::new();
        manager.set_child_registry(registry.clone());
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().to_path_buf();
        let mut client = spawn_malformed_then_sleep_client(
            manager.event_sender_for_test(),
            registry.clone(),
            root.clone(),
        );
        let pid = client.child_pid();
        client.suppress_kill_on_drop_for_test();
        manager.insert_client_for_test(client);

        let deadline = Instant::now() + Duration::from_secs(5);
        let mut saw_exit = false;
        while Instant::now() < deadline {
            let drained = manager.drain_events();
            if drained.events.iter().any(|event| {
                matches!(
                    event,
                    LspEvent::ServerExited {
                        reason: ServerExitReason::ReadError(_),
                        ..
                    }
                )
            }) {
                saw_exit = true;
                break;
            }
            thread::sleep(Duration::from_millis(20));
        }
        assert!(saw_exit, "reader must emit ServerExited with ReadError");
        assert!(
            wait_until_dead(pid, Duration::from_secs(5)),
            "ServerExited handler must kill the still-running child"
        );
        assert!(
            !registry.pids().contains(&pid),
            "ServerExited handler must untrack the child"
        );
        assert_eq!(manager.active_client_count(), 0);
    }

    #[cfg(unix)]
    #[test]
    fn lifecycle_census_reports_dropped_client_until_reaper_cleans_child() {
        let registry = LspChildRegistry::new();
        let (events, _event_rx) = crossbeam_channel::unbounded();
        let root = tempfile::tempdir().expect("tempdir");
        let mut client =
            spawn_malformed_then_sleep_client(events, registry.clone(), root.path().to_path_buf());
        client.suppress_kill_on_drop_for_test();
        drop(client);

        let leaked = registry.health_snapshot();
        assert_eq!(leaked.children_without_client, 1);
        assert_eq!(leaked.children_total, 1);
        assert_eq!(registry.reap_children_without_client(), 1);
        assert_eq!(registry.health_snapshot().children_without_client, 0);
        assert_eq!(registry.health_snapshot().children_total, 0);
    }

    #[cfg(unix)]
    #[test]
    fn ensure_server_reaps_unreferenced_children_before_spawn() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("workspace");
        let src = root.join("src");
        std::fs::create_dir_all(&src).unwrap();
        std::fs::write(root.join("Cargo.toml"), "[package]\nname = \"demo\"\n").unwrap();
        let main_rs = src.join("main.rs");
        std::fs::write(&main_rs, "fn main() {}\n").unwrap();
        let config = Config::default();
        let rust_def = crate::lsp::registry::servers_for_file(&main_rs, &config)
            .into_iter()
            .find(|def| matches!(def.kind, ServerKind::Rust))
            .expect("rust server applies to main.rs");
        let key = super::server_key_for_definition(&rust_def, &main_rs, &config)
            .expect("rust workspace root");

        let registry = LspChildRegistry::new();
        let mut first = spawn_session_sleep();
        let mut second = spawn_session_sleep();
        let pid1 = first.id();
        let pid2 = second.id();
        registry.track_child(pid1, Some(&key.root), Some(&key.root), Some(&key.kind));
        registry.track_child(pid2, Some(&key.root), Some(&key.root), Some(&key.kind));

        let mut manager = LspManager::new();
        manager.set_child_registry(registry.clone());
        manager.override_binary(ServerKind::Rust, Path::new("false").to_path_buf());
        assert_eq!(
            registry.pids_for_server(&key.root, &key.kind).len(),
            2,
            "both orphans must be registered under the spawn key"
        );
        let _ = manager.ensure_server_for_file(&main_rs, &config);

        let deadline = Instant::now() + Duration::from_secs(5);
        loop {
            let first_exited = first.try_wait().ok().flatten().is_some();
            let second_exited = second.try_wait().ok().flatten().is_some();
            if first_exited && second_exited {
                break;
            }
            assert!(
                Instant::now() < deadline,
                "orphans must exit after rebind reap (first alive={}, second alive={})",
                crate::bash_background::process::is_process_alive(pid1),
                crate::bash_background::process::is_process_alive(pid2)
            );
            thread::sleep(Duration::from_millis(20));
        }
        assert!(
            !registry.pids().contains(&pid1) && !registry.pids().contains(&pid2),
            "reaped orphans must be untracked before the new spawn is tracked"
        );
    }
}