mcpls-core 0.8.0

//! Dead-server detection and respawn-backoff bookkeeping.
//!
//! Tracks consecutive respawn failures per server so a crash-looping
//! process backs off exponentially instead of eating a fresh
//! `timeout_seconds` on every tool call that arrives while it is down.

use std::sync::Arc;
use std::time::Instant;

use tokio::sync::Mutex;
use tokio::time::Duration;

use super::Translator;
use super::restart::{
    NotificationReceivers, NotificationRouting, PushDelivery, reset_server_cache,
};
use super::servers::Backend;
use crate::bridge::indexing::IndexingReset;
use crate::bridge::{DiagnosticsKey, DiagnosticsRole, NotificationCache, on_lifecycle};
use crate::config::{LanguageId, ServerId};
use crate::error::{Error, Result};
use crate::lsp::{LspClient, LspServer, ServerInitConfig, SpawnedServer, VettedInitConfig};
use crate::util::lock_std;

/// Tracks respawn attempts for one server, so [`Translator::respawn_if_dead`]
/// can back off a crash-looping process instead of retrying it on every
/// single tool call.
#[derive(Debug, Clone, Copy)]
pub(super) struct RespawnBackoff {
    /// Number of consecutive attempts that have not produced a server which
    /// stayed alive for at least [`RESPAWN_BACKOFF_BASE`]. A spawn failure
    /// counts immediately; a spawn that succeeds but is found dead again
    /// within that window counts too, once that is discovered -- see
    /// [`Translator::reconcile_respawn_stability`]. Without this, a server
    /// that starts, completes `initialize`, and then crashes a second later
    /// (a common real crash-loop shape) would bypass backoff entirely: each
    /// "success" would otherwise look like a fresh, unbacked-off start.
    consecutive_failures: u32,
    /// The most recent attempt and how it ended.
    last: LastAttempt,
}

/// How the most recent respawn attempt ended, and when.
#[derive(Debug, Clone, Copy)]
pub(super) enum LastAttempt {
    /// An outright spawn failure, or a "successful" respawn found dead again
    /// within the stability window (applied once, so the discovery is not
    /// counted twice).
    Failed(Instant),
    /// The attempt completed `initialize`; whether it held up is not known yet.
    Succeeded(Instant),
}

impl LastAttempt {
    const fn at(self) -> Instant {
        match self {
            Self::Failed(at) | Self::Succeeded(at) => at,
        }
    }
}

impl RespawnBackoff {
    /// The state after a failed attempt at `now`, extending any earlier failures.
    fn after_failure(previous: Option<Self>, now: Instant) -> Self {
        Self {
            consecutive_failures: previous
                .map_or_default(|previous| previous.consecutive_failures)
                .saturating_add(1),
            last: LastAttempt::Failed(now),
        }
    }

    /// The state after an attempt that completed `initialize` at `now`; earlier
    /// failures are kept until the server proves itself stable.
    fn after_success(previous: Option<Self>, now: Instant) -> Self {
        Self {
            consecutive_failures: previous.map_or_default(|previous| previous.consecutive_failures),
            last: LastAttempt::Succeeded(now),
        }
    }
}

/// Base delay before the first backed-off retry after a respawn failure.
const RESPAWN_BACKOFF_BASE: Duration = Duration::from_secs(1);

/// Upper bound on the exponential backoff delay between respawn attempts.
const RESPAWN_BACKOFF_MAX: Duration = Duration::from_secs(30);

impl Translator {
    /// The respawn config of the server tracked under `id`, if it is
    /// registered and dead: its process has exited, or its message loop has
    /// stopped while the process is still running.
    ///
    /// The config is read from the registered [`LspServer`] itself, under the
    /// same `servers` lock as the liveness check, so the two cannot
    /// disagree. Returns `None` ("not dead") for an `id` that isn't
    /// registered at all -- that's the separate `ServerInitializing`/
    /// `NoServerForTool` concern callers already handle, not something the
    /// respawn path should react to -- and for any `try_wait` error, on the
    /// conservative assumption that a health check that itself failed should
    /// not trigger a respawn.
    pub(super) fn dead_server_config(&self, id: &ServerId) -> Option<ServerInitConfig> {
        let mut servers = lock_std(&self.servers);
        let server = servers.server_mut(id)?;
        let config = server.is_dead().ok()?.then(|| server.init_config().clone());
        drop(servers);
        config
    }

    /// Return the shared single-flight lock for `id`, creating it on first
    /// use.
    ///
    /// Two concurrent callers racing to respawn the same server both get a
    /// clone of the *same* underlying `Mutex`, so awaiting it actually
    /// serializes them instead of letting both proceed independently.
    pub(crate) fn respawn_lock(&self, id: &ServerId) -> Arc<Mutex<()>> {
        lock_std(&self.servers).respawn_lock(id)
    }

    /// Remaining backoff delay before `id` may be respawned again, or
    /// `None` if it may be attempted right now.
    ///
    /// Only consults recorded *failures* -- a server with no recorded
    /// attempt is never backed off. A server whose last attempt "succeeded"
    /// is reconciled by [`Self::reconcile_respawn_stability`] (called by
    /// [`Self::respawn_if_dead`] before this) into either a failure (died
    /// again too soon) or removed entirely (proven stable), so by the time
    /// this runs, a lingering "succeeded" entry never reaches here.
    fn respawn_backoff_remaining(&self, id: &ServerId) -> Option<Duration> {
        let (consecutive_failures, last_attempt) = {
            let entry = self.backoff_of(id)?;
            (entry.consecutive_failures, entry.last.at())
        };
        if consecutive_failures == 0 {
            return None;
        }
        let shift = consecutive_failures.saturating_sub(1).min(5);
        let delay = RESPAWN_BACKOFF_BASE
            .saturating_mul(1 << shift)
            .min(RESPAWN_BACKOFF_MAX);
        let elapsed = self.clock.now().saturating_duration_since(last_attempt);
        (elapsed < delay).then(|| delay.saturating_sub(elapsed))
    }

    fn backoff_of(&self, id: &ServerId) -> Option<RespawnBackoff> {
        lock_std(&self.servers).get(id)?.backoff
    }

    /// Replaces the backoff state of `id` with `transition` applied to the
    /// previous one; a server without a slot has no backoff to keep.
    fn update_backoff(
        &self,
        id: &ServerId,
        transition: impl FnOnce(Option<RespawnBackoff>, Instant) -> RespawnBackoff,
    ) {
        let now = self.clock.now();
        if let Some(slot) = lock_std(&self.servers).get_mut(id) {
            slot.backoff = Some(transition(slot.backoff, now));
        }
    }

    /// Records a failed respawn attempt for `id`, extending its backoff.
    fn record_respawn_failure(&self, id: &ServerId) {
        self.update_backoff(id, RespawnBackoff::after_failure);
    }

    /// Records that a respawn attempt for `id` completed `initialize`
    /// successfully.
    ///
    /// Does *not* clear `consecutive_failures`: whether this attempt
    /// actually broke the crash loop is only known once the server either
    /// stays alive for a while or is found dead again -- see
    /// [`Self::reconcile_respawn_stability`], which is what acts on this
    /// entry.
    fn record_respawn_success(&self, id: &ServerId) {
        self.update_backoff(id, RespawnBackoff::after_success);
    }

    /// Reconciles `id`'s backoff state against a *newly observed* death,
    /// before deciding whether to back off this respawn attempt.
    ///
    /// A no-op unless the last recorded attempt "succeeded" ([`Self::record_respawn_success`]):
    /// - If it has since survived at least [`RESPAWN_BACKOFF_BASE`], it is
    ///   treated as proven stable and its backoff state is cleared -- a
    ///   later, unrelated crash starts a fresh backoff sequence rather than
    ///   inheriting history from a long-resolved incident.
    /// - Otherwise, the server died again before proving itself: this
    ///   counts as a failure (extending `consecutive_failures`) instead of
    ///   being silently forgotten. Without this, a server that starts,
    ///   completes `initialize`, and crashes again a moment later would
    ///   bypass backoff entirely -- every such cycle would look like a
    ///   fresh, unbacked-off start, spawning one child process per tool
    ///   call forever.
    fn reconcile_respawn_stability(&self, id: &ServerId) {
        let now = self.clock.now();
        if let Some(slot) = lock_std(&self.servers).get_mut(id)
            && let Some(RespawnBackoff {
                last: LastAttempt::Succeeded(at),
                ..
            }) = slot.backoff
        {
            slot.backoff = if now.saturating_duration_since(at) >= RESPAWN_BACKOFF_BASE {
                None
            } else {
                Some(RespawnBackoff::after_failure(slot.backoff, now))
            };
        }
    }

