mcpls-core 0.7.0

//! Diagnostics pump: drains one LSP server's notification lanes into the
//! shared cache and fans publications out to subscribed MCP sessions.

use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::sync::{Arc, Mutex as StdMutex};

use tokio::sync::{Mutex, mpsc, watch};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error};

use crate::bridge::resources::{DiagnosticsResourceUri, PublishedDiagnosticsUri};
use crate::bridge::{
    self, DiagnosticsRole, IndexingReset, NotificationCache, Publication, PublicationKind,
    PublishedPathResolver, PushWrite, WorkspaceRoots,
};
use crate::config::ServerId;
use crate::lsp::{
    BoundedPublish, LostFiles, LspNotification, NotificationInbox, PublishDelivery, PublishReader,
    ServerMessage,
};
use crate::mcp::SubscriptionRegistry;
use crate::util::{catch_panic, lock_std};

/// `Arc`-backed state shared by every `diagnostics_pump` task spawned for one
/// `serve_with` run, factored out of `diagnostics_pump`'s parameter list to
/// keep it under clippy's argument-count lint. `Clone` is cheap (`Arc`
/// clones only).
#[derive(Clone)]
pub struct PumpShared {
    pub(crate) notification_cache: Arc<Mutex<NotificationCache>>,
    /// Every session that has subscribed to anything (one per HTTP session,
    /// or the single stdio session); the pump hands each one the URI and the
    /// session decides whether it is subscribed -- see [`SubscriptionRegistry`].
    pub(crate) subs: SubscriptionRegistry,
    /// Used to reject diagnostics for out-of-workspace URIs (see #234): a
    /// misbehaving server could otherwise flush the FIFO-capped cache with
    /// fabricated URIs.
    pub(crate) workspace_roots: WorkspaceRoots,
    /// The role of every server's pump, so a recomputed role reaches the pump
    /// that is live, including one started for a respawned server.
    pub(crate) roles: DiagnosticsRoles,
}

/// The role of every server's pump, one `watch` sender per server.
///
/// The sender outlives any one pump: a pump started for a respawned or
/// restarted server subscribes to the same sender, so a role recomputed while
/// servers settle reaches whichever pump is live.
#[derive(Clone, Default)]
pub struct DiagnosticsRoles(Arc<StdMutex<HashMap<ServerId, watch::Sender<DiagnosticsRole>>>>);

impl DiagnosticsRoles {
    /// A receiver of `id`'s role. The sender keeps the role it has: a pump that
    /// starts later follows the role the startup recomputation maintains, and
    /// `role` only seeds a server that has none yet.
    pub(crate) fn subscribe(
        &self,
        id: &ServerId,
        role: DiagnosticsRole,
    ) -> watch::Receiver<DiagnosticsRole> {
        let mut senders = lock_std(&self.0);
        let sender = senders
            .entry(id.clone())
            .or_insert_with(|| watch::channel(role).0);
        let receiver = sender.subscribe();
        drop(senders);
        receiver
    }

    /// The current role of `id`, if a pump was ever started for it.
    #[cfg(test)]
    pub(crate) fn get(&self, id: &ServerId) -> Option<DiagnosticsRole> {
        lock_std(&self.0).get(id).map(|sender| *sender.borrow())
    }

    /// Sets `id`'s role if a pump was ever started for it.
    pub(crate) fn set(&self, id: &ServerId, role: DiagnosticsRole) {
        if let Some(sender) = lock_std(&self.0).get(id) {
            sender.send_if_modified(|current| {
                let changed = *current != role;
                *current = role;
                changed
            });
        }
    }
}

impl std::fmt::Debug for DiagnosticsRoles {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DiagnosticsRoles").finish_non_exhaustive()
    }
}

/// What a pump still reads; a source is `None` once it has closed.
struct PumpInputs {
    messages: Option<mpsc::Receiver<ServerMessage>>,
    publishes: Option<PublishReader>,
    lifecycle: Option<mpsc::Receiver<LspNotification>>,
}

impl PumpInputs {
    const fn all_closed(&self) -> bool {
        self.messages.is_none() && self.publishes.is_none() && self.lifecycle.is_none()
    }
}

/// Next frame of `source`; `None` when it just closed (and is dropped), and
/// pending forever once it was.
async fn recv_open<T>(source: &mut Option<mpsc::Receiver<T>>) -> Option<T> {
    let Some(receiver) = source else {
        return std::future::pending().await;
    };
    let frame = receiver.recv().await;
    if frame.is_none() {
        *source = None;
    }
    frame
}

/// Next delivery of the mailbox in `source`, with [`recv_open`]'s closing rule.
async fn recv_publish(source: &mut Option<PublishReader>) -> Option<PublishDelivery> {
    let Some(reader) = source else {
        return std::future::pending().await;
    };
    let delivery = reader.recv().await;
    if delivery.is_none() {
        *source = None;
    }
    delivery
}

/// Whether a pump keeps going.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Flow {
    Continue,
    Stop,
}

/// A drained publication and whether its path is resolved first.
enum BatchItem {
    /// An admitted publication: its path is resolved before it is applied.
    Resolve(BoundedPublish),
    /// Anything else: applied as it is.
    Skip(BoundedPublish),
}

impl BatchItem {
    const fn publish(&self) -> &BoundedPublish {
        match self {
            Self::Resolve(publish) | Self::Skip(publish) => publish,
        }
    }
}

/// Everything one server's pump needs besides what it reads.
struct Pump {
    server_id: ServerId,
    cancel: CancellationToken,
    role_rx: watch::Receiver<DiagnosticsRole>,
    pinned_tsserver: Option<PathBuf>,
    notification_cache: Arc<Mutex<NotificationCache>>,
    subs: SubscriptionRegistry,
    workspace_roots: WorkspaceRoots,
    resolver: PublishedPathResolver,
}

/// Background task that drains LSP notifications, writes them to the cache,
/// and queues `resources/updated` for each MCP session subscribed to the URI.
///
/// Reads three independent sources (P3) rather than one: the inbox's
/// `messages` channel carries log/showMessage frames, its mailbox carries
/// `publishDiagnostics` (coalesced per file; a publish the mailbox could not
/// hold arrives as a list of lost files that the cache then marks), and
/// `lifecycle_rx` carries `$/progress` `begin`/`end` frames and `Other` (which
/// carries e.g. rust-analyzer's `experimental/serverStatus`). Splitting them
/// means a high-volume diagnostics publisher (rust-analyzer republishing
/// whole-workspace diagnostics on every save) can never starve out a
/// low-volume readiness signal, or vice versa -- see
/// `lsp::client::LspClient::message_loop_inner` for where each notification is
/// classified.
///
/// Delivery is per session and never awaits a peer: the pump only calls
/// `SessionState::publish_if_subscribed` on each registered session, whose own
/// task notifies that session's peer. A session that has not subscribed yet is
/// not registered, so notifications arriving before the first subscribe are
/// only cached.
///
/// The task exits when:
/// - **All** sources have closed -- in practice they close together with the
///   `LspClient`, but each is tracked independently so one closing early can
///   never stop the others from still being drained.
/// - The cancellation token is cancelled.
///
/// # Lock independence
/// Cache writes acquire only `Arc<Mutex<NotificationCache>>`, a lock entirely
/// separate from `translator`'s own internal locks (`Arc<Translator>` has no
/// outer mutex; each field manages its own short-lived, independent lock).
/// Neither an in-flight LSP round-trip (e.g. `textDocument/diagnostic`) nor
/// any other translator-side work holds the notification-cache lock, so this
/// pump is never blocked by tool-call activity: a `publishDiagnostics`
/// notification arriving mid-request waits in the mailbox and is cached as
/// soon as the pump runs, instead of being lost.
pub async fn diagnostics_pump(
    server_id: ServerId,
    inbox: impl Into<NotificationInbox>,
    lifecycle_rx: mpsc::Receiver<LspNotification>,
    cancel: CancellationToken,
    role_rx: watch::Receiver<DiagnosticsRole>,
    pinned_tsserver: Option<PathBuf>,
    shared: PumpShared,
) {
    diagnostics_pump_with_resolver(
        server_id,
        inbox,
        lifecycle_rx,
        cancel,
        role_rx,
        pinned_tsserver,
        shared,
        PublishedPathResolver::new(),
    )
    .await;
}

/// [`diagnostics_pump`] over a caller-supplied path resolver, so tests can
/// inject a slow or hanging canonicalizer.
#[expect(
    clippy::too_many_arguments,
    reason = "internal test seam; each parameter is a distinct per-server input"
)]
async fn diagnostics_pump_with_resolver(
    server_id: ServerId,
    inbox: impl Into<NotificationInbox>,
    lifecycle_rx: mpsc::Receiver<LspNotification>,
    cancel: CancellationToken,
    role_rx: watch::Receiver<DiagnosticsRole>,
    pinned_tsserver: Option<PathBuf>,
    shared: PumpShared,
    resolver: PublishedPathResolver,
) {
    let NotificationInbox {
        messages,
        publishes,
    } = inbox.into();
    let PumpShared {
        notification_cache,
        subs,
        workspace_roots,
        roles: _,
    } = shared;
    let mut pump = Pump {
        server_id,
        cancel,
        role_rx,
        pinned_tsserver,
        notification_cache,
        subs,
        workspace_roots,
        resolver,
    };
    pump.run(PumpInputs {
        messages: Some(messages),
        publishes: Some(publishes),
        lifecycle: Some(lifecycle_rx),
    })
    .await;
}

impl Pump {
    async fn run(&mut self, mut inputs: PumpInputs) {
        while !inputs.all_closed() {
            let flow = tokio::select! {
                () = self.cancel.cancelled() => Flow::Stop,
                frame = recv_open(&mut inputs.messages) => {
                    if let Some(message) = frame {
                        self.on_message(message).await;
                    }
                    Flow::Continue
                }
                delivery = recv_publish(&mut inputs.publishes) => match delivery {
                    Some(delivery) => self.apply_delivery(delivery, &mut inputs).await,
                    None => Flow::Continue,
                },
                frame = recv_open(&mut inputs.lifecycle) => {
                    if let Some(notif) = frame {
                        self.on_lifecycle(notif).await;
                    }
                    Flow::Continue
                }
            };
            if flow == Flow::Stop {
                return;
            }
        }
    }

    async fn on_lifecycle(&self, notif: LspNotification) {
        bridge::on_lifecycle(
            Some(&self.notification_cache),
            &self.server_id,
            self.pinned_tsserver.as_deref(),
            notif,
        )
        .await;
    }

    async fn on_message(&self, message: ServerMessage) {
        let mut cache = self.notification_cache.lock().await;
        match message {
            ServerMessage::Log(m) => cache.store_log(m.kind.into(), m.message),
            ServerMessage::Show(m) => cache.store_message(m.kind.into(), m.message),
        }
    }

    /// Applies one mailbox delivery: a publish is batched with the publishes
    /// that follow it, files the mailbox lost are marked in the cache.
    async fn apply_delivery(&mut self, delivery: PublishDelivery, inputs: &mut PumpInputs) -> Flow {
        let mut next = Some(delivery);
        while let Some(delivery) = next.take() {
            let flow = match delivery {
                PublishDelivery::Publish(first) => {
                    let mut batch = vec![first];
                    if let Some(reader) = &mut inputs.publishes {
                        while batch.len() < PUMP_BATCH {
                            match reader.try_recv() {
                                Some(PublishDelivery::Publish(more)) => batch.push(more),
                                other => {
                                    next = other;
                                    break;
                                }
                            }
                        }
                    }
                    self.apply_batch(batch, &mut inputs.lifecycle).await
                }
                PublishDelivery::Lost(files) => {
                    self.apply_lost(&files, &mut inputs.lifecycle).await
                }
                PublishDelivery::LostUnnamed => self.apply_unnamed_loss().await,
            };
            if flow == Flow::Stop {
                return Flow::Stop;
            }
        }
        Flow::Continue
    }

