fix(lsp): pull diagnostics from servers that provide them (#1307)

rust-analyzer reports its own analysis (a missing field, a wrong type) only
to clients that pull diagnostics; it pushes just `cargo check`'s, so in the
editor a Rust type error showed up only after saving. The client now
advertises `textDocument/diagnostic` and pulls an open file's diagnostics
from servers with a `diagnosticProvider`: when it opens or its server
starts, after edits (once typing pauses), on save (and other open files
for servers with inter-file dependencies) and on
`workspace/diagnostic/refresh`. Pulled and pushed diagnostics are kept
apart and shown, and given to code actions, together; a newer pull
supersedes an older reply, a server-cancelled pull is retried, and closing
a file drops what was pulled.

crates/ide-lsp/src/caps.rs

@@ -48,7 +48,7 @@ pub fn client_capabilities() -> Value {
             "semanticTokens": { "refreshSupport": true },
             "codeLens": { "refreshSupport": true },
             "inlayHint": { "refreshSupport": true },
-            "diagnostics": { "refreshSupport": false },
+            "diagnostics": { "refreshSupport": true },
             "textDocumentContent": { "dynamicRegistration": true }
         },
         "textDocument": {
@@ -130,6 +130,7 @@ pub fn client_capabilities() -> Value {
                 "prepareSupportDefaultBehavior": 1,
                 "honorsChangeAnnotations": false
             },
+            "diagnostic": { "dynamicRegistration": false, "relatedDocumentSupport": false },
             "publishDiagnostics": {
                 "relatedInformation": true,
                 "tagSupport": { "valueSet": [1, 2] },

crates/ide-lsp/src/handler.rs

@@ -130,15 +130,8 @@ impl ClientHandler {
         {
             return;
         }
-        let conv = mgr.conv(self.encoding());
-        let items = params
-            .diagnostics
-            .iter()
-            .cloned()
-            .map(|d| convert::diagnostic(&conv, &path, d, &self.server.name))
-            .collect();
-        self.server.set_diagnostics(path.clone(), params.diagnostics);
-        mgr.sink().emit(IdeEvent::Diagnostics { path: path.into(), source: self.server.name.clone(), items });
+        let shown = self.server.set_diagnostics(path.clone(), params.diagnostics);
+        mgr.emit_diagnostics(&self.server, path, shown, self.encoding());
     }
 
     fn show_message(&self, mgr: &LspManager, params: &Value) {
@@ -309,7 +302,11 @@ impl Handler for ClientHandler {
                 mgr.sink().emit(IdeEvent::LspRefresh { server: self.server.id, kind });
                 Ok(Value::Null)
             }
-            "workspace/diagnostic/refresh" | "workspace/inlineValue/refresh" => Ok(Value::Null),
+            "workspace/diagnostic/refresh" => {
+                mgr.refresh_pulled_diagnostics(&self.server);
+                Ok(Value::Null)
+            }
+            "workspace/inlineValue/refresh" => Ok(Value::Null),
             other => Err((codes::METHOD_NOT_FOUND, format!("unsupported method {other}"))),
         }
     }

crates/ide-lsp/src/inspect.rs

@@ -6,8 +6,6 @@
 //! (`textDocument/diagnostic`). The others are waited on: the first
 //! `publishDiagnostics` for the file after it was opened, then a short quiet
 //! period so follow-up publishes (syntax first, semantics later) are caught.
-//! The client never advertises pull support, so servers keep pushing for the
-//! editor as before.
 
 use crate::convert;
 use crate::server::Server;

crates/ide-lsp/src/lib.rs

@@ -18,6 +18,7 @@ mod inspect;
 pub mod jdtls;
 mod launch;
 mod library_source;
+mod pull;
 mod scope;
 mod server;
 mod tcp;
@@ -432,6 +433,7 @@ impl LspManager {
         let wanted: Vec<(DocId, Vec<Arc<Server>>)> =
             open.into_iter().map(|(id, language, path)| (id, self.servers_for(&language, &path))).collect();
         let mut start = Vec::new();
+        let mut opened = Vec::new();
         {
             let mut docs = self.docs.lock();
             for (id, servers) in wanted {
@@ -446,6 +448,7 @@ impl LspManager {
                             user_did_open(server, &run, &d.path, &d.uri, d.did_open_params());
                         }
                         attached.opened_gen = Some(run.generation);
+                        opened.push((id, server.clone()));
                     }
                     d.servers.push(attached);
                     if !start.iter().any(|s: &Arc<Server>| s.id == server.id) {
@@ -456,6 +459,9 @@ impl LspManager {
                 d.servers.sort_by_key(|a| servers.iter().position(|s| s.id == a.server.id).unwrap_or(usize::MAX));
             }
         }
+        for (doc, server) in opened {
+            self.schedule_pull(doc, Some(&server), Duration::ZERO);
+        }
         for server in &start {
             self.ensure_started(server);
         }
@@ -611,6 +617,7 @@ impl LspManager {
         self.refresh_external_project(&server, &run).await;
 
         // Publish readiness and (re)open attached documents atomically w.r.t. did_change.
+        let mut reopened = Vec::new();
         {
             let mut docs = self.docs.lock();
             if !server.mark_ready(generation, run.clone()) {
@@ -620,16 +627,20 @@ impl LspManager {
                 return;
             }
             let open_close = run.caps.read().open_close;
-            for d in docs.values_mut() {
+            for (id, d) in docs.iter_mut() {
                 let params = d.did_open_params();
                 for a in d.servers.iter_mut().filter(|a| a.server.id == server.id && a.opened_gen != Some(generation)) {
                     if open_close && let Err(e) = peer.notify("textDocument/didOpen", params.clone()) {
                         tracing::debug!("{}: didOpen: {e}", server.name);
                     }
                     a.opened_gen = Some(generation);
+                    reopened.push(*id);
                 }
             }
         }
+        for doc in reopened {
+            self.schedule_pull(doc, Some(&server), Duration::ZERO);
+        }
         self.check_clang_tidy(&server).await;
         server.refresh_busy_status(&self.sink);
         server.announce_ready(generation);
@@ -853,6 +864,7 @@ impl LspManager {
             }
             docs.insert(doc, state);
         }
+        self.schedule_pull(doc, None, Duration::ZERO);
         for s in &servers {
             self.ensure_started(s);
             if s.running().is_some()
@@ -930,6 +942,8 @@ impl LspManager {
                 tracing::debug!("{}: didChange: {e}", server.name);
             }
         }
+        drop(docs);
+        self.schedule_pull(doc, None, pull::AFTER_EDIT);
     }
 
     pub fn did_save(&self, doc: DocId, text: &str) {
@@ -952,11 +966,15 @@ impl LspManager {
                 tracing::debug!("{}: didSave: {e}", a.server.name);
             }
         }
+        drop(docs);
+        self.schedule_pull(doc, None, Duration::ZERO);
+        self.pull_dependents(doc);
     }
 
     pub fn did_close(&self, doc: DocId) {
         let Some(d) = self.docs.lock().remove(&doc) else { return };
         for a in &d.servers {
+            self.forget_pulled(&a.server, &d.path, a.server.running().as_ref());
             let Some(run) = a.server.running() else { continue };
             if a.opened_gen == Some(run.generation)
                 && run.caps.read().open_close

crates/ide-lsp/src/pull.rs

@@ -0,0 +1,192 @@
+//! Pull diagnostics (`textDocument/diagnostic`). Servers that provide them may stop pushing
+//! their own analysis to a client that can pull (rust-analyzer pushes only `cargo check`
+//! then): the editor pulls an open file's diagnostics when it opens, after edits and saves,
+//! and when the server asks for a refresh. Pulled diagnostics are kept apart from pushed ones
+//! and shown together.
+
+use crate::server::{Running, Server};
+use crate::{LspManager, convert};
+use ide_core::{DocId, IdeEvent};
+use ide_jsonrpc::RpcError;
+use lsp_types as lsp;
+use serde_json::{Value, json};
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Duration;
+
+/// How long typing must pause before a changed file is pulled again.
+pub(crate) const AFTER_EDIT: Duration = Duration::from_millis(300);
+const TIMEOUT: Duration = Duration::from_secs(30);
+/// Retries of a pull the server cancelled (`ServerCancelled`, e.g. while indexing).
+const RETRIES: u32 = 20;
+
+struct PullTarget {
+    server: Arc<Server>,
+    run: Running,
+    path: PathBuf,
+    uri: lsp::Uri,
+}
+
+fn provides_pull(run: &Running) -> bool {
+    let caps = run.caps.read();
+    caps.restriction.diagnostics() && caps.has("diagnosticProvider")
+}
+
+fn inter_file(run: &Running) -> bool {
+    run.caps.read().raw.pointer("/diagnosticProvider/interFileDependencies").and_then(Value::as_bool) == Some(true)
+}
+
+impl LspManager {
+    /// Pulls `doc`'s diagnostics from its servers that provide them (`only`: from that one)
+    /// after `delay`; a later pull of the same file supersedes it.
+    pub(crate) fn schedule_pull(&self, doc: DocId, only: Option<&Server>, delay: Duration) {
+        let targets: Vec<PullTarget> = {
+            let docs = self.docs.lock();
+            let Some(d) = docs.get(&doc) else { return };
+            d.servers
+                .iter()
+                .filter(|a| only.is_none_or(|s| s.id == a.server.id))
+                .filter_map(|a| {
+                    let run = a.server.running()?;
+                    (a.opened_gen == Some(run.generation) && provides_pull(&run)).then(|| PullTarget {
+                        server: a.server.clone(),
+                        run,
+                        path: d.path.clone(),
+                        uri: d.uri.clone(),
+                    })
+                })
+                .collect()
+        };
+        for t in targets {
+            if t.server.suppresses_diagnostics(&t.path) {
+                continue;
+            }
+            let pull = t.server.next_pull(&t.path);
+            let me = self.me.clone();
+            self.runtime.spawn(async move {
+                tokio::time::sleep(delay).await;
+                if let Some(me) = me.upgrade() {
+                    me.pull(t, pull).await;
+                }
+            });
+        }
+    }
+
+    /// Pulls every open file of `server` again (`workspace/diagnostic/refresh`).
+    pub(crate) fn refresh_pulled_diagnostics(&self, server: &Server) {
+        let docs: Vec<DocId> = self
+            .docs
+            .lock()
+            .iter()
+            .filter(|(_, d)| d.servers.iter().any(|a| a.server.id == server.id))
+            .map(|(id, _)| *id)
+            .collect();
+        for doc in docs {
+            self.schedule_pull(doc, Some(server), Duration::ZERO);
+        }
+    }
+
+    /// After `doc` was saved: servers whose diagnostics depend on other files pull those
+    /// open files too.
+    pub(crate) fn pull_dependents(&self, doc: DocId) {
+        let docs = self.docs.lock();
+        let Some(d) = docs.get(&doc) else { return };
+        let servers: Vec<Arc<Server>> = d
+            .servers
+            .iter()
+            .filter(|a| a.server.running().is_some_and(|run| provides_pull(&run) && inter_file(&run)))
+            .map(|a| a.server.clone())
+            .collect();
+        let others: Vec<(DocId, Vec<Arc<Server>>)> = docs
+            .iter()
+            .filter(|(id, _)| **id != doc)
+            .map(|(id, o)| {
+                (*id, servers.iter().filter(|s| o.servers.iter().any(|a| a.server.id == s.id)).cloned().collect())
+            })
+            .collect();
+        drop(docs);
+        for (other, servers) in others {
+            for server in servers {
+                self.schedule_pull(other, Some(&server), Duration::ZERO);
+            }
+        }
+    }
+
+    /// Drops what was pulled for a closed file and shows what remains.
+    pub(crate) fn forget_pulled(&self, server: &Server, path: &std::path::Path, run: Option<&Running>) {
+        if let Some(shown) = server.clear_pulled(path) {
+            let enc = run.map(|r| r.caps.read().encoding).unwrap_or_default();
+            self.emit_diagnostics(server, path.to_owned(), shown, enc);
+        }
+    }
+
+    async fn pull(&self, t: PullTarget, pull: u64) {
+        let mut attempt = 0;
+        loop {
+            if !t.server.is_latest_pull(&t.path, pull) {
+                return;
+            }
+            let mut params = json!({ "textDocument": { "uri": t.uri } });
+            if let Some(id) = t.server.pulled_result_id(&t.path) {
+                params["previousResultId"] = json!(id);
+            }
+            if let Some(identifier) = t.run.caps.read().raw.pointer("/diagnosticProvider/identifier") {
+                params["identifier"] = identifier.clone();
+            }
+            let request = t.run.peer.request::<Value, Value>("textDocument/diagnostic", params, None);
+            let report = match tokio::time::timeout(TIMEOUT, request).await {
+                Ok(Ok(report)) => report,
+                // This request has no cancellation of its own: the server cancelled it
+                // (`ServerCancelled` while busy, which asks to be asked again).
+                Ok(Err(RpcError::Cancelled)) if attempt < RETRIES => {
+                    attempt += 1;
+                    tokio::time::sleep(Duration::from_millis(250 * u64::from(attempt.min(8)))).await;
+                    continue;
+                }
+                Ok(Err(e)) => {
+                    tracing::debug!("{}: textDocument/diagnostic: {e}", t.server.name);
+                    return;
+                }
+                Err(_) => {
+                    tracing::debug!("{}: textDocument/diagnostic timed out", t.server.name);
+                    return;
+                }
+            };
+            // A newer pull, a closed file or a restarted server makes this report stale.
+            if !t.server.is_latest_pull(&t.path, pull)
+                || t.server.running().is_none_or(|r| r.generation != t.run.generation)
+                || !self.docs.lock().values().any(|d| d.path == t.path)
+            {
+                return;
+            }
+            let result_id = report.get("resultId").and_then(Value::as_str).map(str::to_owned);
+            let items = match report.get("kind").and_then(Value::as_str) {
+                Some("unchanged") => None,
+                _ => match serde_json::from_value(report.get("items").cloned().unwrap_or(json!([]))) {
+                    Ok(items) => Some(items),
+                    Err(e) => {
+                        tracing::debug!("{}: bad textDocument/diagnostic report: {e}", t.server.name);
+                        return;
+                    }
+                },
+            };
+            let shown = t.server.set_pulled(t.path.clone(), result_id, items);
+            let enc = t.run.caps.read().encoding;
+            self.emit_diagnostics(&t.server, t.path, shown, enc);
+            return;
+        }
+    }
+
+    /// Shows `items` as `server`'s diagnostics for `path`.
+    pub(crate) fn emit_diagnostics(
+        &self,
+        server: &Server,
+        path: PathBuf,
+        items: Vec<lsp::Diagnostic>,
+        enc: crate::text::Encoding,
+    ) {
+        let conv = self.conv(enc);
+        let items = items.into_iter().map(|d| convert::diagnostic(&conv, &path, d, &server.name)).collect();
+        self.sink.emit(IdeEvent::Diagnostics { path: path.into(), source: server.name.clone(), items });
+    }
+}

crates/ide-lsp/src/server.rs

@@ -97,7 +97,13 @@ pub struct Server {
     started: Mutex<Option<Instant>>,
     stderr: Mutex<VecDeque<String>>,
     pub progress: Mutex<HashMap<String, Progress>>,
+    /// What the editor shows per path: the pushed and the pulled diagnostics together.
     diagnostics: Mutex<HashMap<PathBuf, Vec<lsp::Diagnostic>>>,
+    /// `publishDiagnostics` per path.
+    pushed: Mutex<HashMap<PathBuf, Vec<lsp::Diagnostic>>>,
+    /// `textDocument/diagnostic` reports per path, and the latest pull asked for each.
+    pulled: Mutex<HashMap<PathBuf, Pulled>>,
+    pulls: Mutex<HashMap<PathBuf, u64>>,
     /// When diagnostics were last published per path (empty lists included).
     published: Mutex<HashMap<PathBuf, Instant>>,
     /// Woken on every `publishDiagnostics` (Inspect Code waits on it).
@@ -132,6 +138,9 @@ impl Server {
             stderr: Mutex::new(VecDeque::new()),
             progress: Mutex::new(HashMap::new()),
             diagnostics: Mutex::new(HashMap::new()),
+            pushed: Mutex::new(HashMap::new()),
+            pulled: Mutex::new(HashMap::new()),
+            pulls: Mutex::new(HashMap::new()),
             published: Mutex::new(HashMap::new()),
             published_notify: Notify::new(),
             idle_notify: Notify::new(),
@@ -493,17 +502,71 @@ impl Server {
         self.progress.lock().drain().map(|(_, p)| (p.task, p.title)).collect()
     }
 
-    pub fn set_diagnostics(&self, path: PathBuf, items: Vec<lsp::Diagnostic>) {
+    /// Records the diagnostics the server pushed for `path`; returns what the file now shows.
+    pub fn set_diagnostics(&self, path: PathBuf, items: Vec<lsp::Diagnostic>) -> Vec<lsp::Diagnostic> {
+        set_or_remove(&mut self.pushed.lock(), path.clone(), items);
+        self.combine(path)
+    }
+
+    /// Records a pulled report for `path` (`None` items: unchanged since `result_id`'s
+    /// predecessor); returns what the file now shows.
+    pub fn set_pulled(
+        &self,
+        path: PathBuf,
+        result_id: Option<String>,
+        items: Option<Vec<lsp::Diagnostic>>,
+    ) -> Vec<lsp::Diagnostic> {
+        {
+            let mut pulled = self.pulled.lock();
+            let entry = pulled.entry(path.clone()).or_default();
+            entry.result_id = result_id;
+            if let Some(items) = items {
+                entry.items = items;
+            }
+        }
+        self.combine(path)
+    }
+
+    /// Forgets the pulled report of a closed file; `None` when there was none.
+    pub fn clear_pulled(&self, path: &Path) -> Option<Vec<lsp::Diagnostic>> {
+        self.pulls.lock().remove(path);
+        self.pulled.lock().remove(path)?;
+        Some(self.combine(path.to_owned()))
+    }
+
+    /// The `resultId` of the last report pulled for `path`.
+    pub fn pulled_result_id(&self, path: &Path) -> Option<String> {
+        self.pulled.lock().get(path).and_then(|p| p.result_id.clone())
+    }
+
+    /// Starts a pull for `path`, superseding earlier ones; see [`Server::is_latest_pull`].
+    pub fn next_pull(&self, path: &Path) -> u64 {
+        static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
+        let id = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        self.pulls.lock().insert(path.to_owned(), id);
+        id
+    }
+
+    pub fn is_latest_pull(&self, path: &Path, id: u64) -> bool {
+        self.pulls.lock().get(path) == Some(&id)
+    }
+
+    fn combine(&self, path: PathBuf) -> Vec<lsp::Diagnostic> {
+        let mut items = self.pushed.lock().get(&path).cloned().unwrap_or_default();
+        if let Some(pulled) = self.pulled.lock().get(&path) {
+            for d in &pulled.items {
+                if !items.contains(d) {
+                    items.push(d.clone());
+                }
+            }
+        }
         {
             let mut d = self.diagnostics.lock();
             self.published.lock().insert(path.clone(), Instant::now());
-            if items.is_empty() {
-                d.remove(&path);
-            } else {
-                d.insert(path, items);
-            }
+            set_or_remove(&mut d, path, items.clone());
         }
         self.published_notify.notify_waiters();
+        items
     }
 
     /// When diagnostics for `path` were last published by this server.
@@ -528,10 +591,28 @@ impl Server {
 
     pub fn take_diagnostic_paths(&self) -> Vec<PathBuf> {
         self.published.lock().clear();
+        self.pushed.lock().clear();
+        self.pulled.lock().clear();
+        self.pulls.lock().clear();
         self.diagnostics.lock().drain().map(|(p, _)| p).collect()
     }
 }
 
+/// A file's last `textDocument/diagnostic` report.
+#[derive(Default)]
+struct Pulled {
+    result_id: Option<String>,
+    items: Vec<lsp::Diagnostic>,
+}
+
+fn set_or_remove(map: &mut HashMap<PathBuf, Vec<lsp::Diagnostic>>, path: PathBuf, items: Vec<lsp::Diagnostic>) {
+    if items.is_empty() {
+        map.remove(&path);
+    } else {
+        map.insert(path, items);
+    }
+}
+
 /// The last `n` stderr lines, led by the exception headline: Node.js / Python /
 /// JVM crashes print the error first and a stack after it, so a plain tail
 /// shows frames without saying what went wrong.

crates/ide-lsp/tests/mock_server.rs

@@ -768,6 +768,113 @@ async fn diagnostics_become_events_and_feed_code_actions() {
     assert_eq!(err.kind, ide_core::error::ErrorKind::Unavailable);
 }
 
+/// The diagnostics shown last for `path` by `source`, as their messages.
+fn shown(h: &Harness, path: &Path, source: &str) -> Option<Vec<String>> {
+    h.events.lock().iter().rev().find_map(|e| match e {
+        IdeEvent::Diagnostics { path: p, source: s, items } if **p == *path && s == source => {
+            Some(items.iter().map(|d| d.message.clone()).collect())
+        }
+        _ => None,
+    })
+}
+
+/// Reports a "bad" diagnostic per line containing `bad`, as `textDocument/diagnostic`; answers
+/// `unchanged` when the text is what `previousResultId` was computed from.
+fn pull_responder() -> Responder {
+    Arc::new(|s: &MockState, method: &str, params: &Value| {
+        if method != "textDocument/diagnostic" {
+            return None;
+        }
+        let text = s.docs.lock().get(params["textDocument"]["uri"].as_str()?).cloned()?;
+        let id = format!("{}", text.len() * 31 + text.matches("bad").count());
+        if params["previousResultId"].as_str() == Some(id.as_str()) {
+            return Some(json!({ "kind": "unchanged", "resultId": id }));
+        }
+        let items: Vec<Value> = text
+            .lines()
+            .enumerate()
+            .filter(|(_, l)| l.contains("bad"))
+            .map(|(i, _)| {
+                json!({ "range": {"start":{"line":i,"character":0},"end":{"line":i,"character":3}},
+                        "severity": 1, "message": "bad" })
+            })
+            .collect();
+        Some(json!({ "kind": "full", "resultId": id, "items": items }))
+    })
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn pulled_diagnostics_follow_edits_and_join_pushed_ones() {
+    let mut pull_caps = incremental_caps();
+    pull_caps["diagnosticProvider"] = json!({ "interFileDependencies": false, "workspaceDiagnostics": false });
+    let h = harness(&[("mock", pull_caps), ("push", incremental_caps())]);
+    let s = h.launcher.state("mock");
+    *s.responder.lock() = Some(pull_responder());
+    let (doc, path) = h.open("a.mk", "ok\nbad\n");
+    h.wait_status("mock", ServerStatus::Ready).await;
+    h.wait_status("push", ServerStatus::Ready).await;
+    assert!(s.requested("initialize")[0]["capabilities"]["textDocument"]["diagnostic"].is_object());
+    eventually(|| shown(&h, &path, "mock") == Some(vec!["bad".into()])).await;
+    assert!(h.launcher.state("push").requested("textDocument/diagnostic").is_empty(), "it pulls from providers only");
+
+    // What the server still pushes (rust-analyzer: `cargo check`) shows next to what was pulled.
+    let pushed = json!({ "range": {"start":{"line":0,"character":0},"end":{"line":0,"character":2}},
+                         "severity": 2, "message": "pushed" });
+    s.peer()
+        .notify("textDocument/publishDiagnostics", json!({ "uri": uri_of(&path), "diagnostics": [pushed] }))
+        .unwrap();
+    eventually(|| shown(&h, &path, "mock") == Some(vec!["pushed".into(), "bad".into()])).await;
+
+    // Edits are pulled again once typing pauses.
+    h.mgr.did_change(doc, 2, &[change(rng(0, 0, 0, 0), "bad\n")], &|| unreachable!());
+    eventually(|| shown(&h, &path, "mock") == Some(vec!["pushed".into(), "bad".into(), "bad".into()])).await;
+    // Pulled diagnostics are code actions' context like pushed ones.
+    let _ = h.mgr.code_actions(doc, rng(2, 1, 2, 1), vec![]).await;
+    let ctx = &s.requested("textDocument/codeAction")[0]["context"]["diagnostics"];
+    assert_eq!(ctx.as_array().unwrap().iter().map(|d| d["message"].as_str().unwrap()).collect::<Vec<_>>(), ["bad"]);
+
+    // Saving unchanged text asks with the last result id; `unchanged` keeps what is shown.
+    let pulls = s.requested("textDocument/diagnostic").len();
+    h.mgr.did_save(doc, "bad\nok\nbad\n");
+    eventually(|| s.requested("textDocument/diagnostic").len() > pulls).await;
+    let last = s.requested("textDocument/diagnostic").last().cloned().unwrap();
+    assert!(last["previousResultId"].is_string());
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    assert_eq!(shown(&h, &path, "mock"), Some(vec!["pushed".into(), "bad".into(), "bad".into()]));
+
+    // Closing drops what was pulled; the pushed diagnostics stay the server's to clear.
+    h.mgr.did_close(doc);
+    eventually(|| shown(&h, &path, "mock") == Some(vec!["pushed".into()])).await;
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn a_cancelled_pull_is_retried_and_a_refresh_pulls_again() {
+    let mut caps = incremental_caps();
+    caps["diagnosticProvider"] = json!({ "interFileDependencies": false, "workspaceDiagnostics": false });
+    let h = harness(&[("mock", caps)]);
+    let s = h.launcher.state("mock");
+    *s.responder.lock() = Some(pull_responder());
+    // Busy indexing: the server cancels and asks to be asked again.
+    s.errors.lock().insert("textDocument/diagnostic".into(), (-32802, "indexing".into()));
+    let (_doc, path) = h.open("a.mk", "bad\n");
+    h.wait_status("mock", ServerStatus::Ready).await;
+    eventually(|| s.requested("textDocument/diagnostic").len() >= 2).await;
+    assert_eq!(shown(&h, &path, "mock"), None);
+    s.errors.lock().clear();
+    eventually(|| shown(&h, &path, "mock") == Some(vec!["bad".into()])).await;
+
+    // `workspace/diagnostic/refresh` (e.g. after a project reload) pulls open files again.
+    *s.responder.lock() = Some(Arc::new(|_: &MockState, method: &str, _: &Value| {
+        (method == "textDocument/diagnostic").then(|| {
+            json!({ "kind": "full", "items": [
+            { "range": {"start":{"line":0,"character":0},"end":{"line":0,"character":3}}, "message": "reloaded" } ] })
+        })
+    }));
+    let refreshed: Value = s.peer().request("workspace/diagnostic/refresh", Value::Null, None).await.unwrap();
+    assert_eq!(refreshed, Value::Null);
+    eventually(|| shown(&h, &path, "mock") == Some(vec!["reloaded".into()])).await;
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn apply_edit_hook_and_server_requests() {
     let h = harness(&[("mock", incremental_caps())]);