    /// Detect whether the server routed to `id` has crashed and, if so,
    /// eagerly respawn and re-initialize it before returning.
    ///
    /// A no-op if `id` names a server that was never registered (routing
    /// resolved to it, but it hasn't started yet or never will) or is still
    /// alive.
    ///
    /// # Concurrency
    ///
    /// Multiple callers can race in here for the same `id` -- e.g. two tool
    /// calls landing back-to-back right after the process dies. They
    /// single-flight on [`Self::respawn_lock`]: the first to acquire it
    /// performs the actual respawn; everyone else waits for that attempt to
    /// finish (or fail), rechecks, and finds nothing left to do.
    ///
    /// Requests still parked in the dead client's `pending_requests` are
    /// failed immediately via [`LspClient::fail_pending_requests`] instead
    /// of being left to time out on their own.
    ///
    /// Notifications from the replacement are discarded rather than wired
    /// into the existing pump (see [`NotificationRouting::Discard`]), and the
    /// diagnostics-route server is flagged push-degraded (#359); a manual
    /// `restart_server` re-wires the pump and clears the flag. See
    /// [`Self::respawn_locked`] for the state reset both paths share.
    ///
    /// A crash-looping server (repeated respawn failures) backs off
    /// exponentially (`RESPAWN_BACKOFF_BASE` up to `RESPAWN_BACKOFF_MAX`)
    /// instead of retrying on every single tool call, each of which would
    /// otherwise cost up to a full `timeout_seconds` inside `initialize`.
    ///
    /// # Errors
    ///
    /// Returns [`Error::ServerUnavailable`] if `id` is currently within its
    /// backoff window. Returns whatever error `LspServer::spawn` produced (e.g. its
    /// command is no longer on `PATH`, or `initialize` fails again) if an
    /// actual respawn attempt failed.
    pub(crate) async fn respawn_if_dead(&self, id: &ServerId) -> Result<()> {
        if self.dead_server_config(id).is_none() {
            return Ok(());
        }

        let lock = self.respawn_lock(id);
        let _guard = lock.lock().await;

        // Another caller may have already respawned it while we waited.
        let Some(config) = self.dead_server_config(id) else {
            return Ok(());
        };

        // A panicked message loop never drains its own pending requests.
        let dead_client = lock_std(&self.servers).client(id);
        if let Some(client) = dead_client {
            client.fail_pending_requests();
        }

        self.reconcile_respawn_stability(id);

        tracing::warn!("LSP server '{id}' has crashed, respawning");
        self.respawn_locked(
            id,
            RespawnConfig::Unvetted(config),
            BackoffPolicy::Honor,
            NotificationRouting::Discard,
        )
        .await
    }

    /// Spawn a replacement for `id` from `config` and swap it in for whatever
    /// is registered, with the caller already holding [`Self::respawn_lock`].
    ///
    /// Shared by the crash path ([`Self::respawn_if_dead`]) and the manual
    /// restart. Ordering matters:
    ///
    /// 1. Backoff (when honoured), then spawn. A failure is recorded in the
    ///    backoff bookkeeping.
    /// 2. Waiting for the notification cache. Dropping the call up to here
    ///    kills the replacement and leaves no trace.
    /// 3. The commit, in one step without an await: the success is recorded,
    ///    the previous notification task for `id` is aborted (asynchronously:
    ///    it may still finish one cache write already in progress), `id`'s
    ///    indexing state is reset (a server that reported a readiness signal
    ///    before reads `Loading` until its replacement reports one or the
    ///    indexing-ready timeout elapses), and for the diagnostics-route
    ///    server its cached diagnostics are cleared and its push-degraded
    ///    flag updated per `routing`. The consumer for the replacement's
    ///    notification lanes starts after that, so a readiness signal or
    ///    diagnostics push the replacement emitted during its own handshake,
    ///    already buffered in its channels, is the last write. Then the
    ///    replacement replaces the registered client and server, and
    ///    `document_tracker` forgets `id` (after the swap: a caller still on
    ///    the old client must not record sync against the new generation).
    /// 4. The old client's pending requests fail and subscribers of the
    ///    cleared diagnostics are told to re-read. This tail runs detached,
    ///    so it finishes even if the caller is dropped after the commit.
    ///
    /// # Errors
    ///
    /// [`Error::ServerUnavailable`] when backoff is honoured and active, or
    /// the spawn error of the replacement.
    #[allow(
        clippy::significant_drop_tightening,
        reason = "the cache guard must be held up to the commit, which takes it by value; the lint cannot see the move"
    )]
    pub(super) async fn respawn_locked(
        &self,
        id: &ServerId,
        config: RespawnConfig,
        backoff: BackoffPolicy,
        routing: NotificationRouting<'_>,
    ) -> Result<()> {
        if backoff == BackoffPolicy::Honor
            && let Some(remaining) = self.respawn_backoff_remaining(id)
        {
            tracing::warn!(
                "LSP server '{id}' is crash-looping, backing off for {remaining:?} \
                 before the next respawn attempt"
            );
            return Err(Error::ServerUnavailable {
                server_id: id.clone(),
                retry_in: remaining,
            });
        }

        let attempt = AttemptGuard::begin(self, id);
        let config = match config {
            RespawnConfig::Unvetted(config) => config.for_respawn().await?,
            RespawnConfig::Vetted(vetted) => vetted,
        };
        let language_id = config.server_config().language_id.clone();
        let pinned_tsserver = config.pinned_tsserver();

        let SpawnedServer {
            server: new_server,
            lanes,
        } = LspServer::spawn(config.into_inner()).await?;
        let receivers = NotificationReceivers::new(lanes, pinned_tsserver);

        // The last await before the commit: a drop up to here kills the new
        // child and changes nothing but the backoff, so a half-done respawn
        // cannot exist.
        let cache = match &self.notification_cache {
            Some(cache) => Some(cache.lock().await),
            None => None,
        };
        let (old_client, cleared) =
            self.commit_respawn(attempt, new_server, receivers, routing, &language_id, cache);

        // Must finish once begun, even if the caller goes away: the old
        // waiters need their error and subscribers their invalidation.
        let wiring = self.wiring.get().cloned();
        let finish = crate::util::run_detached(async move {
            if let Some(old_client) = old_client {
                old_client.fail_pending_requests();
            }
            if let Some(wiring) = wiring {
                wiring.publish_invalidated(&cleared).await;
            }
        });
        if finish.await.is_err() {
            tracing::debug!("runtime shut down before the respawn of '{id}' was announced");
        }

        tracing::info!("LSP server '{id}' respawned successfully");
        Ok(())
    }

    /// The synchronous half of a respawn: records the success, resets what the
    /// old process left behind, starts the replacement's consumer and swaps it
    /// in. Has no await point, so it cannot be abandoned half-done.
    ///
    /// Returns the replaced client and the diagnostics keys cleared for it.
    fn commit_respawn(
        &self,
        attempt: AttemptGuard<'_>,
        new_server: LspServer,
        receivers: NotificationReceivers,
        routing: NotificationRouting<'_>,
        language_id: &LanguageId,
        cache: Option<tokio::sync::MutexGuard<'_, NotificationCache>>,
    ) -> (Option<LspClient>, Vec<DiagnosticsKey>) {
        let id = attempt.disarm();
        self.record_respawn_success(id);
        let role = DiagnosticsRole::from_route(self.is_diagnostics_route(language_id, id));

        if let Some(handle) = self.take_notification_task(id) {
            handle.abort();
        }

        // Only the diagnostics-route server ever wrote push notifications to
        // the cache, and `clear_server_diagnostics` is scoped to one server's
        // own entries (#266), so a crashed rust-analyzer never wipes a
        // healthy pyright's cached diagnostics.
        let cleared = cache.map_or_default(|mut cache| {
            let push = match routing {
                NotificationRouting::Pump(_) => PushDelivery::Live,
                NotificationRouting::Discard => {
                    if role == DiagnosticsRole::Authoritative {
                        tracing::warn!(
                            "LSP server '{id}' (language '{language_id}') respawned; its diagnostics \
                             push notifications are discarded rather than cached until it is \
                             restarted with `restart_server` or mcpls is restarted"
                        );
                    }
                    PushDelivery::Degraded
                }
            };
            reset_server_cache(
                &mut cache,
                id,
                IndexingReset::AwaitReplacement {
                    within: self.indexing_ready_timeout,
                },
                role,
                push,
            )
        });

        let consumer = match routing {
            NotificationRouting::Pump(wiring) => wiring.spawn_pump(id.clone(), receivers, role),
            NotificationRouting::Discard => self.spawn_discard_consumer(id, receivers),
        };
        let old_client = self.swap_in(id, new_server, &consumer);
        self.document_tracker.forget_server(id);
        (old_client, cleared)
    }

    /// Installs `new_server` and its consumer in `id`'s slot in one step and
    /// returns the replaced client. Synchronous so the replaced server is
    /// dropped here, after the `servers` guard, never held across an await.
    fn swap_in(
        &self,
        id: &ServerId,
        new_server: LspServer,
        consumer: &tokio::task::AbortHandle,
    ) -> Option<LspClient> {
        let swapped = {
            let mut servers = lock_std(&self.servers);
            let swapped = servers.slot_for_swap(id, Backend::Process(new_server));
            if swapped.is_ok()
                && let Some(slot) = servers.get_mut(id)
            {
                slot.notification_task = Some(consumer.clone());
            }
            drop(servers);
            swapped
        };
        match swapped {
            Ok(old) => old.as_ref().map(|old| old.client().clone()),
            Err(refused) => {
                tracing::warn!(
                    "LSP server '{id}' was shut down while it respawned; dropping the replacement"
                );
                consumer.abort();
                drop(refused);
                None
            }
        }
    }

    /// Drain the replacement's notification lane and forward its lifecycle
    /// lane into the cache, returning the lifecycle forwarder's handle.
    ///
    /// The forwarder has no cancellation hook (the shutdown watch lives in
    /// `serve_with`'s scope), so it writes until its channel closes.
    fn spawn_discard_consumer(
        &self,
        id: &ServerId,
        receivers: NotificationReceivers,
    ) -> tokio::task::AbortHandle {
        let NotificationReceivers {
            notifications,
            mut lifecycle,
            pinned_tsserver,
        } = receivers;
        // Dropping the mailbox reader makes the writer stop buffering; only the
        // log/showMessage channel needs draining.
        let crate::lsp::NotificationInbox {
            mut messages,
            publishes,
        } = notifications;
        drop(publishes);
        tokio::spawn(async move { while messages.recv().await.is_some() {} });
        let cache = self.notification_cache.clone();
        let lifecycle_id = id.clone();
        let forwarder = tokio::spawn(async move {
            while let Some(notif) = lifecycle.recv().await {
                on_lifecycle(
                    cache.as_deref(),
                    &lifecycle_id,
                    pinned_tsserver.as_deref(),
                    notif,
                )
                .await;
            }
        });
        forwarder.abort_handle()
    }
}