    /// Resolves `publications` while the lifecycle lane keeps being serviced;
    /// `None` when the pump was cancelled meanwhile.
    async fn resolve(
        &mut self,
        publications: &[Publication<'_>],
        lifecycle: &mut Option<mpsc::Receiver<LspNotification>>,
    ) -> Option<Vec<Option<PublishedDiagnosticsUri>>> {
        let mut resolving = std::pin::pin!(
            self.resolver
                .resolve_batch(publications, &self.workspace_roots)
        );
        loop {
            tokio::select! {
                () = self.cancel.cancelled() => return None,
                done = &mut resolving => return Some(done),
                frame = recv_open(lifecycle) => {
                    if let Some(notif) = frame {
                        bridge::on_lifecycle(
                            Some(&self.notification_cache),
                            &self.server_id,
                            self.pinned_tsserver.as_deref(),
                            notif,
                        )
                        .await;
                    }
                }
            }
        }
    }

    /// A cold burst resolves its paths in parallel but is applied strictly in
    /// arrival order, so a clear is never reordered.
    async fn apply_batch(
        &mut self,
        batch: Vec<BoundedPublish>,
        lifecycle: &mut Option<mpsc::Receiver<LspNotification>>,
    ) -> Flow {
        let admission_role = *self.role_rx.borrow();
        let items: Vec<BatchItem> = batch
            .into_iter()
            .map(|publish| {
                if publication_admitted(publish.uri(), admission_role, &self.workspace_roots) {
                    BatchItem::Resolve(publish)
                } else {
                    BatchItem::Skip(publish)
                }
            })
            .collect();
        let publications: Vec<Publication<'_>> = items
            .iter()
            .filter(|item| matches!(item, BatchItem::Resolve(_)))
            .map(BatchItem::publish)
            .map(|publish| Publication {
                uri: publish.uri(),
                kind: if publish.diagnostics().as_slice().is_empty() {
                    PublicationKind::Clear
                } else {
                    PublicationKind::Diagnostics
                },
            })
            .collect();
        let Some(resolved) = self.resolve(&publications, lifecycle).await else {
            return Flow::Stop;
        };
        drop(publications);
        let mut resolved = resolved.into_iter();

        for item in items {
            let (publish, published) = match item {
                BatchItem::Resolve(publish) => (publish, resolved.next().flatten()),
                BatchItem::Skip(publish) => (publish, None),
            };
            // Re-read after the (possibly slow) resolve: a demotion in the
            // meantime must stop caching and fan-out (#174 s8).
            let role = *self.role_rx.borrow();
            apply_publish(
                &self.server_id,
                publish,
                published,
                role,
                &self.notification_cache,
                &self.subs,
            )
            .await;
        }
        Flow::Continue
    }

    /// Marks the files whose publishes the mailbox could not hold, so they
    /// read as lost instead of `pending`, and tells their subscribers.
    async fn apply_lost(
        &mut self,
        lost: &LostFiles,
        lifecycle: &mut Option<mpsc::Receiver<LspNotification>>,
    ) -> Flow {
        if *self.role_rx.borrow() == DiagnosticsRole::Secondary {
            return Flow::Continue;
        }
        let uris: Vec<&lsp_types::Uri> = lost
            .uris()
            .iter()
            .filter(|uri| self.workspace_roots.admits_uri(uri))
            .collect();
        let publications: Vec<Publication<'_>> = uris
            .iter()
            .map(|uri| Publication {
                uri,
                kind: PublicationKind::Lost,
            })
            .collect();
        let Some(resolved) = self.resolve(&publications, lifecycle).await else {
            return Flow::Stop;
        };
        drop(publications);
        let files: Vec<bridge::DiagnosticsKey> = resolved
            .into_iter()
            .flatten()
            .map(|published| bridge::DiagnosticsKey::of(published.canonical()))
            .collect();
        let changed = self
            .notification_cache
            .lock()
            .await
            .record_lost_files(&self.server_id, &files);
        publish_invalidated(&self.subs, &changed).await;
        Flow::Continue
    }

    /// Marks that the mailbox lost more publishes than it could name.
    async fn apply_unnamed_loss(&self) -> Flow {
        if *self.role_rx.borrow() != DiagnosticsRole::Secondary {
            self.notification_cache
                .lock()
                .await
                .record_unnamed_loss(&self.server_id);
        }
        Flow::Continue
    }
}

/// Notifications drained per resolve round.
const PUMP_BATCH: usize = 64;

/// Whether a publication for `uri` is one this pump should canonicalize and
/// cache.
fn publication_admitted(
    uri: &lsp_types::Uri,
    role: DiagnosticsRole,
    workspace_roots: &WorkspaceRoots,
) -> bool {
    role == DiagnosticsRole::Authoritative && workspace_roots.admits_uri(uri)
}

/// Applies one drained publication; `published` is the resolution of one that
/// passed [`publication_admitted`], `None` when it resolved outside the
/// workspace or could not be resolved.
async fn apply_publish(
    server_id: &ServerId,
    publish: BoundedPublish,
    published: Option<PublishedDiagnosticsUri>,
    role: DiagnosticsRole,
    notification_cache: &Mutex<NotificationCache>,
    subs: &SubscriptionRegistry,
) {
    // Only the server the router resolves `Diagnostics` to for this
    // notification's language caches (and notifies subscribers of) it
    // -- see #174 §8.
    if role == DiagnosticsRole::Secondary {
        return;
    }
    let (uri, version, diagnostics) = publish.into_parts();
    let Some(published) = published else {
        debug!(
            "dropping diagnostics for URI outside the workspace or unresolvable: {}",
            uri.as_ref()
        );
        return;
    };
    let write = notification_cache.lock().await.write_published_diagnostics(
        server_id,
        &published,
        version.map(crate::bridge::DocumentVersion::new),
        diagnostics,
    );

    if matches!(write, PushWrite::Stored { .. }) {
        publish_to_subscribers(subs, || DiagnosticsResourceUri::for_published(&published)).await;
    }
    publish_invalidated(subs, write.evicted()).await;
}

/// Queues the resource `make_uri` builds on every session subscribed to it.
///
/// `make_uri` is only called when some session has a subscription at all, so
/// the common case of nobody subscribed costs no URI construction.
async fn publish_to_subscribers(
    subs: &SubscriptionRegistry,
    make_uri: impl FnOnce() -> Option<DiagnosticsResourceUri>,
) {
    if !subs.any_subscription().await {
        return;
    }

    let Some(mcp_uri) = make_uri() else {
        return;
    };
    for session in &subs.live_sessions() {
        session.publish_if_subscribed(&mcp_uri).await;
    }
}

/// Tells every session subscribed to a file in `files` that its diagnostics
/// changed: cleared by a respawn, or evicted from the cache.
async fn publish_invalidated(subs: &SubscriptionRegistry, files: &[bridge::DiagnosticsKey]) {
    if files.is_empty() {
        return;
    }
    let files: HashSet<&bridge::DiagnosticsKey> = files.iter().collect();
    subs.publish_matching(|uri| {
        bridge::diagnostics_cache_key(uri).is_some_and(|key| files.contains(&key))
    })
    .await;
}

/// Stops trusting a panicked pump's server: marks it push-degraded and resets
/// its indexing state, so callers poll instead of waiting on pushes that no
/// longer arrive.
async fn degrade_after_pump_panic(cache: &Mutex<NotificationCache>, id: &ServerId) {
    let mut cache = cache.lock().await;
    cache.mark_push_degraded(id);
    cache.reset_indexing_state(id, IndexingReset::Forget);
}

/// [`diagnostics_pump`] with its panic contained: a panic is logged and the
/// server degraded ([`degrade_after_pump_panic`]) instead of unwinding the
/// task.
///
/// The one way a pump is started, for the initial servers and for restarts.
pub async fn supervised_pump(
    server_id: ServerId,
    receivers: bridge::NotificationReceivers,
    cancel: CancellationToken,
    role_rx: tokio::sync::watch::Receiver<DiagnosticsRole>,
    shared: PumpShared,
) {
    let cache = Arc::clone(&shared.notification_cache);
    let pump = diagnostics_pump(
        server_id.clone(),
        receivers.notifications,
        receivers.lifecycle,
        cancel,
        role_rx,
        receivers.pinned_tsserver,
        shared,
    );
    contain_pump_panic(pump, &server_id, &cache).await;
}

/// Awaits `pump`, degrading `server_id` when it panics.
async fn contain_pump_panic(
    pump: impl Future<Output = ()>,
    server_id: &ServerId,
    cache: &Mutex<NotificationCache>,
) {
    if let Err(panicked) = catch_panic(pump).await {
        error!(
            "Diagnostics pump for LSP server '{server_id}' panicked: {}",
            panicked.message()
        );
        degrade_after_pump_panic(cache, server_id).await;
    }
}

/// Re-starts diagnostics pumps for manually restarted servers over the same
/// shared state and shutdown token the initial pumps use.
#[derive(Clone)]
pub struct PumpWiring {
    shared: PumpShared,
    cancel: CancellationToken,
}

impl std::fmt::Debug for PumpWiring {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PumpWiring").finish_non_exhaustive()
    }
}

impl PumpWiring {
    /// Wires restarted pumps to `shared` and to the shutdown token `cancel`.
    pub(crate) const fn new(shared: PumpShared, cancel: CancellationToken) -> Self {
        Self { shared, cancel }
    }
}

impl bridge::NotificationWiring for PumpWiring {
    fn spawn_pump(
        &self,
        id: ServerId,
        receivers: bridge::NotificationReceivers,
        role: DiagnosticsRole,
    ) -> tokio::task::AbortHandle {
        let role_rx = self.shared.roles.subscribe(&id, role);
        tokio::spawn(supervised_pump(
            id,
            receivers,
            self.cancel.clone(),
            role_rx,
            self.shared.clone(),
        ))
        .abort_handle()
    }

    fn publish_invalidated<'a>(
        &'a self,
        cleared: &'a [bridge::DiagnosticsKey],
    ) -> futures::future::BoxFuture<'a, ()> {
        Box::pin(publish_invalidated(&self.shared.subs, cleared))
    }

    fn has_subscriptions(&self) -> futures::future::BoxFuture<'_, bool> {
        Box::pin(self.shared.subs.any_subscription())
    }

    fn publish_changed<'a>(
        &'a self,
        file: &'a lsp_types::Uri,
    ) -> futures::future::BoxFuture<'a, ()> {
        Box::pin(publish_to_subscribers(&self.shared.subs, || {
            DiagnosticsResourceUri::for_canonical_file(file)
        }))
    }
}

#[cfg(test)]
mod pump_tests {
    use std::assert_matches;
    use std::time::Duration;

    use lsp_types::{PublishDiagnosticsParams, Uri};
    use tokio::sync::{mpsc, watch};

    use super::*;
    use crate::bridge::{IndexingState, Translator};
    use crate::config::LanguageId;
    use crate::runtime::startup::publish_startup_failures;
    use crate::{Error, ProjectConfigStatus, config, mcp};

    fn make_cache() -> Arc<Mutex<NotificationCache>> {
        Arc::new(Mutex::new(NotificationCache::new()))
    }

    fn make_subs() -> SubscriptionRegistry {
        SubscriptionRegistry::new()
    }

