mcpls-core 0.7.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};
use super::servers::Backend;
use crate::bridge::indexing::IndexingReset;
use crate::bridge::{DiagnosticsRole, on_lifecycle};
use crate::config::ServerId;
use crate::error::{Error, Result};
use crate::lsp::{LspClient, LspServer, ServerInitConfig};
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,
    /// When the most recent attempt was made, or (if `last_attempt_succeeded`)
    /// when that success was last found to have not held up.
    last_attempt: Instant,
    /// Whether the most recent attempt completed `initialize` successfully.
    /// `false` for an outright spawn failure. Also reset to `false` once a
    /// "successful" respawn is found to have died again within the
    /// stability window, so that discovery is applied only once.
    last_attempt_succeeded: bool,
}

/// 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.
    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_attempt)
        };
        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
    }

    /// Applies `update` to the backoff state of `id`, creating it with
    /// `initial` first; a server without a slot has no backoff to keep.
    fn update_backoff(
        &self,
        id: &ServerId,
        initial: RespawnBackoff,
        update: impl FnOnce(&mut RespawnBackoff),
    ) {
        if let Some(slot) = lock_std(&self.servers).get_mut(id) {
            update(slot.backoff.get_or_insert(initial));
        }
    }

    /// Records a failed respawn attempt for `id`, extending its backoff.
    fn record_respawn_failure(&self, id: &ServerId) {
        let now = self.clock.now();
        self.update_backoff(
            id,
            RespawnBackoff {
                consecutive_failures: 0,
                last_attempt: now,
                last_attempt_succeeded: false,
            },
            |entry| {
                entry.consecutive_failures = entry.consecutive_failures.saturating_add(1);
                entry.last_attempt = now;
                entry.last_attempt_succeeded = false;
            },
        );
    }

    /// 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) {
        let now = self.clock.now();
        self.update_backoff(
            id,
            RespawnBackoff {
                consecutive_failures: 0,
                last_attempt: now,
                last_attempt_succeeded: true,
            },
            |entry| {
                entry.last_attempt = now;
                entry.last_attempt_succeeded = true;
            },
        );
    }

    /// 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 Some(entry) = self.backoff_of(id) else {
            return;
        };
        if !entry.last_attempt_succeeded {
            return;
        }
        if self
            .clock
            .now()
            .saturating_duration_since(entry.last_attempt)
            >= RESPAWN_BACKOFF_BASE
        {
            if let Some(slot) = lock_std(&self.servers).get_mut(id) {
                slot.backoff = None;
            }
        } else if let Some(slot) = lock_std(&self.servers).get_mut(id)
            && let Some(current) = slot.backoff.as_mut()
        {
            current.consecutive_failures = current.consecutive_failures.saturating_add(1);
            current.last_attempt = self.clock.now();
            current.last_attempt_succeeded = false;
        }
    }

    /// 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().await;
        }

        self.reconcile_respawn_stability(id);

        tracing::warn!("LSP server '{id}' has crashed, respawning");
        self.respawn_locked(
            id,
            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, recorded in the backoff
    ///    bookkeeping.
    /// 2. The previous notification task for `id` is aborted. Abort is
    ///    asynchronous: it may still finish one cache write already in
    ///    progress.
    /// 3. `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`. This precedes step 4 so a readiness
    ///    signal or diagnostics push the replacement emitted during its own
    ///    handshake, already buffered in its channels, is the last write.
    /// 4. The consumer for the replacement's notification lanes is started.
    /// 5. The replacement replaces the registered client and server,
    ///    `document_tracker` forgets `id` (after the swap: a caller still on
    ///    the old client must not record sync against the new generation),
    ///    the old client's pending requests fail, and subscribers of the
    ///    cleared diagnostics are told to re-read.
    ///
    /// # Errors
    ///
    /// [`Error::ServerUnavailable`] when backoff is honoured and active, or
    /// the spawn error of the replacement.
    pub(super) async fn respawn_locked(
        &self,
        id: &ServerId,
        config: ServerInitConfig,
        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 config = match config.for_respawn().await {
            Ok(config) => config,
            Err(err) => {
                self.record_respawn_failure(id);
                return Err(err);
            }
        };
        let language_id = config.server_config().language_id.clone();
        let pinned_tsserver = config.pinned_tsserver();

        let mut new_server = match LspServer::spawn(config).await {
            Ok(server) => {
                self.record_respawn_success(id);
                server
            }
            Err(err) => {
                self.record_respawn_failure(id);
                return Err(err);
            }
        };
        let receivers = NotificationReceivers {
            notifications: new_server.take_notification_inbox(),
            lifecycle: new_server.take_lifecycle_rx(),
            pinned_tsserver,
        };

        let stale_task = lock_std(&self.servers)
            .get_mut(id)
            .and_then(|slot| slot.notification_task.take());
        if let Some(handle) = stale_task {
            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 diagnostics_route = self.is_diagnostics_route(&language_id, id);
        let mut cleared = Vec::new();
        if let Some(cache) = &self.notification_cache {
            let mut cache = cache.lock().await;
            cache.reset_indexing_state(
                id,
                IndexingReset::AwaitReplacement {
                    within: self.indexing_ready_timeout,
                },
            );
            if diagnostics_route {
                cleared = cache.clear_server_diagnostics(id);
                match routing {
                    NotificationRouting::Pump(_) => cache.clear_push_degraded(id),
                    NotificationRouting::Discard => {
                        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"
                        );
                        cache.mark_push_degraded(id);
                    }
                }
            }
        }

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

        self.document_tracker.forget_server(id);

        if let Some(old_client) = old_client {
            old_client.fail_pending_requests().await;
        }

        if let Some(wiring) = self.wiring.get() {
            wiring.publish_invalidated(&cleared).await;
        }

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

    /// 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()
    }
}

/// 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_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::WorkspaceRoots;
        use crate::bridge::translator::testing::{
            pid_is_running, stub_server_config, write_crash_after_init_script,
            write_responder_script,
        };
        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();
            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();
            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();
            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();
            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();
            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();
            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();
            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();
            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();
            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();

            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();

            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();

            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();

            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();
            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();

            // `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: vec![],
                    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: vec![],
                    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();
            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"
            );
        }
    }
}