/// One respawn attempt in the backoff bookkeeping: unless disarmed before the
/// commit, dropping it counts the attempt as a failure, so a call cancelled
/// mid-respawn and retried backs off instead of paying a full spawn each time.
struct AttemptGuard<'a> {
    translator: &'a Translator,
    id: &'a ServerId,
    armed: bool,
}

impl<'a> AttemptGuard<'a> {
    const fn begin(translator: &'a Translator, id: &'a ServerId) -> Self {
        Self {
            translator,
            id,
            armed: true,
        }
    }

    /// Ends the attempt without counting a failure; hands back the server id.
    fn disarm(mut self) -> &'a ServerId {
        self.armed = false;
        self.id
    }
}

impl Drop for AttemptGuard<'_> {
    fn drop(&mut self) {
        if self.armed {
            self.translator.record_respawn_failure(self.id);
        }
    }
}

/// The config [`Translator::respawn_locked`] spawns from, and whether it was
/// vetted already.
pub(super) enum RespawnConfig {
    /// The config of a server that died; vetted right before the spawn.
    Unvetted(ServerInitConfig),
    /// A config a restart vetted before it stopped the old server, so the
    /// respawn skips the workspace-trust vetting; the spawn itself still
    /// re-runs the argument backstop on purpose (#736).
    Vetted(VettedInitConfig),
}