    /// A single real workspace root shared by the pump-mechanics tests
    /// below, cfg-gated because `Url::to_file_path` requires a drive
    /// letter on Windows -- mirrors
    /// `test_pump_drops_diagnostics_outside_workspace_roots`.
    #[cfg(windows)]
    fn test_workspace_roots() -> WorkspaceRoots {
        WorkspaceRoots::for_test(vec![PathBuf::from(r"C:\test")], vec![])
    }
    #[cfg(not(windows))]
    fn test_workspace_roots() -> WorkspaceRoots {
        WorkspaceRoots::for_test(vec![PathBuf::from("/test")], vec![])
    }

    /// A `file://` URI for `file` beneath [`test_workspace_roots`]'s root.
    #[cfg(windows)]
    fn test_uri(file: &str) -> Uri {
        Uri::from(format!("file:///C:/test/{file}").as_str())
    }
    #[cfg(not(windows))]
    fn test_uri(file: &str) -> Uri {
        Uri::from(format!("file:///test/{file}").as_str())
    }

    /// #566: a pinned server that reports another tsserver source is
    /// warned about; a report of the pinned source is not.
    #[tokio::test]
    async fn test_pump_warns_only_when_pinned_tsserver_is_ignored() {
        use tracing_subscriber::layer::SubscriberExt as _;

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

        let captured = CapturedLogs::default();
        let _guard =
            tracing::subscriber::set_default(tracing_subscriber::registry().with(captured.clone()));
        let (lifecycle_tx, _cancel_tx) =
            spawn_test_pump_with_tsserver_pin(PathBuf::from("/pin/tsserver.js"));
        let report = |source: &str| LspNotification::Other {
            method: "$/typescriptVersion".into(),
            params: Some(serde_json::json!({"version": "5.0", "source": source})),
        };
        let ignored = || {
            captured
                .messages()
                .iter()
                .filter(|m| m.contains("ignored the configured tsserver.path"))
                .count()
        };

        lifecycle_tx.send(report("user-setting")).await.unwrap();
        lifecycle_tx.send(report("workspace")).await.unwrap();
        tokio::time::timeout(std::time::Duration::from_secs(2), async {
            while ignored() == 0 {
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("mismatching source was never warned about");

        assert_eq!(ignored(), 1, "{:?}", captured.messages());
    }

    /// `PublishDiagnostics` is cached even when the peer is not yet connected.
    #[tokio::test]
    async fn test_pump_caches_before_peer_set() {
        let cache = make_cache();
        let subs = make_subs();
        let (tx, rx) = pump_feed();
        let (_lifecycle_tx, lifecycle_rx) = mpsc::channel(8);
        let cancel = CancellationToken::new();

        let c = Arc::clone(&cache);
        tokio::spawn(diagnostics_pump(
            ServerId::from_static("rust"),
            rx,
            lifecycle_rx,
            cancel,
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: c,
                subs: subs.clone(),
                workspace_roots: test_workspace_roots(),
            },
        ));

        let uri: Uri = test_uri("main.rs");
        tx.send(LspNotification::PublishDiagnostics(
            PublishDiagnosticsParams {
                uri: uri.clone(),
                diagnostics: vec![],
                version: None,
            },
        ))
        .await
        .unwrap();
        drop(tx);

        // Poll until the pump processes the message or we time out.
        let cached = tokio::time::timeout(std::time::Duration::from_secs(5), async {
            loop {
                tokio::task::yield_now().await;
                let found = {
                    let guard = cache.lock().await;
                    guard.diagnostics(&uri).is_some()
                };
                if found {
                    return true;
                }
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("pump did not cache diagnostics within 5 s");
        assert!(cached, "diagnostics should be cached before peer is set");
    }

    /// #234 (S1 hardening): diagnostics for URIs outside the configured
    /// workspace roots must be dropped rather than cached, closing the
    /// vector where a misbehaving server floods the FIFO-bounded cache
    /// with fabricated URIs to evict every legitimate entry.
    #[tokio::test]
    async fn test_pump_drops_diagnostics_outside_workspace_roots() {
        let cache = make_cache();
        let subs = make_subs();
        let (tx, rx) = pump_feed();
        let (_lifecycle_tx, lifecycle_rx) = mpsc::channel(8);
        let cancel = CancellationToken::new();

        // See `test_admits_uri_under_root_and_alias`
        // for why Windows needs a drive-letter path here.
        #[cfg(windows)]
        let (workspace_root, outside_uri_str, inside_uri_str) = (
            PathBuf::from(r"C:\workspace"),
            "file:///C:/etc/passwd",
            "file:///C:/workspace/src/main.rs",
        );
        #[cfg(not(windows))]
        let (workspace_root, outside_uri_str, inside_uri_str) = (
            PathBuf::from("/workspace"),
            "file:///etc/passwd",
            "file:///workspace/src/main.rs",
        );
        let workspace_roots = WorkspaceRoots::for_test(vec![workspace_root], vec![]);

        tokio::spawn(diagnostics_pump(
            ServerId::from_static("rust"),
            rx,
            lifecycle_rx,
            cancel,
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: Arc::clone(&cache),
                subs: subs.clone(),
                workspace_roots,
            },
        ));

        let outside_uri: Uri = Uri::from(outside_uri_str);
        let inside_uri: Uri = Uri::from(inside_uri_str);

        tx.send(LspNotification::PublishDiagnostics(
            PublishDiagnosticsParams {
                uri: outside_uri.clone(),
                diagnostics: vec![],
                version: None,
            },
        ))
        .await
        .unwrap();
        tx.send(LspNotification::PublishDiagnostics(
            PublishDiagnosticsParams {
                uri: inside_uri.clone(),
                diagnostics: vec![],
                version: None,
            },
        ))
        .await
        .unwrap();
        drop(tx);

        // Poll until the (later-sent) in-workspace sentinel is cached --
        // proves the pump already processed the earlier out-of-workspace
        // message too, since the channel preserves send order.
        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            loop {
                {
                    let guard = cache.lock().await;
                    if guard.diagnostics(&inside_uri).is_some() {
                        return;
                    }
                }
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("pump did not cache in-workspace diagnostics within 5 s");

        let found_outside = cache.lock().await.diagnostics(&outside_uri).is_some();
        assert!(
            !found_outside,
            "diagnostics for a URI outside workspace roots must not be cached"
        );
    }

    /// Pump exits cleanly when the cancel token is cancelled.
    #[tokio::test]
    async fn test_pump_exits_on_cancel() {
        let cache = make_cache();
        let subs = make_subs();
        let (_tx, rx) = pump_feed();
        let (_lifecycle_tx, lifecycle_rx) = mpsc::channel::<LspNotification>(8);
        let cancel = CancellationToken::new();

        let handle = tokio::spawn(diagnostics_pump(
            ServerId::from_static("rust"),
            rx,
            lifecycle_rx,
            cancel.clone(),
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: cache,
                subs,
                workspace_roots: test_workspace_roots(),
            },
        ));

        cancel.cancel();
        // Pump must finish within a short time after cancellation.
        tokio::time::timeout(std::time::Duration::from_millis(200), handle)
            .await
            .expect("pump did not exit within timeout")
            .unwrap();
    }

    /// Dropping the guard of the shutdown token cancels the pump.
    #[tokio::test]
    async fn test_pump_exits_when_cancel_guard_dropped() {
        let cache = make_cache();
        let subs = make_subs();
        let (_tx, rx) = pump_feed();
        let (_lifecycle_tx, lifecycle_rx) = mpsc::channel::<LspNotification>(8);
        let cancel = CancellationToken::new();
        let guard = cancel.clone().drop_guard();

        let handle = tokio::spawn(diagnostics_pump(
            ServerId::from_static("rust"),
            rx,
            lifecycle_rx,
            cancel,
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: cache,
                subs,
                workspace_roots: test_workspace_roots(),
            },
        ));

        drop(guard);
        tokio::time::timeout(std::time::Duration::from_millis(200), handle)
            .await
            .expect("pump did not exit within timeout")
            .unwrap();
    }

    /// Dropping a clone of the token is not a cancellation: the pump keeps running.
    #[tokio::test]
    async fn test_pump_survives_dropped_cancel_clone() {
        let cache = make_cache();
        let subs = make_subs();
        let (_tx, rx) = pump_feed();
        let (_lifecycle_tx, lifecycle_rx) = mpsc::channel::<LspNotification>(8);
        let cancel = CancellationToken::new();

        let handle = tokio::spawn(diagnostics_pump(
            ServerId::from_static("rust"),
            rx,
            lifecycle_rx,
            cancel.clone(),
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: cache,
                subs,
                workspace_roots: test_workspace_roots(),
            },
        ));

        drop(cancel);
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        assert!(
            !handle.is_finished(),
            "dropping a token clone must not stop the pump"
        );
        handle.abort();
    }

    /// Regression test for #104: the pump must cache a notification promptly
    /// even while another task holds the translator lock for far longer than
    /// any acceptable pump latency. Before the `NotificationCache` split, the
    /// pump locked `Arc<Mutex<Translator>>` to cache diagnostics, so it would
    /// have stalled here until the holder released the lock.
    #[tokio::test]
    async fn test_pump_makes_progress_while_translator_lock_held() {
        let translator = Arc::new(Mutex::new(Translator::new()));
        let cache = make_cache();
        let subs = make_subs();
        let (tx, rx) = pump_feed();
        let (_lifecycle_tx, lifecycle_rx) = mpsc::channel(8);
        let cancel = CancellationToken::new();

        // Simulate a slow in-flight MCP request (e.g. `pull_diagnostics`)
        // holding the translator lock across an LSP round-trip.
        let lock_acquired = Arc::new(tokio::sync::Notify::new());
        let notify = Arc::clone(&lock_acquired);
        let holder = tokio::spawn(async move {
            let _guard = translator.lock().await;
            notify.notify_one();
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        });
        lock_acquired.notified().await;

        tokio::spawn(diagnostics_pump(
            ServerId::from_static("rust"),
            rx,
            lifecycle_rx,
            cancel,
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: Arc::clone(&cache),
                subs,
                workspace_roots: test_workspace_roots(),
            },
        ));

        let uri: Uri = test_uri("locked.rs");
        tx.send(LspNotification::PublishDiagnostics(
            PublishDiagnosticsParams {
                uri: uri.clone(),
                diagnostics: vec![],
                version: None,
            },
        ))
        .await
        .unwrap();
        drop(tx);

        // Well within the 2 s translator lock hold: a translator-locking
        // pump would still be blocked at this point.
        tokio::time::timeout(std::time::Duration::from_millis(500), async {
            loop {
                {
                    let guard = cache.lock().await;
                    if guard.diagnostics(&uri).is_some() {
                        return;
                    }
                }
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("pump stalled behind translator lock");

        holder.await.unwrap();
    }

    /// The `Other` arm (custom/unrecognized notifications, e.g.
    /// rust-analyzer's `experimental/serverStatus`) must reach
    /// `NotificationCache::observe_indexing_signal` via the lifecycle
    /// lane -- this is the one place in production that notification
    /// actually gets from the LSP transport into the indexing-readiness
    /// gate; every other test for the gate pre-seeds the cache by hand
    /// and would not have caught a pump wiring regression.
    #[tokio::test]
    async fn test_pump_routes_other_notifications_to_indexing_signal() {
        let cache = make_cache();
        let subs = make_subs();
        let (_tx, rx) = pump_feed();
        let (lifecycle_tx, lifecycle_rx) = mpsc::channel(8);
        let cancel = CancellationToken::new();
        let server_id = ServerId::from_static("rust");

        tokio::spawn(diagnostics_pump(
            server_id.clone(),
            rx,
            lifecycle_rx,
            cancel,
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: Arc::clone(&cache),
                subs,
                workspace_roots: test_workspace_roots(),
            },
        ));

        lifecycle_tx
            .send(LspNotification::Other {
                method: std::borrow::Cow::Borrowed("experimental/serverStatus"),
                params: Some(serde_json::json!({"quiescent": false})),
            })
            .await
            .unwrap();
        drop(lifecycle_tx);

        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            loop {
                {
                    let guard = cache.lock().await;
                    if guard.indexing_state(&server_id) == IndexingState::Loading {
                        return;
                    }
                }
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
        })
        .await
        .expect(
            "pump did not route the Other{experimental/serverStatus} notification into \
             NotificationCache::observe_indexing_signal within 5s",
        );
    }

    /// P3: a `$/progress` `report` frame must never reach the lifecycle
    /// lane at all (S3) -- classified and dropped by
    /// `LspClient::message_loop_inner` before enqueueing, not merely
    /// ignored once received. This test exercises the pump side: even
    /// if a `report` somehow arrived on the lifecycle lane, the pump
    /// itself only recognizes `begin`/`end` shapes via
    /// `NotificationCache::observe_progress`, so a `begin` sent
    /// afterward must still be the one that flips the state.
    #[tokio::test]
    async fn test_pump_routes_progress_begin_to_indexing_signal() {
        let cache = make_cache();
        let subs = make_subs();
        let (_tx, rx) = pump_feed();
        let (lifecycle_tx, lifecycle_rx) = mpsc::channel(8);
        let cancel = CancellationToken::new();
        let server_id = ServerId::from_static("gopls");

        tokio::spawn(diagnostics_pump(
            server_id.clone(),
            rx,
            lifecycle_rx,
            cancel,
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: Arc::clone(&cache),
                subs,
                workspace_roots: test_workspace_roots(),
            },
        ));

        lifecycle_tx
            .send(LspNotification::Progress(lsp_types::ProgressParams {
                token: lsp_types::ProgressToken::String("indexing".to_string()),
                value: serde_json::json!({"kind": "begin", "title": "Loading"}),
            }))
            .await
            .unwrap();
        drop(lifecycle_tx);

        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            loop {
                {
                    let guard = cache.lock().await;
                    if guard.indexing_state(&server_id) == IndexingState::Loading {
                        return;
                    }
                }
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
        })
        .await
        .expect(
            "pump did not route the Progress(begin) notification into \
             NotificationCache::observe_progress within 5s",
        );
    }

    /// P3: saturating the diagnostics lane to capacity must not stall the
    /// lifecycle lane -- a `begin`/`end` frame arriving while the
    /// notification lane is backed up must still reach the readiness
    /// gate promptly.
    #[tokio::test]
    async fn test_lifecycle_lane_unaffected_by_saturated_notification_lane() {
        let cache = make_cache();
        let subs = make_subs();
        let (tx, rx) = pump_feed();
        let (lifecycle_tx, lifecycle_rx) = mpsc::channel(8);
        let cancel = CancellationToken::new();
        let server_id = ServerId::from_static("gopls");

        // Fill the notification lane to capacity before the pump drains it, forcing a backlog.
        for _ in 0..2 {
            tx.send(LspNotification::PublishDiagnostics(
                PublishDiagnosticsParams {
                    uri: test_uri("saturate.rs"),
                    diagnostics: vec![],
                    version: None,
                },
            ))
            .await
            .unwrap();
        }

        tokio::spawn(diagnostics_pump(
            server_id.clone(),
            rx,
            lifecycle_rx,
            cancel,
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: Arc::clone(&cache),
                subs,
                workspace_roots: test_workspace_roots(),
            },
        ));

        lifecycle_tx
            .send(LspNotification::Other {
                method: std::borrow::Cow::Borrowed("experimental/serverStatus"),
                params: Some(serde_json::json!({"quiescent": false})),
            })
            .await
            .unwrap();
        drop(tx);
        drop(lifecycle_tx);

        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            loop {
                {
                    let guard = cache.lock().await;
                    if guard.indexing_state(&server_id) == IndexingState::Loading {
                        return;
                    }
                }
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("lifecycle lane must still be served while the notification lane is backed up");
    }

    use crate::test_lsp::{pump_feed, spawn_test_pump, spawn_test_pump_with_cache};

    fn test_mcp_uri(file: &str) -> DiagnosticsResourceUri {
        DiagnosticsResourceUri::for_published(&PublishedDiagnosticsUri::for_test(
            test_uri(file),
            test_uri(file),
        ))
        .unwrap()
    }

    fn publish(file: &str) -> LspNotification {
        LspNotification::PublishDiagnostics(PublishDiagnosticsParams {
            uri: test_uri(file),
            diagnostics: vec![],
            version: None,
        })
    }

    async fn recv_within(rx: &mut mpsc::Receiver<String>) -> String {
        tokio::time::timeout(std::time::Duration::from_secs(5), rx.recv())
            .await
            .expect("no resources/updated within 5 s")
            .expect("delivery channel closed")
    }

    /// #468 isolation: a session never receives updates for URIs it did not
    /// subscribe to. B's first message must be Y even though X was
    /// published first (ordering, not a timeout).
    #[tokio::test]
    async fn test_pump_delivers_only_to_subscribed_sessions() {
        use crate::mcp::{SessionHandle, Target};

        let subs = make_subs();
        let session_a = SessionHandle::new(subs.clone());
        let session_b = SessionHandle::new(subs.clone());
        let (tx_a, mut rx_a) = mpsc::channel(8);
        let (tx_b, mut rx_b) = mpsc::channel(8);
        let (x, y) = (test_mcp_uri("x.rs"), test_mcp_uri("y.rs"));
        for uri in [&x, &y] {
            session_a
                .subscribe_for_test(uri, Target::Channel(tx_a.clone()))
                .await
                .unwrap();
        }
        session_b
            .subscribe_for_test(&y, Target::Channel(tx_b.clone()))
            .await
            .unwrap();

        let (tx, _cancel_tx) = spawn_test_pump(subs, test_workspace_roots());
        tx.send(publish("x.rs")).await.unwrap();
        assert_eq!(recv_within(&mut rx_a).await, x.as_str());
        // Gives a wrongly queued X time to reach B before Y exists.
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        assert_matches!(
            rx_b.try_recv(),
            Err(mpsc::error::TryRecvError::Empty),
            "B never subscribed to X"
        );

        tx.send(publish("y.rs")).await.unwrap();
        assert_eq!(recv_within(&mut rx_b).await, y.as_str());
        assert_eq!(recv_within(&mut rx_a).await, y.as_str());
    }

    /// A restart tells the subscribers of the diagnostics it cleared to
    /// re-read them, matching the cache key to the subscription URI even
    /// for a percent-encoded path, and leaves other subscribers alone.
    #[tokio::test]
    async fn test_publish_invalidated_notifies_only_subscribers_of_cleared_files() {
        use crate::mcp::{SessionHandle, Target};

        let workspace = tempfile::TempDir::new().unwrap();
        let root = dunce::canonicalize(workspace.path()).unwrap();
        let cleared_file = root.join("a b.rs");
        let other_file = root.join("other.rs");
        for file in [&cleared_file, &other_file] {
            std::fs::write(file, "fn main() {}").unwrap();
        }

        let cache = make_cache();
        let id = ServerId::from_static("rust");
        let roots = WorkspaceRoots::from_paths(&[root]).unwrap();
        let published = bridge::resolve_one(&bridge::path_to_uri(&cleared_file).unwrap(), &roots)
            .await
            .unwrap();
        let error = lsp_types::Diagnostic {
            message: "boom".to_owned().into(),
            ..Default::default()
        };
        let cleared = {
            let mut cache = cache.lock().await;
            cache.store_published_diagnostics(&id, &published, None, vec![error]);
            cache.clear_server_diagnostics(&id)
        };
        assert!(!cleared.is_empty());

        let subs = make_subs();
        let session = SessionHandle::new(subs.clone());
        let (tx_cleared, mut rx_cleared) = mpsc::channel(8);
        let (tx_other, mut rx_other) = mpsc::channel(8);
        let cleared_uri = bridge::resources::make_uri(&cleared_file).unwrap();
        let other_uri = bridge::resources::make_uri(&other_file).unwrap();
        session
            .subscribe_for_test(
                &DiagnosticsResourceUri::for_test(&cleared_uri),
                Target::Channel(tx_cleared.clone()),
            )
            .await
            .unwrap();
        session
            .subscribe_for_test(
                &DiagnosticsResourceUri::for_test(&other_uri),
                Target::Channel(tx_other.clone()),
            )
            .await
            .unwrap();
        let cancel = CancellationToken::new();
        let wiring = PumpWiring {
            shared: PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: cache,
                subs,
                workspace_roots: roots,
            },
            cancel,
        };

        bridge::NotificationWiring::publish_invalidated(&wiring, &cleared).await;