/// Whether [`Translator::respawn_locked`] honours the crash-loop backoff.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum BackoffPolicy {
    /// Refuse while `id` is inside its backoff window (automatic respawn).
    Honor,
    /// Attempt immediately; the outcome still feeds the bookkeeping (manual restart).
    Bypass,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::bridge::translator::clock::{Clock, FakeClock};
    use crate::config::ServerId;
    #[cfg(unix)]
    use crate::test_lsp::client_path;

    #[test]
    fn test_respawn_backoff_remaining_returns_none_once_delay_elapsed() {
        let clock = Arc::new(FakeClock::new());
        let translator = Translator::new().with_clock(Arc::clone(&clock) as Arc<dyn Clock>);
        let id = ServerId::from_static("rust");

        translator.set_expected_servers(std::collections::HashSet::from([id.clone()]));
        translator.record_respawn_failure(&id);
        assert!(
            translator.respawn_backoff_remaining(&id).is_some(),
            "immediately after a failure, the backoff window must still be active"
        );

        clock.advance(RESPAWN_BACKOFF_MAX);
        assert!(
            translator.respawn_backoff_remaining(&id).is_none(),
            "once the fake clock has advanced past the computed delay, \
             the backoff window must be reported as elapsed"
        );
    }

    #[test]
    fn test_backoff_transitions_keep_failures_until_stability_is_proven() {
        let now = Instant::now();
        let failed = RespawnBackoff::after_failure(None, now);
        assert_eq!(failed.consecutive_failures, 1);
        std::assert_matches!(failed.last, LastAttempt::Failed(_));

        let again = RespawnBackoff::after_failure(Some(failed), now);
        assert_eq!(again.consecutive_failures, 2);

        let succeeded = RespawnBackoff::after_success(Some(again), now);
        assert_eq!(succeeded.consecutive_failures, 2);
        std::assert_matches!(succeeded.last, LastAttempt::Succeeded(_));
        assert_eq!(
            RespawnBackoff::after_success(None, now).consecutive_failures,
            0
        );
    }

    #[test]
    fn test_reconcile_respawn_stability_clears_backoff_after_proven_stable() {
        let clock = Arc::new(FakeClock::new());
        let translator = Translator::new().with_clock(Arc::clone(&clock) as Arc<dyn Clock>);
        let id = ServerId::from_static("rust");

        translator.set_expected_servers(std::collections::HashSet::from([id.clone()]));
        translator.record_respawn_failure(&id);
        translator.record_respawn_success(&id);
        assert!(
            translator.backoff_of(&id).is_some(),
            "a recorded success must still leave a backoff entry pending reconciliation"
        );

        clock.advance(RESPAWN_BACKOFF_BASE);
        translator.reconcile_respawn_stability(&id);

        assert!(
            translator.backoff_of(&id).is_none(),
            "once proven stable (survived at least RESPAWN_BACKOFF_BASE), \
             the backoff entry must be cleared entirely"
        );
    }

    // These three are pure logic (no process spawning), so they run on
    // every platform rather than being swept under `respawn_tests`'s
    // `#[cfg(unix)]` gate below -- otherwise Windows CI would have zero
    // #249 coverage at all.
    #[test]
    fn test_respawn_lock_is_shared_across_lookups_for_same_id() {
        let translator = Translator::new();
        let id = ServerId::from_static("rust");
        translator.set_expected_servers(std::collections::HashSet::from([id.clone()]));

        let first = translator.respawn_lock(&id);
        let second = translator.respawn_lock(&id);

        assert!(
            Arc::ptr_eq(&first, &second),
            "two lookups for the same id must return the same underlying lock, \
             otherwise concurrent respawns would not actually be serialized"
        );
    }

    #[test]
    fn test_unknown_id_has_a_detached_lock_and_no_backoff() {
        let translator = Translator::new();
        let id = ServerId::from_static("ghost");

        assert!(!Arc::ptr_eq(
            &translator.respawn_lock(&id),
            &translator.respawn_lock(&id)
        ));
        translator.record_respawn_failure(&id);
        translator.record_respawn_success(&id);
        assert!(translator.backoff_of(&id).is_none());
        assert!(translator.respawn_backoff_remaining(&id).is_none());
    }

    #[test]
    fn test_respawn_lock_differs_across_ids() {
        let translator = Translator::new();

        let rust_lock = translator.respawn_lock(&ServerId::from_static("rust"));
        let python_lock = translator.respawn_lock(&ServerId::from_static("python"));

        assert!(!Arc::ptr_eq(&rust_lock, &python_lock));
    }

    #[test]
    fn test_dead_server_config_none_when_not_registered() {
        let translator = Translator::new();
        assert!(
            translator
                .dead_server_config(&ServerId::from_static("rust"))
                .is_none()
        );
    }

    // Gated `#[cfg(unix)]`: this module's fake-LSP-server test double is a
    // hand-written `sh` script (POSIX parameter expansion, `printf`-framed
    // LSP responses, file-based invocation counters), which has no
    // equivalent on Windows. CI's "Test (unit)" job matrix includes
    // `windows-latest`.
    #[cfg(unix)]
    mod respawn_tests {
        use std::collections::HashMap;
        use std::{assert_matches, fs};

        use tempfile::TempDir;
        use tokio::time::Duration;

        use super::*;
        use crate::bridge::translator::testing::{
            pid_is_running, stub_server_config, write_crash_after_init_script,
            write_responder_script,
        };
        use crate::bridge::{NotificationWiring, WorkspaceRoots};
        use crate::config::{
            FileExtension, LanguageId, ServerCommand, TimeoutSecs, ToolKind, ToolRouter, ToolSet,
        };
        use crate::lsp::ServerInitConfig;
        use crate::test_lsp::with_read_preamble;

        /// Replaces the registered server's `init_config`, i.e. the config
        /// the next respawn will use.
        fn set_respawn_config(translator: &Translator, id: &ServerId, config: ServerInitConfig) {
            lock_std(&translator.servers)
                .server_mut(id)
                .unwrap()
                .set_init_config(config);
        }

        /// Polls `dead_server_config` until it reports a dead server, bounding the wait
        /// so a broken script fails the test instead of hanging it.
        async fn wait_until_dead(translator: &Translator, id: &ServerId) {
            tokio::time::timeout(Duration::from_secs(2), async {
                loop {
                    if translator.dead_server_config(id).is_some() {
                        return;
                    }
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
            })
            .await
            .expect("seed server never reported as exited");
        }

        /// #542: respawning a crashed server kills the descendants it left
        /// behind in its process group.
        #[tokio::test]
        async fn test_respawn_if_dead_kills_descendants_of_the_crashed_server() {
            let dir = TempDir::new().unwrap();
            let pid_file = dir.path().join("grandchild.pid");
            let crashing = dir.path().join("crashing.sh");
            let body = with_read_preamble(&format!(
                r#"sleep 600 &
echo $! > '{}'
body='{{"jsonrpc":"2.0","id":1,"result":{{"capabilities":{{}}}}}}'
printf 'Content-Length: %d\r\n\r\n%s' ${{#body}} "$body"
sleep 0.3
"#,
                pid_file.display()
            ));
            fs::write(&crashing, body).unwrap();
            let id = ServerId::from_static("rust");

            let server = LspServer::spawn(stub_server_config("rust", &crashing))
                .await
                .unwrap()
                .server;
            let translator = Translator::new();
            translator.register_server_complete(server);
            wait_until_dead(&translator, &id).await;
            let grandchild: u32 = fs::read_to_string(&pid_file)
                .unwrap()
                .trim()
                .parse()
                .unwrap();
            assert!(pid_is_running(grandchild));

            let responder = write_responder_script(dir.path(), 5);
            set_respawn_config(&translator, &id, stub_server_config("rust", &responder));
            translator.respawn_if_dead(&id).await.unwrap();

            let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
            while pid_is_running(grandchild) {
                assert!(
                    tokio::time::Instant::now() < deadline,
                    "descendant of the crashed server survived the respawn"
                );
                tokio::time::sleep(Duration::from_millis(50)).await;
            }
        }

        #[tokio::test]
        async fn test_respawn_if_dead_noop_when_server_alive() {
            let dir = TempDir::new().unwrap();
            let script = write_responder_script(dir.path(), 1);
            let id = ServerId::from_static("rust");
            let config = stub_server_config("rust", &script);

            let server = LspServer::spawn(config).await.unwrap().server;
            let translator = Translator::new();
            translator.register_server_complete(server);

            assert!(translator.dead_server_config(&id).is_none());
            assert!(translator.respawn_if_dead(&id).await.is_ok());
        }

        /// A live child behind a stopped message loop (e.g. a panicked loop)
        /// is dead for routing purposes and must be respawned.
        #[cfg(unix)]
        #[tokio::test]
        async fn test_respawn_if_dead_replaces_server_with_stopped_message_loop() {
            let dir = TempDir::new().unwrap();
            let script = write_responder_script(dir.path(), 5);
            let id = ServerId::from_static("rust");

            let seed = crate::lsp::fake_lsp_server_with_dead_loop_and_live_child();
            let translator = Translator::new();
            translator.register_client(id.clone(), seed.client().clone());
            translator.register_server(id.clone(), seed);
            set_respawn_config(&translator, &id, stub_server_config("rust", &script));
            wait_until_dead(&translator, &id).await;

            translator.respawn_if_dead(&id).await.unwrap();

            assert!(translator.dead_server_config(&id).is_none());
        }

        /// #529: the respawn config is the registered server's own
        /// `init_config`, with no separately stored copy.
        #[tokio::test]
        async fn test_respawn_if_dead_uses_config_of_registered_server() {
            let dir = TempDir::new().unwrap();
            let script = write_crash_after_init_script(dir.path());
            let id = ServerId::from_static("rust");
            let config = stub_server_config("rust", &script);

            let server = LspServer::spawn(config.clone()).await.unwrap().server;
            let translator = Translator::new();
            translator.register_server_complete(server);
            wait_until_dead(&translator, &id).await;

            let dead = translator.dead_server_config(&id).unwrap();
            assert_eq!(dead.server_config().args, config.server_config().args);

            translator.respawn_if_dead(&id).await.unwrap();
        }

        /// #566: a respawned server that falls through to the workspace
        /// tsserver despite a configured pin is warned about by the respawn
        /// lifecycle forwarder, not only by the startup pump.
        #[tokio::test]
        async fn test_respawn_warns_when_pinned_tsserver_is_ignored() {
            use tracing_subscriber::layer::SubscriberExt as _;

            use crate::test_lsp::{CapturedLogs, with_read_preamble};

            let dir = TempDir::new().unwrap();
            let crash = write_crash_after_init_script(dir.path());
            let id = ServerId::from_static("rust");
            let server = LspServer::spawn(stub_server_config("rust", &crash))
                .await
                .unwrap()
                .server;
            let translator = Translator::new();
            translator.register_client(id.clone(), server.client().clone());
            translator.register_server(id.clone(), server);
            wait_until_dead(&translator, &id).await;

            let reporter = dir.path().join("reporter.sh");
            fs::write(
                &reporter,
                with_read_preamble(
                    r#"body='{"jsonrpc":"2.0","id":1,"result":{"capabilities":{}}}'
printf 'Content-Length: %d\r\n\r\n%s' ${#body} "$body"
note='{"jsonrpc":"2.0","method":"$/typescriptVersion","params":{"version":"5.0","source":"workspace"}}'
printf 'Content-Length: %d\r\n\r\n%s' ${#note} "$note"
sleep 5
"#,
                ),
            )
            .unwrap();
            let mut pinned = stub_server_config("rust", &reporter);
            pinned.server_config_mut().initialization_options =
                Some(serde_json::json!({"tsserver": {"path": "/pin/tsserver.js"}}));
            set_respawn_config(&translator, &id, pinned);

            let captured = CapturedLogs::default();
            let _guard = tracing::subscriber::set_default(
                tracing_subscriber::registry().with(captured.clone()),
            );
            translator.respawn_if_dead(&id).await.unwrap();

            tokio::time::timeout(Duration::from_secs(5), async {
                while !captured
                    .messages()
                    .iter()
                    .any(|m| m.contains("ignored the configured tsserver.path"))
                {
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
            })
            .await
            .expect("respawned server's ignored pin was never warned about");
        }

        #[tokio::test]
        async fn test_respawn_if_dead_propagates_spawn_failure() {
            let dir = TempDir::new().unwrap();
            let script = write_crash_after_init_script(dir.path());
            let id = ServerId::from_static("rust");
            let seed_config = stub_server_config("rust", &script);

            let server = LspServer::spawn(seed_config).await.unwrap().server;
            let translator = Translator::new();
            translator.register_client(id.clone(), server.client().clone());
            translator.register_server(id.clone(), server);
            wait_until_dead(&translator, &id).await;

            let mut broken = stub_server_config("rust", &script);
            broken.server_config_mut().command =
                ServerCommand::from_static("nonexistent-lsp-cmd-xyz").into();
            set_respawn_config(&translator, &id, broken);

            let err = translator.respawn_if_dead(&id).await.unwrap_err();
            assert_matches!(err, Error::ServerNotFound { .. }, "got {err:?}");
        }

        /// #249: two concurrent tool calls that both observe the same dead
        /// server must not each perform their own respawn -- only one
        /// replacement process should ever be spawned, and both callers
        /// must still resolve successfully.
        ///
        /// The fake server script counts every invocation and, on its
        /// first run only, exits right after answering `initialize`
        /// (simulating "was alive, then crashed"); every later invocation
        /// answers and then sleeps, standing in for a healthy replacement.
        /// If single-flighting were broken, both concurrent callers would
        /// spawn their own replacement and the invocation count would be
        /// 3 (seed + two independent respawns) instead of 2 (seed + one
        /// shared respawn).
        #[tokio::test]
        async fn test_respawn_if_dead_single_flights_concurrent_callers() {
            let dir = TempDir::new().unwrap();
            let marker = dir.path().join("marker");
            let counter = dir.path().join("invocations");
            let script_path = dir.path().join("flaky.sh");
            let template = with_read_preamble(
                r#"echo x >> "__COUNTER__"
if [ -f "__MARKER__" ]; then
  body='{"jsonrpc":"2.0","id":1,"result":{"capabilities":{}}}'
  printf 'Content-Length: %d\r\n\r\n%s' ${#body} "$body"
  sleep 1
else
  touch "__MARKER__"
  body='{"jsonrpc":"2.0","id":1,"result":{"capabilities":{}}}'
  printf 'Content-Length: %d\r\n\r\n%s' ${#body} "$body"
  sleep 0.3
fi
"#,
            );
            let script_body = template
                .replace("__COUNTER__", &counter.display().to_string())
                .replace("__MARKER__", &marker.display().to_string());
            fs::write(&script_path, script_body).unwrap();

            let id = ServerId::from_static("rust");
            let config = stub_server_config("rust", &script_path);

            let seed = LspServer::spawn(config.clone()).await.unwrap().server;
            let translator = Arc::new(Translator::new());
            translator.register_client(id.clone(), seed.client().clone());
            translator.register_server(id.clone(), seed);
            set_respawn_config(&translator, &id, config);
            wait_until_dead(&translator, &id).await;

            let (t1, id1) = (Arc::clone(&translator), id.clone());
            let (t2, id2) = (Arc::clone(&translator), id.clone());
            let (r1, r2) = tokio::join!(
                tokio::spawn(async move { t1.respawn_if_dead(&id1).await }),
                tokio::spawn(async move { t2.respawn_if_dead(&id2).await }),
            );
            assert!(r1.unwrap().is_ok());
            assert!(r2.unwrap().is_ok());

            let invocations = fs::read_to_string(&counter).unwrap();
            assert_eq!(
                invocations.lines().count(),
                2,
                "expected exactly one seed spawn + one single-flighted \
                 respawn, got:\n{invocations}"
            );
        }

        /// #249 S2 regression: a second `respawn_if_dead` call within the
        /// backoff window must fail fast via `Error::ServerUnavailable`
        /// instead of repeating a real spawn attempt -- proven by the
        /// *kind* of error changing between the two calls, not by timing:
        /// the first call's failure is the genuine `LspServer::spawn` error
        /// (`Error::ServerNotFound`, from a command that does not
        /// exist), and the second, immediately following, is the distinct
        /// backoff error.
        #[tokio::test]
        async fn test_respawn_if_dead_backs_off_after_repeated_failure() {
            let dir = TempDir::new().unwrap();
            let seed_script = write_crash_after_init_script(dir.path());
            let id = ServerId::from_static("rust");
            let seed_config = stub_server_config("rust", &seed_script);

            let seed = LspServer::spawn(seed_config).await.unwrap().server;
            let clock = Arc::new(FakeClock::new());
            let translator = Translator::new().with_clock(Arc::clone(&clock) as Arc<dyn Clock>);
            translator.register_client(id.clone(), seed.client().clone());
            translator.register_server(id.clone(), seed);
            wait_until_dead(&translator, &id).await;

            let mut broken = stub_server_config("rust", &seed_script);
            broken.server_config_mut().command =
                ServerCommand::from_static("nonexistent-lsp-cmd-xyz").into();
            set_respawn_config(&translator, &id, broken);

            let err1 = translator.respawn_if_dead(&id).await.unwrap_err();
            assert_matches!(
                err1,
                Error::ServerNotFound { .. },
                "first attempt should be a real (failed) spawn, got {err1:?}"
            );

            let err2 = translator.respawn_if_dead(&id).await.unwrap_err();
            assert_matches!(
                err2,
                Error::ServerUnavailable { .. },
                "second call within the backoff window must fail fast \
                 without attempting another real spawn, got {err2:?}"
            );
        }

        /// #292 regression: once the backoff window has elapsed, the next
        /// `respawn_if_dead` call must actually attempt a fresh respawn
        /// instead of continuing to fail fast -- proven by swapping in a
        /// config that succeeds and observing `Ok(())`, not merely a
        /// different error kind.
        #[tokio::test]
        async fn test_respawn_if_dead_reattempts_once_backoff_window_elapses() {
            let dir = TempDir::new().unwrap();
            let seed_script = write_crash_after_init_script(dir.path());
            let id = ServerId::from_static("rust");
            let seed_config = stub_server_config("rust", &seed_script);

            let seed = LspServer::spawn(seed_config).await.unwrap().server;
            let clock = Arc::new(FakeClock::new());
            let translator = Translator::new().with_clock(Arc::clone(&clock) as Arc<dyn Clock>);
            translator.register_client(id.clone(), seed.client().clone());
            translator.register_server(id.clone(), seed);
            wait_until_dead(&translator, &id).await;

            let mut broken = stub_server_config("rust", &seed_script);
            broken.server_config_mut().command =
                ServerCommand::from_static("nonexistent-lsp-cmd-xyz").into();
            set_respawn_config(&translator, &id, broken);

            let err1 = translator.respawn_if_dead(&id).await.unwrap_err();
            assert_matches!(
                err1,
                Error::ServerNotFound { .. },
                "first attempt should be a real (failed) spawn, got {err1:?}"
            );

            let err2 = translator.respawn_if_dead(&id).await.unwrap_err();
            assert_matches!(
                err2,
                Error::ServerUnavailable { .. },
                "second call within the backoff window must still fail fast, got {err2:?}"
            );

            // Advance well past the computed backoff delay and swap in a
            // config that will actually succeed this time.
            clock.advance(RESPAWN_BACKOFF_MAX);
            let working_script = write_crash_after_init_script(dir.path());
            set_respawn_config(
                &translator,
                &id,
                stub_server_config("rust", &working_script),
            );

            let result = translator.respawn_if_dead(&id).await;
            assert!(
                result.is_ok(),
                "once the backoff window has elapsed, respawn_if_dead must actually \
                 reattempt a respawn instead of continuing to short-circuit, got {result:?}"
            );
        }

        /// #249 R3 regression: a respawn that *succeeds* (completes
        /// `initialize`) but dies again almost immediately must still
        /// engage backoff -- this is the more realistic crash-loop shape
        /// (start, initialize, then OOM-die a second later) than an
        /// outright spawn failure, and without this fix every such cycle
        /// looked like a fresh, unbacked-off start, spawning one child
        /// process per tool call forever.
        #[tokio::test]
        async fn test_respawn_if_dead_backs_off_after_quick_recrash_following_success() {
            let dir = TempDir::new().unwrap();
            let seed_script = write_crash_after_init_script(dir.path());
            let id = ServerId::from_static("rust");
            let seed_config = stub_server_config("rust", &seed_script);

            let seed = LspServer::spawn(seed_config).await.unwrap().server;
            let clock = Arc::new(FakeClock::new());
            let translator = Translator::new().with_clock(Arc::clone(&clock) as Arc<dyn Clock>);
            translator.register_client(id.clone(), seed.client().clone());
            translator.register_server(id.clone(), seed);
            wait_until_dead(&translator, &id).await;

            // Reuse the same crash-after-init script as the respawn target:
            // every attempt completes `initialize` successfully, then dies
            // ~0.3s later -- a post-init crash loop, not a spawn failure.
            set_respawn_config(&translator, &id, stub_server_config("rust", &seed_script));

            translator
                .respawn_if_dead(&id)
                .await
                .expect("the replacement completes initialize, so this attempt succeeds");
            wait_until_dead(&translator, &id).await;

            let err = translator.respawn_if_dead(&id).await.unwrap_err();
            assert_matches!(
                err,
                Error::ServerUnavailable { .. },
                "a respawn that dies again within the stability window must \
                 back off instead of being treated as a fresh attempt, got {err:?}"
            );
        }

        /// #249 C1 regression: respawning the *diagnostics-route* server
        /// for a language must invalidate that server's diagnostics cache
        /// entries, rather than leaving stale entries to be merged into
        /// fresh pull results as if still current -- the crashed process's
        /// pump is gone and will never update or clear them itself.
        ///
        /// Covers the "under-clear" failure mode a scoped-to-synced-URIs
        /// clear has: a real diagnostics-route server (e.g. rust-analyzer)
        /// publishes workspace-wide (`cargo check` results for files never
        /// opened through mcpls), so `never_opened_uri` below stands in for
        /// an entry that must still be cleared despite never having gone
        /// through `ensure_open`.
        ///
        /// #266 S2 regression (over-clear direction, multi-language case):
        /// `other_language_uri` is owned by a *different* diagnostics-route
        /// server (e.g. pyright for Python, in the same workspace as the
        /// rust-analyzer under test here) and must survive -- `clear_server_diagnostics`
        /// replaced a workspace-wide `clear_all_diagnostics` that used to
        /// wipe every language's cache on any single server's respawn.
        #[tokio::test]
        async fn test_respawn_if_dead_clears_diagnostics_cache_when_diagnostics_route() {
            let dir = TempDir::new().unwrap();
            let seed_script = write_crash_after_init_script(dir.path());
            let id = ServerId::from_static("rust");
            let seed_config = stub_server_config("rust", &seed_script);

            let seed = LspServer::spawn(seed_config).await.unwrap().server;

            let cache = Arc::new(Mutex::new(crate::bridge::NotificationCache::new()));
            let translator = Translator::new()
                .with_router(ToolRouter::catch_all([(
                    id.clone(),
                    LanguageId::from_static("rust"),
                )]))
                .with_notification_cache(Arc::clone(&cache));
            translator.register_client(id.clone(), seed.client().clone());
            translator.register_server(id.clone(), seed);

            let synced_uri: lsp_types::Uri = lsp_types::Uri::from("file:///workspace/opened.rs");
            let never_opened_uri: lsp_types::Uri =
                lsp_types::Uri::from("file:///workspace/never_opened.rs");
            let other_language_uri: lsp_types::Uri =
                lsp_types::Uri::from("file:///workspace/main.py");
            cache
                .lock()
                .await
                .store_diagnostics(&id, &synced_uri, None, vec![]);
            cache
                .lock()
                .await
                .store_diagnostics(&id, &never_opened_uri, None, vec![]);
            cache.lock().await.store_diagnostics(
                &ServerId::from_static("python"),
                &other_language_uri,
                None,
                vec![],
            );

            wait_until_dead(&translator, &id).await;

            let respawn_script = write_responder_script(dir.path(), 1);
            set_respawn_config(
                &translator,
                &id,
                stub_server_config("rust", &respawn_script),
            );

            translator.respawn_if_dead(&id).await.unwrap();

            let guard = cache.lock().await;
            assert!(
                guard.diagnostics(&synced_uri).is_none(),
                "diagnostics attributed to the crashed connection must be \
                 invalidated on respawn, not served as current"
            );
            assert!(
                guard.diagnostics(&never_opened_uri).is_none(),
                "workspace-wide diagnostics for a file mcpls never opened \
                 must also be invalidated, not just synced documents"
            );
            assert!(
                guard.diagnostics(&other_language_uri).is_some(),
                "a different diagnostics-route server's entries must survive \
                 an unrelated server's respawn-triggered cache clear"
            );
            assert!(
                guard.is_push_degraded(&id),
                "#359: the respawned diagnostics-route server must be marked as \
                 push-degraded, since its replacement's notifications are discarded"
            );
            assert!(
                !guard.is_push_degraded(&ServerId::from_static("python")),
                "an unrelated, never-respawned server must not be marked degraded"
            );
            drop(guard);
        }

        /// A respawned server that reported a readiness signal before must
        /// read `Loading` until its replacement reports one (#667), not carry
        /// over the stale `Ready` of the crashed connection -- the
        /// replacement has indexed nothing yet, so a stale `Ready` or an open
        /// `Unknown` would let whole-workspace queries through against an
        /// empty index. Unlike diagnostics-cache clearing, this must happen
        /// regardless of diagnostics-route status -- indexing readiness gates
        /// every routed server's whole-workspace tools, not just the one that
        /// owns diagnostics.
        #[tokio::test]
        async fn test_respawn_if_dead_awaits_replacement_indexing_signal() {
            let dir = TempDir::new().unwrap();
            let seed_script = write_crash_after_init_script(dir.path());
            let id = ServerId::from_static("rust");
            let seed_config = stub_server_config("rust", &seed_script);

            let seed = LspServer::spawn(seed_config).await.unwrap().server;

            let cache = Arc::new(Mutex::new(crate::bridge::NotificationCache::new()));
            let translator = Translator::new()
                .with_router(ToolRouter::catch_all([(
                    id.clone(),
                    LanguageId::from_static("rust"),
                )]))
                .with_notification_cache(Arc::clone(&cache));
            translator.register_client(id.clone(), seed.client().clone());
            translator.register_server(id.clone(), seed);

            cache.lock().await.observe_indexing_signal(
                &id,
                "experimental/serverStatus",
                Some(&serde_json::json!({"quiescent": true})),
            );
            assert_eq!(
                cache.lock().await.indexing_state(&id),
                crate::bridge::IndexingState::Ready
            );

            wait_until_dead(&translator, &id).await;

            let respawn_script = write_responder_script(dir.path(), 1);
            set_respawn_config(
                &translator,
                &id,
                stub_server_config("rust", &respawn_script),
            );

            translator.respawn_if_dead(&id).await.unwrap();

            assert_eq!(
                cache.lock().await.indexing_state(&id),
                crate::bridge::IndexingState::Loading,
                "a respawned server that reported signals before is loading until its \
                 replacement reports one"
            );
        }

        /// #667: a server that never reported a readiness signal is not
        /// gated after a respawn (no new fixed delay, bridge/012 US-004).
        #[tokio::test]
        async fn test_respawn_if_dead_without_signal_history_stays_unknown() {
            let dir = TempDir::new().unwrap();
            let seed_script = write_crash_after_init_script(dir.path());
            let id = ServerId::from_static("rust");
            let seed = LspServer::spawn(stub_server_config("rust", &seed_script))
                .await
                .unwrap()
                .server;

            let cache = Arc::new(Mutex::new(crate::bridge::NotificationCache::new()));
            let translator = Translator::new()
                .with_router(ToolRouter::catch_all([(
                    id.clone(),
                    LanguageId::from_static("rust"),
                )]))
                .with_notification_cache(Arc::clone(&cache));
            translator.register_client(id.clone(), seed.client().clone());
            translator.register_server(id.clone(), seed);
            wait_until_dead(&translator, &id).await;

            let respawn_script = write_responder_script(dir.path(), 1);
            set_respawn_config(
                &translator,
                &id,
                stub_server_config("rust", &respawn_script),
            );
            translator.respawn_if_dead(&id).await.unwrap();

            assert_eq!(
                cache.lock().await.indexing_state(&id),
                crate::bridge::IndexingState::Unknown
            );
        }

        /// #425 regression: the *replacement* process's own lifecycle-lane
        /// notifications (`$/progress` and `experimental/serverStatus`) must
        /// be wired into the shared `NotificationCache`, not drained and
        /// discarded like before -- otherwise a respawned server can never
        /// re-acquire indexing readiness and stays `Unknown`/fail-open for
        /// the rest of the process's life. The respawn script answers
        /// `initialize` and *immediately* (no delay) emits its own
        /// `experimental/serverStatus` notification, so it is already
        /// buffered in the lifecycle channel by the time `respawn_if_dead`
        /// gets to wiring it up -- exactly like a real rust-analyzer
        /// replacement reporting readiness during its own handshake. Runs on
        /// a multi-thread runtime (matching `crates/mcpls-cli/src/main.rs`'s
        /// production `#[tokio::main]`) so the forwarding task can race
        /// `respawn_if_dead`'s own continuation for real, which is what
        /// exposed the S1 ordering bug this test is meant to catch: with
        /// `reset_indexing_state` (wrongly) called *after* the forwarder was
        /// spawned, the buffered signal could be drained and immediately
        /// erased before this test ever observes it.
        #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
        async fn test_respawn_if_dead_reacquires_indexing_state_from_replacement() {
            let dir = TempDir::new().unwrap();
            let seed_script = write_crash_after_init_script(dir.path());
            let id = ServerId::from_static("rust");
            let seed_config = stub_server_config("rust", &seed_script);

            let seed = LspServer::spawn(seed_config).await.unwrap().server;

            let cache = Arc::new(Mutex::new(crate::bridge::NotificationCache::new()));
            let translator = Translator::new()
                .with_router(ToolRouter::catch_all([(
                    id.clone(),
                    LanguageId::from_static("rust"),
                )]))
                .with_notification_cache(Arc::clone(&cache));
            translator.register_client(id.clone(), seed.client().clone());
            translator.register_server(id.clone(), seed);
            wait_until_dead(&translator, &id).await;

            let respawn_script_path = dir.path().join("respawn_with_status.sh");
            let respawn_script_body = with_read_preamble(
                r#"body='{"jsonrpc":"2.0","id":1,"result":{"capabilities":{}}}'
notif='{"jsonrpc":"2.0","method":"experimental/serverStatus","params":{"quiescent":true}}'
printf 'Content-Length: %d\r\n\r\n%sContent-Length: %d\r\n\r\n%s' ${#body} "$body" ${#notif} "$notif"
sleep 1
"#,
            );
            fs::write(&respawn_script_path, respawn_script_body).unwrap();
            set_respawn_config(
                &translator,
                &id,
                stub_server_config("rust", &respawn_script_path),
            );

            translator.respawn_if_dead(&id).await.unwrap();

            let observed_ready = tokio::time::timeout(Duration::from_secs(2), async {
                loop {
                    if cache.lock().await.indexing_state(&id) == crate::bridge::IndexingState::Ready
                    {
                        return;
                    }
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
            })
            .await;

            assert!(
                observed_ready.is_ok(),
                "the replacement's own experimental/serverStatus notification must reach \
                 the NotificationCache, not be discarded"
            );
        }

        /// Code-review regression: `respawn_if_dead` must actually *abort*
        /// any previously-registered lifecycle-forwarder handle for `id`,
        /// not merely replace the map entry (which an earlier version of
        /// this test could not tell apart from a forwarder that happened to
        /// finish on its own -- e.g. because its channel closed naturally).
        /// Pre-registers a handle to a task built from [`std::future::pending`],
        /// which by construction can *never* complete except via
        /// cancellation, so observing it end after a respawn is unambiguous
        /// proof that `respawn_if_dead` called `.abort()` on it.
        #[tokio::test]
        async fn test_respawn_if_dead_aborts_previous_forwarder_handle() {
            let dir = TempDir::new().unwrap();
            let seed_script = write_crash_after_init_script(dir.path());
            let id = ServerId::from_static("rust");
            let seed_config = stub_server_config("rust", &seed_script);

            let seed = LspServer::spawn(seed_config).await.unwrap().server;
            let translator = Translator::new().with_router(ToolRouter::catch_all([(
                id.clone(),
                LanguageId::from_static("rust"),
            )]));
            translator.register_client(id.clone(), seed.client().clone());
            translator.register_server(id.clone(), seed);
            wait_until_dead(&translator, &id).await;

            let never_completes = tokio::spawn(std::future::pending::<()>());
            lock_std(&translator.servers)
                .get_mut(&id)
                .unwrap()
                .notification_task = Some(never_completes.abort_handle());

            let respawn_script = write_responder_script(dir.path(), 1);
            set_respawn_config(
                &translator,
                &id,
                stub_server_config("rust", &respawn_script),
            );
            translator.respawn_if_dead(&id).await.unwrap();

            let outcome = tokio::time::timeout(Duration::from_secs(2), never_completes)
                .await
                .expect("respawn_if_dead must abort the stale handle promptly, not hang it");
            assert!(
                outcome.is_err_and(|join_err| join_err.is_cancelled()),
                "a task that can never complete on its own must have been aborted \
                 by the later respawn, not merely replaced in the map"
            );

            assert!(
                lock_std(&translator.servers)
                    .get(&id)
                    .and_then(|slot| slot.notification_task.as_ref())
                    .is_some_and(|current| !current.is_finished()),
                "the new respawn's own forwarder must be registered and still running"
            );
        }

        /// #249 C1 regression (over-clear direction): respawning a server
        /// that is *not* the diagnostics route for its language must not
        /// touch the cache at all -- otherwise a crashed hover-only server
        /// would wipe out a healthy, still-running diagnostics server's
        /// valid entries for the same files.
        #[tokio::test]
        async fn test_respawn_if_dead_does_not_clear_cache_when_not_diagnostics_route() {
            use crate::config::LspServerConfig;

            let dir = TempDir::new().unwrap();
            let seed_script = write_crash_after_init_script(dir.path());
            let hover_id = ServerId::from_static("hover-only");
            let hover_seed_config = stub_server_config("hover-only", &seed_script);

            let seed = LspServer::spawn(hover_seed_config).await.unwrap().server;

            // `hover_id` handles only Hover; a separate (never-registered
            // here, purely routing-table) server is the catch-all and thus
            // the diagnostics route.
            let configs = [
                LspServerConfig {
                    language_id: LanguageId::from_static("rust"),
                    command: ServerCommand::from_static("sh").into(),
                    args: crate::config::ServerArgs::default(),
                    env: crate::config::ServerEnv::default(),
                    file_patterns: vec![],
                    initialization_options: None,
                    settings: None,
                    timeout_seconds: TimeoutSecs::new(5).unwrap(),
                    request_timeout_seconds: TimeoutSecs::new(5).unwrap(),
                    heuristics: None,
                    name: Some(ServerId::from_static("hover-only")),
                    handles: Some(ToolSet::new(vec![ToolKind::Hover]).unwrap()),
                    indexing: crate::bridge::IndexingPolicy::Auto,
                },
                LspServerConfig {
                    language_id: LanguageId::from_static("rust"),
                    command: ServerCommand::from_static("sh").into(),
                    args: crate::config::ServerArgs::default(),
                    env: crate::config::ServerEnv::default(),
                    file_patterns: vec![],
                    initialization_options: None,
                    settings: None,
                    timeout_seconds: TimeoutSecs::new(5).unwrap(),
                    request_timeout_seconds: TimeoutSecs::new(5).unwrap(),
                    heuristics: None,
                    name: Some(ServerId::from_static("diag-catchall")),
                    handles: None,
                    indexing: crate::bridge::IndexingPolicy::Auto,
                },
            ];
            let router = ToolRouter::from_configs(configs.iter()).unwrap();

            let cache = Arc::new(Mutex::new(crate::bridge::NotificationCache::new()));
            let translator = Translator::new()
                .with_router(router)
                .with_notification_cache(Arc::clone(&cache));
            translator.register_client(hover_id.clone(), seed.client().clone());
            translator.register_server(hover_id.clone(), seed);

            let owned_by_healthy_server: lsp_types::Uri =
                lsp_types::Uri::from("file:///workspace/still_healthy.rs");
            cache
                .lock()
                .await
                .store_diagnostics(&hover_id, &owned_by_healthy_server, None, vec![]);

            wait_until_dead(&translator, &hover_id).await;

            let respawn_script = write_responder_script(dir.path(), 1);
            // `language_id` must match the router's ("rust"), not the
            // routing identity ("hover-only"): otherwise `is_diagnostics_route`
            // returns `false` because of a language mismatch rather than
            // because of the `handles: Some([Hover])` restriction this test
            // means to exercise, which would pass for the wrong reason.
            let mut respawn_config = stub_server_config("hover-only", &respawn_script);
            respawn_config.server_config_mut().language_id = LanguageId::from_static("rust");
            set_respawn_config(&translator, &hover_id, respawn_config);

            translator.respawn_if_dead(&hover_id).await.unwrap();

            assert!(
                cache
                    .lock()
                    .await
                    .diagnostics(&owned_by_healthy_server)
                    .is_some(),
                "respawning a non-diagnostics-route server must not clear \
                 the diagnostics-route server's cache entries"
            );
            assert!(
                !cache.lock().await.is_push_degraded(&hover_id),
                "#359: respawning a server that is not the diagnostics route \
                 must not mark it push-degraded either"
            );
        }

        /// #249 test-gap closure: proves `resolve_client_for_file`'s
        /// dead-server branch is actually reached through the shared
        /// entry point every public tool handler (`handle_hover`,
        /// `handle_definition`, ...) funnels through -- not just through
        /// the private `respawn_if_dead`/`dead_server_config` calls the other
        /// tests in this module make directly.
        #[tokio::test]
        async fn test_prepare_document_respawns_dead_server_through_shared_entry_point() {
            let dir = TempDir::new().unwrap();
            let workspace = dir.path();
            let file_path = workspace.join("main.rs");
            fs::write(&file_path, "fn main() {}").unwrap();

            let seed_script = write_crash_after_init_script(dir.path());
            let id = ServerId::from_static("rust");
            let seed_config = stub_server_config("rust", &seed_script);

            let seed = LspServer::spawn(seed_config).await.unwrap().server;
            let mut translator = Translator::new()
                .with_router(ToolRouter::catch_all([(
                    id.clone(),
                    LanguageId::from_static("rust"),
                )]))
                .with_extensions(HashMap::from([(
                    FileExtension::from_static("rs"),
                    LanguageId::from_static("rust"),
                )]));
            translator.set_workspace_roots(
                WorkspaceRoots::from_paths(&[workspace.to_path_buf()]).unwrap(),
            );
            translator.register_client(id.clone(), seed.client().clone());
            translator.register_server(id.clone(), seed);
            wait_until_dead(&translator, &id).await;

            let respawn_script = write_responder_script(dir.path(), 1);
            set_respawn_config(
                &translator,
                &id,
                stub_server_config("rust", &respawn_script),
            );

            let result = translator
                .prepare_document(
                    &client_path(file_path.to_string_lossy().into_owned()),
                    ToolKind::Hover,
                )
                .await;
            assert!(result.is_ok(), "got {result:?}");

            assert!(
                translator.dead_server_config(&id).is_none(),
                "the respawned replacement should be alive"
            );
        }

        /// A diagnostics-route translator whose seed server has crashed, with a
        /// cache holding one diagnostics entry for it.
        async fn crashed_diagnostics_translator(
            dir: &TempDir,
        ) -> (
            Translator,
            Arc<Mutex<crate::bridge::NotificationCache>>,
            ServerId,
        ) {
            let id = ServerId::from_static("rust");
            let seed_script = write_crash_after_init_script(dir.path());
            let seed = LspServer::spawn(stub_server_config("rust", &seed_script))
                .await
                .unwrap()
                .server;
            let cache = Arc::new(Mutex::new(crate::bridge::NotificationCache::new()));
            let translator = Translator::new()
                .with_router(ToolRouter::catch_all([(
                    id.clone(),
                    LanguageId::from_static("rust"),
                )]))
                .with_notification_cache(Arc::clone(&cache));
            translator.register_client(id.clone(), seed.client().clone());
            translator.register_server(id.clone(), seed);
            cache.lock().await.store_diagnostics(
                &id,
                &lsp_types::Uri::from("file:///workspace/opened.rs"),
                None,
                vec![],
            );
            wait_until_dead(&translator, &id).await;
            let respawn_script = write_responder_script(dir.path(), 5);
            set_respawn_config(
                &translator,
                &id,
                stub_server_config("rust", &respawn_script),
            );
            (translator, cache, id)
        }

        /// #687 S1: a respawn dropped before its commit leaves no trace: no
        /// success recorded, the dead server still registered, the cache
        /// untouched. A later respawn then works.
        #[tokio::test]
        async fn test_respawn_dropped_before_the_commit_changes_nothing() {
            let dir = TempDir::new().unwrap();
            let (translator, cache, id) = crashed_diagnostics_translator(&dir).await;
            let uri = lsp_types::Uri::from("file:///workspace/opened.rs");

            let cache_guard = cache.lock().await;
            let abandoned =
                tokio::time::timeout(Duration::from_millis(500), translator.respawn_if_dead(&id))
                    .await;
            assert!(abandoned.is_err(), "the respawn must wait for the cache");
            assert!(
                translator
                    .backoff_of(&id)
                    .is_some_and(|b| matches!(b.last, LastAttempt::Failed(_))),
                "an abandoned respawn counts as a failed attempt"
            );
            assert!(translator.dead_server_config(&id).is_some());
            assert!(cache_guard.diagnostics(&uri).is_some());
            drop(cache_guard);

            assert_matches!(
                translator.respawn_if_dead(&id).await,
                Err(Error::ServerUnavailable { .. }),
                "the retried call must back off"
            );
            let config = translator.dead_server_config(&id).unwrap();
            translator
                .respawn_locked(
                    &id,
                    RespawnConfig::Unvetted(config),
                    BackoffPolicy::Bypass,
                    NotificationRouting::Discard,
                )
                .await
                .unwrap();
            assert!(translator.dead_server_config(&id).is_none());
            assert!(cache.lock().await.diagnostics(&uri).is_none());
        }

        /// #687 S1: once the replacement is swapped in, the subscribers are
        /// still told even if the caller goes away while that happens.
        #[tokio::test]
        async fn test_respawn_dropped_after_the_commit_still_announces_it() {
            let dir = TempDir::new().unwrap();
            let (translator, _cache, id) = crashed_diagnostics_translator(&dir).await;
            let wiring = Arc::new(crate::test_lsp::GatedWiring::closed());
            translator.install_wiring(Arc::clone(&wiring) as Arc<dyn NotificationWiring>);

            let abandoned =
                tokio::time::timeout(Duration::from_secs(2), translator.respawn_if_dead(&id)).await;
            assert!(abandoned.is_err(), "the announcement is held at the gate");
            assert!(
                translator.dead_server_config(&id).is_none(),
                "the replacement was committed before the caller was dropped"
            );
            assert_eq!(wiring.published(), 0);

            wiring.release();
            tokio::time::timeout(Duration::from_secs(2), wiring.wait_published())
                .await
                .expect("the detached tail must still publish");
            assert_eq!(wiring.published(), 1);
        }
    }
}