        assert_eq!(recv_within(&mut rx_cleared).await, cleared_uri);
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        assert_matches!(rx_other.try_recv(), Err(mpsc::error::TryRecvError::Empty));
    }

    /// #574 FR-007: a file that already has a pulled slot is notified exactly
    /// once per accepted push, as before the pulled slot existed.
    #[tokio::test]
    async fn test_push_beside_a_pulled_slot_notifies_exactly_once() {
        use crate::mcp::{SessionHandle, Target};

        let subs = make_subs();
        let session = SessionHandle::new(subs.clone());
        let (tx_session, mut rx_session) = mpsc::channel(8);
        let x = test_mcp_uri("x.rs");
        session
            .subscribe_for_test(&x, Target::Channel(tx_session.clone()))
            .await
            .unwrap();
        let cache = make_cache();
        cache.lock().await.store_pulled_for_test(
            &ServerId::from_static("rust"),
            &test_uri("x.rs"),
            vec![lsp_types::Diagnostic::default()],
        );

        let (tx, _cancel_tx) =
            spawn_test_pump_with_cache(subs, test_workspace_roots(), Arc::clone(&cache));
        tx.send(publish("x.rs")).await.unwrap();

        assert_eq!(recv_within(&mut rx_session).await, x.as_str());
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        assert_matches!(rx_session.try_recv(), Err(mpsc::error::TryRecvError::Empty));
        assert!(
            cache
                .lock()
                .await
                .diagnostic_sources(&test_uri("x.rs"))
                .merge()
                .is_some_and(|info| info.diagnostics.len() == 1)
        );
    }

    /// #649: a push that evicts another file's entry tells that file's
    /// subscribers, and nobody who did not subscribe to it.
    #[tokio::test]
    async fn test_push_eviction_notifies_the_evicted_files_subscribers() {
        use crate::mcp::{SessionHandle, Target};

        let subs = make_subs();
        let session = SessionHandle::new(subs.clone());
        let (tx_evicted, mut rx_evicted) = mpsc::channel(8);
        let (tx_other, mut rx_other) = mpsc::channel(8);
        let (evicted, other) = (test_mcp_uri("first.rs"), test_mcp_uri("filler1.rs"));
        session
            .subscribe_for_test(&evicted, Target::Channel(tx_evicted.clone()))
            .await
            .unwrap();
        session
            .subscribe_for_test(&other, Target::Channel(tx_other.clone()))
            .await
            .unwrap();
        let cache = make_cache();
        let error = lsp_types::Diagnostic::default();
        let mut guard = cache.lock().await;
        let server = ServerId::from_static("rust");
        guard.store_diagnostics(&server, &test_uri("first.rs"), None, vec![error.clone()]);
        for i in 1..bridge::MAX_DIAGNOSTIC_ENTRIES {
            guard.store_diagnostics(
                &server,
                &test_uri(&format!("filler{i}.rs")),
                None,
                vec![error.clone()],
            );
        }
        drop(guard);

        let (tx, _cancel_tx) =
            spawn_test_pump_with_cache(subs, test_workspace_roots(), Arc::clone(&cache));
        tx.send(publish("new.rs")).await.unwrap();

        assert_eq!(recv_within(&mut rx_evicted).await, evicted.as_str());
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        assert_matches!(
            rx_evicted.try_recv(),
            Err(mpsc::error::TryRecvError::Empty),
            "notified once"
        );
        assert_matches!(rx_other.try_recv(), Err(mpsc::error::TryRecvError::Empty));
    }

    /// A pull that changed a file notifies the subscribers of that file only.
    #[tokio::test]
    async fn test_publish_changed_notifies_only_subscribers_of_the_file() {
        use crate::mcp::{SessionHandle, Target};

        let subs = make_subs();
        let session = SessionHandle::new(subs.clone());
        let (tx_x, mut rx_x) = mpsc::channel(8);
        let (tx_y, mut rx_y) = mpsc::channel(8);
        let (x, y) = (test_mcp_uri("x.rs"), test_mcp_uri("y.rs"));
        session
            .subscribe_for_test(&x, Target::Channel(tx_x.clone()))
            .await
            .unwrap();
        session
            .subscribe_for_test(&y, Target::Channel(tx_y.clone()))
            .await
            .unwrap();
        let cancel = CancellationToken::new();
        let wiring = PumpWiring {
            shared: PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: make_cache(),
                subs,
                workspace_roots: test_workspace_roots(),
            },
            cancel,
        };

        bridge::NotificationWiring::publish_changed(&wiring, &test_uri("x.rs")).await;

        assert_eq!(recv_within(&mut rx_x).await, x.as_str());
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        assert_matches!(rx_y.try_recv(), Err(mpsc::error::TryRecvError::Empty));
    }

    /// With nobody subscribed a changed pull builds no URI and queues nothing.
    #[tokio::test]
    async fn test_publish_changed_without_subscribers_is_a_no_op() {
        let cancel = CancellationToken::new();
        let wiring = PumpWiring {
            shared: PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: make_cache(),
                subs: make_subs(),
                workspace_roots: test_workspace_roots(),
            },
            cancel,
        };

        bridge::NotificationWiring::publish_changed(&wiring, &test_uri("x.rs")).await;
    }

    /// The wiring reports whether any session subscribed, so a caller can skip
    /// work only subscribers care about.
    #[tokio::test]
    async fn test_wiring_reports_whether_anything_is_subscribed() {
        use crate::mcp::{SessionHandle, Target};

        let subs = make_subs();
        let cancel = CancellationToken::new();
        let wiring = PumpWiring {
            shared: PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: make_cache(),
                subs: subs.clone(),
                workspace_roots: test_workspace_roots(),
            },
            cancel,
        };
        assert!(!bridge::NotificationWiring::has_subscriptions(&wiring).await);

        let session = SessionHandle::new(subs);
        let (tx_session, _rx_session) = mpsc::channel(8);
        session
            .subscribe_for_test(&test_mcp_uri("x.rs"), Target::Channel(tx_session))
            .await
            .unwrap();

        assert!(bridge::NotificationWiring::has_subscriptions(&wiring).await);
    }

    /// #532: diagnostics a server publishes through a symlinked spelling
    /// notify the subscriber of the canonical URI and read back under the
    /// canonical key; a link pointing outside the workspace is dropped.
    #[cfg(unix)]
    #[tokio::test]
    async fn test_pump_keys_symlinked_publish_by_canonical_path() {
        use crate::mcp::{SessionHandle, Target};

        let workspace = tempfile::TempDir::new().unwrap();
        let root = dunce::canonicalize(workspace.path()).unwrap();
        let file = root.join("main.rs");
        std::fs::write(&file, "fn main() {}").unwrap();
        let link = root.join("link.rs");
        std::os::unix::fs::symlink(&file, &link).unwrap();
        let outside = tempfile::TempDir::new().unwrap();
        std::fs::write(outside.path().join("x.rs"), "").unwrap();
        let escape = root.join("escape.rs");
        std::os::unix::fs::symlink(outside.path().join("x.rs"), &escape).unwrap();

        let cache = make_cache();
        let subs = make_subs();
        let session = SessionHandle::new(subs.clone());
        let (tx_session, mut rx_session) = mpsc::channel(8);
        let canonical = bridge::resources::make_uri(&file).unwrap();
        session
            .subscribe_for_test(
                &DiagnosticsResourceUri::for_test(&canonical),
                Target::Channel(tx_session),
            )
            .await
            .unwrap();

        let (tx, _cancel_tx) = spawn_test_pump_with_cache(
            subs,
            WorkspaceRoots::from_paths(&[root]).unwrap(),
            Arc::clone(&cache),
        );
        let error = lsp_types::Diagnostic {
            message: "boom".to_owned().into(),
            ..Default::default()
        };
        for (path, diagnostics) in [(&escape, vec![error.clone()]), (&link, vec![error])] {
            tx.send(LspNotification::PublishDiagnostics(
                PublishDiagnosticsParams {
                    uri: bridge::path_to_uri(path).unwrap(),
                    diagnostics,
                    version: None,
                },
            ))
            .await
            .unwrap();
        }

        assert_eq!(recv_within(&mut rx_session).await, canonical);
        let canonical_lsp = bridge::path_to_uri(&file).unwrap();
        let (info, escaped) = {
            let guard = cache.lock().await;
            (
                guard.diagnostic_sources(&canonical_lsp),
                guard.has_diagnostics(&bridge::path_to_uri(&escape).unwrap()),
            )
        };
        assert_eq!(info.merge().unwrap().diagnostics.len(), 1);
        assert!(
            !escaped,
            "a symlink pointing outside the workspace must be dropped"
        );
    }

    /// A clear whose path hits a persistent transient filesystem error is
    /// still applied, and a later publication for the same URI wins.
    #[tokio::test]
    async fn test_pump_clear_survives_transient_error_and_stays_ordered() {
        let workspace = tempfile::TempDir::new().unwrap();
        let root = dunce::canonicalize(workspace.path()).unwrap();
        let failing: Arc<bridge::CanonicalizeFn> =
            Arc::new(|_: &std::path::Path| Err(std::io::Error::from(std::io::ErrorKind::TimedOut)));
        let cache = make_cache();
        let (tx, rx) = pump_feed();
        let (_lifecycle_tx, lifecycle_rx) = mpsc::channel(8);
        let cancel = CancellationToken::new();
        tokio::spawn(diagnostics_pump_with_resolver(
            ServerId::from_static("rust"),
            rx,
            lifecycle_rx,
            cancel,
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: Arc::clone(&cache),
                subs: make_subs(),
                workspace_roots: WorkspaceRoots::from_paths(std::slice::from_ref(&root)).unwrap(),
            },
            PublishedPathResolver::with_canonicalizer(failing),
        ));
        let uri = bridge::path_to_uri(&root.join("a.rs")).unwrap();
        tx.send(LspNotification::PublishDiagnostics(
            PublishDiagnosticsParams {
                uri: uri.clone(),
                diagnostics: vec![],
                version: None,
            },
        ))
        .await
        .unwrap();

        tokio::time::timeout(Duration::from_secs(5), async {
            while !cache.lock().await.has_diagnostics(&uri) {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("the clear must be cached despite the transient error");
    }

    /// A lifecycle notification is applied while a diagnostics batch is
    /// still waiting on a slow filesystem.
    #[tokio::test]
    async fn test_pump_services_lifecycle_lane_while_a_batch_resolves() {
        let workspace = tempfile::TempDir::new().unwrap();
        let root = dunce::canonicalize(workspace.path()).unwrap();
        let (release, gate) = std::sync::mpsc::channel::<()>();
        let gate = std::sync::Mutex::new(gate);
        let slow: Arc<bridge::CanonicalizeFn> = Arc::new(move |p: &std::path::Path| {
            gate.lock().map(|rx| rx.recv()).ok();
            Ok(p.to_path_buf())
        });
        let cache = make_cache();
        let (tx, rx) = pump_feed();
        let (lifecycle_tx, lifecycle_rx) = mpsc::channel(8);
        let cancel = CancellationToken::new();
        let server_id = ServerId::from_static("rust");
        tokio::spawn(diagnostics_pump_with_resolver(
            server_id.clone(),
            rx,
            lifecycle_rx,
            cancel,
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: Arc::clone(&cache),
                subs: make_subs(),
                workspace_roots: WorkspaceRoots::from_paths(std::slice::from_ref(&root)).unwrap(),
            },
            PublishedPathResolver::with_canonicalizer(slow),
        ));
        tx.send(LspNotification::PublishDiagnostics(
            PublishDiagnosticsParams {
                uri: bridge::path_to_uri(&root.join("a.rs")).unwrap(),
                diagnostics: vec![],
                version: None,
            },
        ))
        .await
        .unwrap();
        tokio::time::sleep(Duration::from_millis(50)).await;

        lifecycle_tx
            .send(LspNotification::Other {
                method: std::borrow::Cow::Borrowed("experimental/serverStatus"),
                params: Some(serde_json::json!({"quiescent": false})),
            })
            .await
            .unwrap();

        tokio::time::timeout(Duration::from_secs(5), async {
            while cache.lock().await.indexing_state(&server_id) != IndexingState::Loading {
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("lifecycle lane must be serviced while the batch is still resolving");
        drop(release);
    }

    /// #658: a `$/typescriptVersion` taken by the lifecycle arm that runs
    /// while a diagnostics batch resolves gets the pin-ignored warning too.
    #[tokio::test]
    async fn test_pump_warns_about_ignored_pin_while_a_batch_resolves() {
        use tracing_subscriber::layer::SubscriberExt as _;

        use crate::test_lsp::CapturedLogs;

        let captured = CapturedLogs::default();
        let _guard =
            tracing::subscriber::set_default(tracing_subscriber::registry().with(captured.clone()));
        let workspace = tempfile::TempDir::new().unwrap();
        let root = dunce::canonicalize(workspace.path()).unwrap();
        let (release, gate) = std::sync::mpsc::channel::<()>();
        let gate = std::sync::Mutex::new(gate);
        let slow: Arc<bridge::CanonicalizeFn> = Arc::new(move |p: &std::path::Path| {
            gate.lock().map(|rx| rx.recv()).ok();
            Ok(p.to_path_buf())
        });
        let (tx, rx) = pump_feed();
        let (lifecycle_tx, lifecycle_rx) = mpsc::channel(8);
        let cancel = CancellationToken::new();
        tokio::spawn(diagnostics_pump_with_resolver(
            ServerId::from_static("tsls"),
            rx,
            lifecycle_rx,
            cancel,
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            Some(PathBuf::from("/pin/tsserver.js")),
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: make_cache(),
                subs: make_subs(),
                workspace_roots: WorkspaceRoots::from_paths(std::slice::from_ref(&root)).unwrap(),
            },
            PublishedPathResolver::with_canonicalizer(slow),
        ));
        tx.send(LspNotification::PublishDiagnostics(
            PublishDiagnosticsParams {
                uri: bridge::path_to_uri(&root.join("a.ts")).unwrap(),
                diagnostics: vec![],
                version: None,
            },
        ))
        .await
        .unwrap();
        tokio::time::sleep(Duration::from_millis(50)).await;

        lifecycle_tx
            .send(LspNotification::Other {
                method: "$/typescriptVersion".into(),
                params: Some(serde_json::json!({"version": "5.0", "source": "bundled"})),
            })
            .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(5)).await;
            }
        })
        .await
        .expect("the warning must not depend on which arm took the notification");
        drop(release);
    }

    /// #658: a pump panic is contained, logged, and degrades its server;
    /// a pump that ends normally leaves the server alone.
    #[tokio::test]
    async fn test_contain_pump_panic_degrades_only_after_a_panic() {
        let cache = Mutex::new(NotificationCache::new());
        let id = ServerId::from_static("rust");
        cache.lock().await.observe_indexing_signal(
            &id,
            "experimental/serverStatus",
            Some(&serde_json::json!({"quiescent": true})),
        );

        contain_pump_panic(async {}, &id, &cache).await;
        assert!(!cache.lock().await.is_push_degraded(&id));
        assert_eq!(cache.lock().await.indexing_state(&id), IndexingState::Ready);

        contain_pump_panic(
            async {
                panic!("pump boom");
            },
            &id,
            &cache,
        )
        .await;
        let guard = cache.lock().await;
        assert!(guard.is_push_degraded(&id));
        assert_eq!(guard.indexing_state(&id), IndexingState::Unknown);
    }

    /// A demotion to `Secondary` while a batch is still resolving stops
    /// that batch from being cached.
    #[tokio::test]
    async fn test_pump_demotion_during_resolve_stops_caching() {
        let workspace = tempfile::TempDir::new().unwrap();
        let root = dunce::canonicalize(workspace.path()).unwrap();
        let (release, gate) = std::sync::mpsc::channel::<()>();
        let gate = std::sync::Mutex::new(gate);
        let slow: Arc<bridge::CanonicalizeFn> = Arc::new(move |p: &std::path::Path| {
            gate.lock().map(|rx| rx.recv()).ok();
            Ok(p.to_path_buf())
        });
        let cache = make_cache();
        let (tx, rx) = pump_feed();
        let (_lifecycle_tx, lifecycle_rx) = mpsc::channel(8);
        let cancel = CancellationToken::new();
        let (role_tx, role_rx) = watch::channel(DiagnosticsRole::Authoritative);
        tokio::spawn(diagnostics_pump_with_resolver(
            ServerId::from_static("rust"),
            rx,
            lifecycle_rx,
            cancel,
            role_rx,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: Arc::clone(&cache),
                subs: make_subs(),
                workspace_roots: WorkspaceRoots::from_paths(std::slice::from_ref(&root)).unwrap(),
            },
            PublishedPathResolver::with_canonicalizer(slow),
        ));
        let uri = bridge::path_to_uri(&root.join("a.rs")).unwrap();
        let error = lsp_types::Diagnostic {
            message: "boom".to_owned().into(),
            ..Default::default()
        };
        tx.send(LspNotification::PublishDiagnostics(
            PublishDiagnosticsParams {
                uri: uri.clone(),
                diagnostics: vec![error],
                version: None,
            },
        ))
        .await
        .unwrap();
        tx.send(LspNotification::LogMessage(lsp_types::LogMessageParams {
            kind: lsp_types::MessageType::Log,
            message: "fence".to_owned(),
        }))
        .await
        .unwrap();
        tokio::time::sleep(Duration::from_millis(50)).await;

        role_tx.send(DiagnosticsRole::Secondary).unwrap();
        drop(release);

        tokio::time::timeout(Duration::from_secs(5), async {
            while cache.lock().await.logs_count() == 0 {
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("the log after the publication must be applied");
        assert!(!cache.lock().await.has_diagnostics(&uri));
    }

    /// Cancellation must end the pump even while a canonicalization is
    /// stuck on a hung filesystem.
    #[tokio::test]
    async fn test_pump_cancel_completes_while_canonicalize_hangs() {
        let workspace = tempfile::TempDir::new().unwrap();
        let root = dunce::canonicalize(workspace.path()).unwrap();
        let (release, gate) = std::sync::mpsc::channel::<()>();
        let gate = std::sync::Mutex::new(gate);
        let hung: Arc<bridge::CanonicalizeFn> = Arc::new(move |p: &std::path::Path| {
            gate.lock().map(|rx| rx.recv()).ok();
            Ok(p.to_path_buf())
        });
        let (tx, rx) = pump_feed();
        let (_lifecycle_tx, lifecycle_rx) = mpsc::channel(8);
        let cancel = CancellationToken::new();
        let pump = tokio::spawn(diagnostics_pump_with_resolver(
            ServerId::from_static("rust"),
            rx,
            lifecycle_rx,
            cancel.clone(),
            tokio::sync::watch::channel(crate::bridge::DiagnosticsRole::Authoritative).1,
            None,
            PumpShared {
                roles: DiagnosticsRoles::default(),
                notification_cache: make_cache(),
                subs: make_subs(),
                workspace_roots: WorkspaceRoots::from_paths(std::slice::from_ref(&root)).unwrap(),
            },
            PublishedPathResolver::with_canonicalizer(hung),
        ));
        tx.send(LspNotification::PublishDiagnostics(
            PublishDiagnosticsParams {
                uri: bridge::path_to_uri(&root.join("a.rs")).unwrap(),
                diagnostics: vec![],
                version: None,
            },
        ))
        .await
        .unwrap();
        tokio::time::sleep(Duration::from_millis(50)).await;

        cancel.cancel();

        tokio::time::timeout(Duration::from_secs(2), pump)
            .await
            .expect("pump must exit promptly on cancel")
            .unwrap();
        drop(release);
    }

    /// A cold burst resolved in parallel is still applied in arrival order:
    /// a clear published after an error for the same file wins, and no
    /// publication of the burst is lost.
    #[tokio::test]
    async fn test_pump_cold_burst_keeps_clear_after_error_and_loses_nothing() {
        let workspace = tempfile::TempDir::new().unwrap();
        let root = dunce::canonicalize(workspace.path()).unwrap();
        let cache = make_cache();
        let (tx, _cancel_tx) = spawn_test_pump_with_cache(
            make_subs(),
            WorkspaceRoots::from_paths(std::slice::from_ref(&root)).unwrap(),
            Arc::clone(&cache),
        );
        let error = lsp_types::Diagnostic {
            message: "boom".to_owned().into(),
            ..Default::default()
        };
        let publish = |name: String, diagnostics: Vec<lsp_types::Diagnostic>| {
            LspNotification::PublishDiagnostics(PublishDiagnosticsParams {
                uri: bridge::path_to_uri(&root.join(name)).unwrap(),
                diagnostics,
                version: None,
            })
        };

        tx.send(publish("a.rs".to_owned(), vec![error.clone()]))
            .await
            .unwrap();
        for i in 0..150 {
            tx.send(publish(format!("f{i}.rs"), vec![error.clone()]))
                .await
                .unwrap();
        }
        tx.send(publish("a.rs".to_owned(), vec![])).await.unwrap();
        tx.send(publish("last.rs".to_owned(), vec![error.clone()]))
            .await
            .unwrap();

        let last = bridge::path_to_uri(&root.join("last.rs")).unwrap();
        tokio::time::timeout(Duration::from_secs(10), async {
            while !cache.lock().await.has_diagnostics(&last) {
                tokio::task::yield_now().await;
            }
        })
        .await
        .unwrap();

        let a = bridge::path_to_uri(&root.join("a.rs")).unwrap();
        let (missing, remaining) = {
            let guard = cache.lock().await;
            let missing: Vec<usize> = (0..150)
                .filter(|i| {
                    let uri = bridge::path_to_uri(&root.join(format!("f{i}.rs"))).unwrap();
                    !guard.has_diagnostics(&uri)
                })
                .collect();
            let remaining = guard
                .diagnostic_sources(&a)
                .merge()
                .map_or(0, |info| info.diagnostics.len());
            drop(guard);
            (missing, remaining)
        };
        assert!(missing.is_empty(), "publications missing: {missing:?}");
        assert_eq!(remaining, 0, "clear was reordered");
    }

    /// A server publishing under the configured spelling of a root that is a
    /// symlink (or the logical `$PWD`) is not dropped by the pre-filter: the
    /// diagnostics are cached under the canonical key.
    #[cfg(unix)]
    #[tokio::test]
    async fn test_pump_accepts_publish_under_a_configured_root_alias() {
        let workspace = tempfile::TempDir::new().unwrap();
        let base = dunce::canonicalize(workspace.path()).unwrap();
        let real = base.join("real");
        std::fs::create_dir(&real).unwrap();
        std::fs::write(real.join("main.rs"), "fn main() {}").unwrap();
        let link = base.join("link");
        std::os::unix::fs::symlink(&real, &link).unwrap();
        let roots = WorkspaceRoots::from_paths(std::slice::from_ref(&link)).unwrap();
        assert_eq!(roots.canonical(), std::slice::from_ref(&real));

        let cache = make_cache();
        let subs = make_subs();
        let (tx, _cancel_tx) = spawn_test_pump_with_cache(subs, roots, Arc::clone(&cache));
        let error = lsp_types::Diagnostic {
            message: "boom".to_owned().into(),
            ..Default::default()
        };
        tx.send(LspNotification::PublishDiagnostics(
            PublishDiagnosticsParams {
                uri: bridge::path_to_uri(&link.join("main.rs")).unwrap(),
                diagnostics: vec![error],
                version: None,
            },
        ))
        .await
        .unwrap();

        let canonical_uri = bridge::path_to_uri(&real.join("main.rs")).unwrap();
        let info = tokio::time::timeout(std::time::Duration::from_secs(5), async {
            loop {
                let sources = cache.lock().await.diagnostic_sources(&canonical_uri);
                if let Some(info) = sources.merge() {
                    return info;
                }
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
        })
        .await;
        let info = info.unwrap_or_else(|_| panic!("alias publish was dropped by the pump"));
        assert_eq!(info.diagnostics.len(), 1);
    }

    /// #532: every stored publish notifies the canonical subscriber, and
    /// an empty canonical publish after errors on an alias keeps the errors.
    #[cfg(unix)]
    #[tokio::test]
    async fn test_pump_notifies_canonical_key_on_each_alias_publish() {
        use crate::mcp::{SessionHandle, Target};

        let workspace = tempfile::TempDir::new().unwrap();
        let root = dunce::canonicalize(workspace.path()).unwrap();
        let file = root.join("main.rs");
        std::fs::write(&file, "fn main() {}").unwrap();
        let link = root.join("link.rs");
        std::os::unix::fs::symlink(&file, &link).unwrap();

        let cache = make_cache();
        let subs = make_subs();
        let session = SessionHandle::new(subs.clone());
        let (tx_session, mut rx_session) = mpsc::channel(8);
        let canonical = bridge::resources::make_uri(&file).unwrap();
        session
            .subscribe_for_test(
                &DiagnosticsResourceUri::for_test(&canonical),
                Target::Channel(tx_session),
            )
            .await
            .unwrap();
        let (tx, _cancel_tx) = spawn_test_pump_with_cache(
            subs,
            WorkspaceRoots::from_paths(&[root]).unwrap(),
            Arc::clone(&cache),
        );
        let error = lsp_types::Diagnostic {
            message: "boom".to_owned().into(),
            ..Default::default()
        };
        let publishes = [(&link, vec![error]), (&file, vec![]), (&link, vec![])];
        for (path, diagnostics) in publishes {
            tx.send(LspNotification::PublishDiagnostics(
                PublishDiagnosticsParams {
                    uri: bridge::path_to_uri(path).unwrap(),
                    diagnostics,
                    version: None,
                },
            ))
            .await
            .unwrap();
            assert_eq!(recv_within(&mut rx_session).await, canonical);
        }

        let canonical_lsp = bridge::path_to_uri(&file).unwrap();
        let info = cache
            .lock()
            .await
            .diagnostic_sources(&canonical_lsp)
            .merge()
            .unwrap();
        assert!(info.diagnostics.is_empty(), "the alias's clear must apply");
    }

    /// #692: a publish the cache refuses (the file's URI spellings are at their
    /// cap) is not announced; the next accepted publish is.
    #[cfg(unix)]
    #[tokio::test]
    async fn test_pump_does_not_notify_for_a_dropped_publish() {
        use crate::mcp::{SessionHandle, Target};

        let workspace = tempfile::TempDir::new().unwrap();
        let root = dunce::canonicalize(workspace.path()).unwrap();
        let file = root.join("main.rs");
        std::fs::write(&file, "fn main() {}").unwrap();
        let links: Vec<PathBuf> = (0..9)
            .map(|n| {
                let link = root.join(format!("link{n}.rs"));
                std::os::unix::fs::symlink(&file, &link).unwrap();
                link
            })
            .collect();

        let cache = make_cache();
        let subs = make_subs();
        let session = SessionHandle::new(subs.clone());
        let (tx_session, mut rx_session) = mpsc::channel(32);
        let canonical = bridge::resources::make_uri(&file).unwrap();
        session
            .subscribe_for_test(
                &DiagnosticsResourceUri::for_test(&canonical),
                Target::Channel(tx_session),
            )
            .await
            .unwrap();
        let (tx, _cancel_tx) = spawn_test_pump_with_cache(
            subs,
            WorkspaceRoots::from_paths(&[root]).unwrap(),
            Arc::clone(&cache),
        );
        let publish = |path: PathBuf| {
            let tx = &tx;
            async move {
                tx.send(LspNotification::PublishDiagnostics(
                    PublishDiagnosticsParams {
                        uri: bridge::path_to_uri(&path).unwrap(),
                        diagnostics: vec![lsp_types::Diagnostic {
                            message: "boom".to_owned().into(),
                            ..Default::default()
                        }],
                        version: None,
                    },
                ))
                .await
                .unwrap();
            }
        };

        for link in links.iter().take(8) {
            publish(link.clone()).await;
            assert_eq!(recv_within(&mut rx_session).await, canonical);
        }
        publish(links[8].clone()).await;
        publish(file.clone()).await;

        assert_eq!(recv_within(&mut rx_session).await, canonical);
        tokio::task::yield_now().await;
        assert!(
            rx_session.try_recv().is_err(),
            "the dropped publish must not have been announced"
        );
    }

    /// #468: a session whose peer stopped reading neither blocks the pump
    /// nor other sessions, and loses no URI once it resumes.
    #[tokio::test]
    async fn test_pump_stalled_session_does_not_block_others_or_lose_updates() {
        use crate::mcp::{SessionHandle, Target};

        const URI_COUNT: usize = 5;

        let subs = make_subs();
        let stalled = SessionHandle::new(subs.clone());
        let healthy = SessionHandle::new(subs.clone());
        let (tx_stalled, mut rx_stalled) = mpsc::channel(1);
        let (tx_healthy, mut rx_healthy) = mpsc::channel(URI_COUNT);
        let expected: HashSet<DiagnosticsResourceUri> = (0..URI_COUNT)
            .map(|i| test_mcp_uri(&format!("f{i}.rs")))
            .collect();
        for uri in &expected {
            stalled
                .subscribe_for_test(uri, Target::Channel(tx_stalled.clone()))
                .await
                .unwrap();
            healthy
                .subscribe_for_test(uri, Target::Channel(tx_healthy.clone()))
                .await
                .unwrap();
        }

        let (tx, _cancel_tx) = spawn_test_pump(subs, test_workspace_roots());
        for i in 0..URI_COUNT {
            tx.send(publish(&format!("f{i}.rs"))).await.unwrap();
        }

        let mut healthy_received = HashSet::new();
        for _ in 0..URI_COUNT {
            healthy_received.insert(DiagnosticsResourceUri::for_test(
                &recv_within(&mut rx_healthy).await,
            ));
        }
        assert_eq!(healthy_received, expected);

        let mut stalled_received = HashSet::new();
        for _ in 0..URI_COUNT {
            stalled_received.insert(DiagnosticsResourceUri::for_test(
                &recv_within(&mut rx_stalled).await,
            ));
        }
        assert_eq!(stalled_received, expected);
    }

    type DuplexReader = tokio::io::BufReader<tokio::io::ReadHalf<tokio::io::DuplexStream>>;

    /// Reads newline-delimited JSON-RPC lines until one satisfies `matches`.
    async fn read_json_line_matching(
        reader: &mut DuplexReader,
        matches: impl Fn(&serde_json::Value) -> bool + Send + Sync,
    ) -> serde_json::Value {
        use tokio::io::AsyncBufReadExt as _;

        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            loop {
                let mut line = String::new();
                let n = reader.read_line(&mut line).await.unwrap();
                assert!(n > 0, "stream closed before the expected message");
                let value: serde_json::Value = serde_json::from_str(line.trim()).unwrap();
                if matches(&value) {
                    return value;
                }
            }
        })
        .await
        .expect("expected JSON-RPC message not received within 5 s")
    }

    /// #468/#492 stdio path: a real `Peer` over an in-memory duplex stream,
    /// driven with raw JSON-RPC, receives `resources/updated` from the real
    /// `diagnostics_pump` after a guarded subscribe.
    #[tokio::test]
    async fn test_stdio_shaped_peer_receives_resource_updates_through_pump() {
        use rmcp::ServiceExt as _;
        use tokio::io::AsyncWriteExt as _;

        let workspace = tempfile::TempDir::new().unwrap();
        let root = dunce::canonicalize(workspace.path()).unwrap();
        let file = root.join("main.rs");
        std::fs::write(&file, "fn main() {}").unwrap();
        let canonical_uri = bridge::resources::make_uri(&file).unwrap();

        let subs = make_subs();
        let server = mcp::McplsServer::new(
            Arc::new(Translator::new()),
            make_cache(),
            WorkspaceRoots::from_paths(std::slice::from_ref(&root)).unwrap(),
            subs.clone(),
            ProjectConfigStatus::NotIgnored,
            config::McpConfig::default(),
        );
        let (client_io, server_io) = tokio::io::duplex(64 * 1024);
        let serving = tokio::spawn(async move { server.serve(server_io).await.unwrap() });
        let (read_half, mut write_half) = tokio::io::split(client_io);
        let mut reader = tokio::io::BufReader::new(read_half);

        let initialize = r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"0"}}}"#;
        write_half
            .write_all(format!("{initialize}\n").as_bytes())
            .await
            .unwrap();
        read_json_line_matching(&mut reader, |v| v["id"] == 1).await;

        let initialized = r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#;
        let subscribe = serde_json::json!({
            "jsonrpc": "2.0", "id": 2, "method": "resources/subscribe",
            "params": {"uri": canonical_uri},
        });
        for line in [initialized.to_owned(), subscribe.to_string()] {
            write_half
                .write_all(format!("{line}\n").as_bytes())
                .await
                .unwrap();
        }
        // Keeps the service (and its peer) alive for the rest of the test.
        let _running = serving.await.unwrap();
        let response = read_json_line_matching(&mut reader, |v| v["id"] == 2).await;
        assert!(
            response.get("error").is_none(),
            "subscribe failed: {response}"
        );

        let (tx, _cancel_tx) = spawn_test_pump(subs, WorkspaceRoots::from_paths(&[root]).unwrap());
        tx.send(LspNotification::PublishDiagnostics(
            PublishDiagnosticsParams {
                uri: bridge::path_to_uri(&file).unwrap(),
                diagnostics: vec![],
                version: None,
            },
        ))
        .await
        .unwrap();

        let update = read_json_line_matching(&mut reader, |v| {
            v["method"] == "notifications/resources/updated"
        })
        .await;
        assert_eq!(update["params"]["uri"], canonical_uri);
    }

    /// A real `resources/subscribe` client over an in-memory duplex stream,
    /// against a server whose only language server (`rust`) is still
    /// starting; [`Self::settle_failed`] then records its startup failure.
    struct SubscribeHarness {
        translator: Arc<Translator>,
        subs: SubscriptionRegistry,
        uri: String,
        reader: DuplexReader,
        write_half: tokio::io::WriteHalf<tokio::io::DuplexStream>,
        next_id: u32,
        _running: rmcp::service::RunningService<rmcp::RoleServer, mcp::McplsServer>,
        _workspace: tempfile::TempDir,
    }

    impl SubscribeHarness {
        async fn start() -> Self {
            use rmcp::ServiceExt as _;

            let workspace = tempfile::TempDir::new().unwrap();
            let root = dunce::canonicalize(workspace.path()).unwrap();
            let file = root.join("main.rs");
            std::fs::write(&file, "fn main() {}").unwrap();
            let uri = bridge::resources::make_uri(&file).unwrap();

            let id = ServerId::from_static("rust");
            let mut translator = Translator::new()
                .with_extensions(crate::test_lsp::test_extensions())
                .with_router(config::ToolRouter::catch_all([(
                    id.clone(),
                    LanguageId::from_static("rust"),
                )]));
            translator.set_workspace_roots(
                WorkspaceRoots::from_paths(std::slice::from_ref(&root)).unwrap(),
            );
            translator.set_expected_servers(HashSet::from([id]));
            let translator = Arc::new(translator);

            let subs = make_subs();
            let server = mcp::McplsServer::new(
                Arc::clone(&translator),
                make_cache(),
                WorkspaceRoots::from_paths(&[root]).unwrap(),
                subs.clone(),
                ProjectConfigStatus::NotIgnored,
                config::McpConfig::default(),
            );
            let (client_io, server_io) = tokio::io::duplex(64 * 1024);
            let serving = tokio::spawn(async move { server.serve(server_io).await.unwrap() });
            let (read_half, mut write_half) = tokio::io::split(client_io);
            let mut reader = tokio::io::BufReader::new(read_half);

            let initialize = serde_json::json!({
                "jsonrpc": "2.0", "id": 0, "method": "initialize",
                "params": {
                    "protocolVersion": "2025-06-18",
                    "capabilities": {},
                    "clientInfo": {"name": "test", "version": "0"},
                },
            });
            let initialized =
                serde_json::json!({"jsonrpc": "2.0", "method": "notifications/initialized"});
            send_json_line(&mut write_half, &initialize).await;
            read_json_line_matching(&mut reader, |v| v["id"] == 0).await;
            send_json_line(&mut write_half, &initialized).await;
            let running = serving.await.unwrap();

            Self {
                translator,
                subs,
                uri,
                reader,
                write_half,
                next_id: 1,
                _running: running,
                _workspace: workspace,
            }
        }

        /// Sends `resources/subscribe` for the fixture file and returns the response.
        async fn subscribe(&mut self) -> serde_json::Value {
            let id = self.next_id;
            self.next_id += 1;
            let request = serde_json::json!({
                "jsonrpc": "2.0", "id": id, "method": "resources/subscribe",
                "params": {"uri": self.uri},
            });
            send_json_line(&mut self.write_half, &request).await;
            read_json_line_matching(&mut self.reader, |v| v["id"] == id).await
        }

        /// Records the `rust` server's startup failure the way `init_lsp_servers` does.
        fn settle_failed(&self) {
            self.translator
                .record_startup_failures(&[crate::error::ServerSpawnFailure {
                    server_id: ServerId::from_static("rust"),
                    language_id: LanguageId::from_static("rust"),
                    command: crate::config::ServerCommand::from_static("rust-analyzer"),
                    reason: crate::error::StartupFailure::Spawn(Arc::new(Error::ServerNotFound {
                        command: crate::config::ServerCommand::from_static("rust-analyzer"),
                        source: std::io::Error::from(std::io::ErrorKind::NotFound),
                    })),
                }]);
            self.translator.rebind_router(&HashSet::new());
            self.translator.clear_expected_servers();
        }

        /// Counts `resources/updated` notifications that arrive within `window`.
        async fn updates_within(&mut self, window: std::time::Duration) -> usize {
            use tokio::io::AsyncBufReadExt as _;

            let mut count = 0;
            let _ = tokio::time::timeout(window, async {
                loop {
                    let mut line = String::new();
                    if self.reader.read_line(&mut line).await.unwrap() == 0 {
                        return;
                    }
                    let value: serde_json::Value = serde_json::from_str(line.trim()).unwrap();
                    if value["method"] == "notifications/resources/updated" {
                        assert_eq!(value["params"]["uri"], self.uri);
                        count += 1;
                    }
                }
            })
            .await;
            count
        }
    }

    async fn send_json_line(
        write_half: &mut tokio::io::WriteHalf<tokio::io::DuplexStream>,
        value: &serde_json::Value,
    ) {
        use tokio::io::AsyncWriteExt as _;

        write_half
            .write_all(format!("{value}\n").as_bytes())
            .await
            .unwrap();
    }

    const UPDATE_WINDOW: std::time::Duration = std::time::Duration::from_millis(300);

    /// FR-010: a subscribe to a failed route errors and rolls back the
    /// subscription it added, so a later publish reaches nobody.
    #[tokio::test]
    async fn test_subscribe_to_failed_route_errors_and_rolls_back() {
        let mut harness = SubscribeHarness::start().await;
        harness.settle_failed();

        let response = harness.subscribe().await;

        assert!(response.get("error").is_some(), "{response}");
        assert!(response.to_string().contains("rust-analyzer"), "{response}");
        harness.subs.publish_matching(|_| true).await;
        assert_eq!(harness.updates_within(UPDATE_WINDOW).await, 0);
    }

    /// FR-010: a re-subscribe to a failed route errors but keeps the
    /// subscription made while the server was still starting.
    #[tokio::test]
    async fn test_resubscribe_to_failed_route_keeps_earlier_subscription() {
        let mut harness = SubscribeHarness::start().await;
        let first = harness.subscribe().await;
        assert!(first.get("error").is_none(), "{first}");
        harness.settle_failed();

        let second = harness.subscribe().await;

        assert!(second.get("error").is_some(), "{second}");
        harness.subs.publish_matching(|_| true).await;
        assert_eq!(harness.updates_within(UPDATE_WINDOW).await, 1);
    }

    /// FR-010/FR-012: subscribing while the server starts succeeds, and
    /// the settle publish then notifies exactly once for the failed URI.
    #[tokio::test]
    async fn test_subscribe_during_startup_gets_one_update_when_startup_fails() {
        let mut harness = SubscribeHarness::start().await;
        let response = harness.subscribe().await;
        assert!(response.get("error").is_none(), "{response}");
        assert_eq!(
            harness.updates_within(UPDATE_WINDOW).await,
            0,
            "nothing may publish before settle"
        );

        harness.settle_failed();
        publish_startup_failures(&harness.translator, &harness.subs).await;

        assert_eq!(harness.updates_within(UPDATE_WINDOW).await, 1);
    }
}

#[cfg(test)]
mod burst_tests {
    use std::time::Duration;

    use lsp_types::{Diagnostic, PublishDiagnosticsParams, Uri};

    use super::*;
    use crate::bridge::DiagnosticsAvailability;
    use crate::lsp::{MailboxLimits, NotificationSink};
    use crate::test_lsp::CapturedLogs;

    fn make_cache() -> Arc<Mutex<NotificationCache>> {
        Arc::new(Mutex::new(NotificationCache::new()))
    }

    #[cfg(windows)]
    fn roots_and_uri_prefix() -> (WorkspaceRoots, &'static str) {
        (
            WorkspaceRoots::for_test(vec![PathBuf::from(r"C:\test")], vec![]),
            "file:///C:/test/",
        )
    }
    #[cfg(not(windows))]
    fn roots_and_uri_prefix() -> (WorkspaceRoots, &'static str) {
        (
            WorkspaceRoots::for_test(vec![PathBuf::from("/test")], vec![]),
            "file:///test/",
        )
    }

    fn shared_over(
        cache: &Arc<Mutex<NotificationCache>>,
        workspace_roots: WorkspaceRoots,
    ) -> PumpShared {
        PumpShared {
            roles: DiagnosticsRoles::default(),
            notification_cache: Arc::clone(cache),
            subs: SubscriptionRegistry::new(),
            workspace_roots,
        }
    }

    /// #704: a burst bigger than the mailbox holds, with the pump not running
    /// while it arrives, leaves every file cached or marked lost -- none reads
    /// `pending` -- and the overflow writes at most one warning.
    #[tokio::test]
    async fn test_burst_over_the_mailbox_cap_caches_or_marks_every_file() {
        use tracing_subscriber::layer::SubscriberExt as _;

        const FILES: usize = 50;
        const CAP: usize = 10;
        let captured = CapturedLogs::default();
        let _guard =
            tracing::subscriber::set_default(tracing_subscriber::registry().with(captured.clone()));
        let (roots, prefix) = roots_and_uri_prefix();
        let uri = |n: usize| Uri::from(format!("{prefix}f{n}.rs"));
        let (messages_tx, messages_rx) = mpsc::channel(4);
        let (sink, publish_rx) = NotificationSink::with_limits(
            messages_tx,
            MailboxLimits {
                max_pending: CAP,
                max_bytes: usize::MAX,
            },
        );
        for n in 0..FILES {
            sink.publishes().publish(PublishDiagnosticsParams {
                uri: uri(n),
                diagnostics: vec![Diagnostic {
                    message: format!("error {n}").into(),
                    ..Diagnostic::default()
                }],
                version: None,
            });
        }
        drop(sink);
        let (lifecycle_tx, lifecycle_rx) = mpsc::channel(4);
        drop(lifecycle_tx);
        let cache = make_cache();

        let pump = tokio::spawn(diagnostics_pump(
            ServerId::from_static("rust"),
            NotificationInbox::new(messages_rx, publish_rx),
            lifecycle_rx,
            CancellationToken::new(),
            watch::channel(DiagnosticsRole::Authoritative).1,
            None,
            shared_over(&cache, roots),
        ));
        tokio::time::timeout(Duration::from_secs(5), pump)
            .await
            .expect("the pump must end once the mailbox is drained and closed")
            .unwrap();

        let cache = cache.lock().await;
        let states: Vec<DiagnosticsAvailability> = (0..FILES)
            .map(|n| cache.availability(&uri(n), Some(&ServerId::from_static("rust"))))
            .collect();
        assert!(
            !states.contains(&DiagnosticsAvailability::Pending),
            "{states:?}"
        );
        let published = states
            .iter()
            .filter(|state| **state == DiagnosticsAvailability::Published)
            .count();
        assert_eq!(published, CAP);
        assert_eq!(states.len() - published, FILES - CAP);
        assert!(
            states
                .iter()
                .skip(CAP)
                .all(|state| *state == DiagnosticsAvailability::Evicted)
        );
        let warnings = captured
            .entries()
            .iter()
            .filter(|(level, message)| {
                *level == tracing::Level::WARN && message.contains("mailbox is full")
            })
            .count();
        assert_eq!(warnings, 1);
    }

    /// #726: a lost mark is placed even when its path cannot be resolved
    /// (persistent transient filesystem error), so the file reads evicted
    /// instead of keeping stale content.
    #[tokio::test]
    async fn test_lost_mark_survives_a_transient_path_failure() {
        let workspace = tempfile::TempDir::new().unwrap();
        let root = dunce::canonicalize(workspace.path()).unwrap();
        let failing: Arc<bridge::CanonicalizeFn> =
            Arc::new(|_: &std::path::Path| Err(std::io::Error::from(std::io::ErrorKind::TimedOut)));
        let published = bridge::path_to_uri(&root.join("kept.rs")).unwrap();
        let lost = bridge::path_to_uri(&root.join("lost.rs")).unwrap();
        let (messages_tx, messages_rx) = mpsc::channel(4);
        let (sink, publish_rx) = NotificationSink::with_limits(
            messages_tx,
            MailboxLimits {
                max_pending: 1,
                max_bytes: usize::MAX,
            },
        );
        for uri in [&published, &lost] {
            sink.publishes().publish(PublishDiagnosticsParams {
                uri: uri.clone(),
                diagnostics: vec![Diagnostic {
                    message: "error".to_owned().into(),
                    ..Diagnostic::default()
                }],
                version: None,
            });
        }
        drop(sink);
        let (lifecycle_tx, lifecycle_rx) = mpsc::channel(1);
        drop(lifecycle_tx);
        let cache = make_cache();

        let pump = tokio::spawn(diagnostics_pump_with_resolver(
            ServerId::from_static("rust"),
            NotificationInbox::new(messages_rx, publish_rx),
            lifecycle_rx,
            CancellationToken::new(),
            watch::channel(DiagnosticsRole::Authoritative).1,
            None,
            shared_over(
                &cache,
                WorkspaceRoots::from_paths(std::slice::from_ref(&root)).unwrap(),
            ),
            PublishedPathResolver::with_canonicalizer(failing),
        ));
        tokio::time::timeout(Duration::from_secs(60), pump)
            .await
            .expect("the pump must end once the mailbox is drained and closed")
            .unwrap();

        assert_eq!(
            cache
                .lock()
                .await
                .availability(&lost, Some(&ServerId::from_static("rust"))),
            DiagnosticsAvailability::Evicted
        );
    }

    /// #696: a pump started for a respawned server shares the role sender, so
    /// a role recomputed afterwards reaches it.
    #[test]
    fn test_replacement_pump_receives_later_role_changes() {
        let roles = DiagnosticsRoles::default();
        let id = ServerId::from_static("rust");
        let original = roles.subscribe(&id, DiagnosticsRole::Secondary);
        let replacement = roles.subscribe(&id, DiagnosticsRole::Secondary);

        roles.set(&id, DiagnosticsRole::Authoritative);

        assert_eq!(*original.borrow(), DiagnosticsRole::Authoritative);
        assert_eq!(*replacement.borrow(), DiagnosticsRole::Authoritative);
    }

    /// The pump ends when the mailbox and both channels are closed.
    #[tokio::test]
    async fn test_pump_ends_once_every_source_closed() {
        let (messages_tx, messages_rx) = mpsc::channel(1);
        let (sink, publish_rx) = NotificationSink::new(messages_tx);
        let (lifecycle_tx, lifecycle_rx) = mpsc::channel(1);
        let (roots, _) = roots_and_uri_prefix();
        let cache = make_cache();
        let pump = tokio::spawn(diagnostics_pump(
            ServerId::from_static("rust"),
            NotificationInbox::new(messages_rx, publish_rx),
            lifecycle_rx,
            CancellationToken::new(),
            watch::channel(DiagnosticsRole::Authoritative).1,
            None,
            shared_over(&cache, roots),
        ));

        drop(sink);
        drop(lifecycle_tx);

        tokio::time::timeout(Duration::from_secs(5), pump)
            .await
            .expect("the pump must end")
            .unwrap();
    }
}