fix(lsp): Start JSON-RPC requests after the notifications before them

ide-jsonrpc handled incoming notifications in order on a dedicated thread
but started each incoming request on its own blocking task right away, so
a request could be handled before a notification that arrived ahead of
it. The mock language server in the ide-lsp tests answered
`textDocument/references` before applying the `didOpen` of a file opened
for a project-wide search (missing results, a search finishing before the
test could cancel it or open the file), and on the client side an ACP
`session/request_permission` could overtake the `session/update` that
announced its tool call (ide-agent polled for up to 200 ms to cover it).

A request now starts once every notification received before it has been
handled; requests still run concurrently. The ide-agent polling is gone.

Tests no longer depend on timing:
- the references tests hold the mock's answer with a gate and cancel or
  open the file while it is held, instead of sleeping and measuring;
- the indexing retry ends indexing once the first answer is out;
- the crash/restart test counts Ready/Crashed statuses instead of reading
  the last one, which a restart can already have replaced;
- the mock records a notification under the same lock as its effect, and
  `eventually` waits up to 30 s;
- the TCP tests hold their port bound (not listening) instead of freeing
  it for another test to take, and listen only after the command started.

Claude-Session: https://claude.ai/code/session_01VUYsdXA2nkF5v6zsC93dpn

crates/ide-agent/src/lib.rs

@@ -923,15 +923,6 @@ impl Task {
                     .collect()
             })
             .unwrap_or_default();
-        // The tool call it is about arrives as a notification on another thread: let it land first.
-        if let Some(tool) = params.pointer("/toolCall/toolCallId").and_then(Value::as_str) {
-            for _ in 0..20 {
-                if self.state.lock().timeline.has_tool(tool) {
-                    break;
-                }
-                std::thread::sleep(Duration::from_millis(10));
-            }
-        }
         let (tx, rx) = std::sync::mpsc::channel();
         let entry = self.push_entry(AgentEntry::Permission { id, title, options: options.clone(), chosen: None });
         self.permissions.lock().insert(id, tx);

crates/ide-agent/src/timeline.rs

@@ -37,10 +37,6 @@ impl Timeline {
         t
     }
 
-    pub fn has_tool(&self, id: &str) -> bool {
-        self.tools.contains_key(id)
-    }
-
     pub fn push(&mut self, entry: AgentEntry) -> usize {
         self.entries.push(entry);
         self.entries.len() - 1

crates/ide-jsonrpc/src/lib.rs

@@ -202,7 +202,8 @@ pub mod codes {
 ///
 /// Calls run on blocking-capable threads, never on the reader task, so a slow
 /// handler cannot stall response delivery. Notifications are delivered one at
-/// a time in arrival order; requests run concurrently.
+/// a time in arrival order; requests run concurrently, each once every
+/// notification that arrived before it has been handled.
 pub trait Handler: Send + Sync + 'static {
     fn on_notification(&self, method: &str, params: Value);
     /// Respond to a server→client request (e.g. `workspace/configuration`).
@@ -219,6 +220,13 @@ impl Handler for NoopHandler {
     }
 }
 
+/// Work for the in-order notification thread.
+enum Inbound {
+    Notification(String, Value),
+    /// A request may start: the notifications before it are handled.
+    Release(oneshot::Sender<()>),
+}
+
 type Pending = HashMap<i64, oneshot::Sender<Result<Value, RpcError>>>;
 
 struct Inner {
@@ -499,13 +507,20 @@ impl Peer {
         }
 
         // Notifications are delivered in order on a dedicated thread.
-        let (notif_tx, notif_rx) = std::sync::mpsc::channel::<(String, Value)>();
+        let (notif_tx, notif_rx) = std::sync::mpsc::channel::<Inbound>();
         {
             let handler = handler.clone();
             std::thread::Builder::new()
                 .name("jsonrpc-notify".into())
                 .spawn(move || {
-                    while let Ok((method, params)) = notif_rx.recv() {
+                    while let Ok(inbound) = notif_rx.recv() {
+                        let (method, params) = match inbound {
+                            Inbound::Notification(method, params) => (method, params),
+                            Inbound::Release(start) => {
+                                let _ = start.send(());
+                                continue;
+                            }
+                        };
                         let h = handler.clone();
                         let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                             h.on_notification(&method, params)
@@ -680,12 +695,7 @@ fn parse_id(id: &Value) -> Option<i64> {
     }
 }
 
-fn dispatch(
-    inner: &Arc<Inner>,
-    handler: &Arc<dyn Handler>,
-    notif_tx: &std::sync::mpsc::Sender<(String, Value)>,
-    body: &[u8],
-) {
+fn dispatch(inner: &Arc<Inner>, handler: &Arc<dyn Handler>, notif_tx: &std::sync::mpsc::Sender<Inbound>, body: &[u8]) {
     tracing::trace!(target: "ide_jsonrpc::wire", "<-- {}", String::from_utf8_lossy(body));
     let msg: Value = match serde_json::from_slice(body) {
         Ok(v) => v,
@@ -706,7 +716,11 @@ fn dispatch(
                 Some(id) if !id.is_null() => {
                     let handler = handler.clone();
                     let inner = inner.clone();
+                    let (start, started) = oneshot::channel();
+                    let _ = notif_tx.send(Inbound::Release(start));
                     tokio::spawn(async move {
+                        // E.g. a `didOpen` before a request must be applied when the request is answered.
+                        let _ = started.await;
                         let m = method.clone();
                         let result = tokio::task::spawn_blocking(move || handler.on_request(&m, params)).await;
                         let reply = match result {
@@ -723,7 +737,7 @@ fn dispatch(
                     });
                 }
                 _ => {
-                    let _ = notif_tx.send((method, params));
+                    let _ = notif_tx.send(Inbound::Notification(method, params));
                 }
             }
         }

crates/ide-jsonrpc/src/tests.rs

@@ -251,6 +251,27 @@ async fn server_to_client_request() {
     assert!(matches!(err, RpcError::Server { code: codes::METHOD_NOT_FOUND, .. }));
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn requests_start_after_the_notifications_before_them() {
+    // `seen` answers with the notifications handled so far; handling one is slow.
+    struct Server(Recorder);
+    impl Handler for Server {
+        fn on_notification(&self, method: &str, params: Value) {
+            std::thread::sleep(Duration::from_millis(20));
+            self.0.on_notification(method, params)
+        }
+        fn on_request(&self, _: &str, _: Value) -> Result<Value, (i64, String)> {
+            Ok(json!(self.0.notifications.lock().len()))
+        }
+    }
+    let (client, _server) = pair(Arc::new(NoopHandler), Arc::new(Server(Recorder::default())));
+    for n in 1..=5 {
+        client.notify("open", n).unwrap();
+        let seen: usize = client.request("seen", (), None).await.unwrap();
+        assert_eq!(seen, n);
+    }
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn slow_handler_does_not_block_responses() {
     // The client handles a slow server→client request while its own request completes.

crates/ide-lsp/tests/mock_server.rs

@@ -72,6 +72,8 @@ struct MockHandler(Arc<MockState>);
 impl Handler for MockHandler {
     fn on_notification(&self, method: &str, params: Value) {
         let s = &self.0;
+        // Held throughout: a test that sees the effect (a document, the peer gone) also sees the record.
+        let mut log = s.notifications.lock();
         let uri = params["textDocument"]["uri"].as_str().unwrap_or_default().to_owned();
         match method {
             "textDocument/didOpen" => {
@@ -110,7 +112,7 @@ impl Handler for MockHandler {
             }
             _ => {}
         }
-        s.notifications.lock().push((method.to_owned(), params));
+        log.push((method.to_owned(), params));
     }
 
     fn on_request(&self, method: &str, params: Value) -> Result<Value, (i64, String)> {
@@ -276,14 +278,37 @@ impl Harness {
     }
 }
 
+/// Waits for `f`; the bound only keeps a broken test from hanging (machines under load are slow).
 async fn eventually(mut f: impl FnMut() -> bool) {
-    for _ in 0..500 {
-        if f() {
-            return;
-        }
+    let deadline = std::time::Instant::now() + Duration::from_secs(30);
+    while !f() {
+        assert!(std::time::Instant::now() < deadline, "condition not reached in time");
         tokio::time::sleep(Duration::from_millis(10)).await;
     }
-    panic!("condition not reached in time");
+}
+
+/// Holds mock request handlers until the test opens it.
+#[derive(Default)]
+struct Gate {
+    entered: std::sync::atomic::AtomicBool,
+    open: std::sync::atomic::AtomicBool,
+}
+
+impl Gate {
+    /// Called from a handler: blocks until [`Gate::open`] (or a generous bound).
+    fn hold(&self) {
+        self.entered.store(true, Ordering::SeqCst);
+        let deadline = std::time::Instant::now() + Duration::from_secs(30);
+        while !self.open.load(Ordering::SeqCst) && std::time::Instant::now() < deadline {
+            std::thread::sleep(Duration::from_millis(1));
+        }
+    }
+    fn entered(&self) -> bool {
+        self.entered.load(Ordering::SeqCst)
+    }
+    fn open(&self) {
+        self.open.store(true, Ordering::SeqCst);
+    }
 }
 
 fn uri_of(path: &Path) -> String {
@@ -313,7 +338,7 @@ async fn initialize_and_incremental_sync() {
     assert_eq!(init["capabilities"]["textDocument"]["completion"]["completionItem"]["snippetSupport"], true);
     assert_eq!(init["initializationOptions"], json!({ "flag": true, "nested": { "deep": 1 } }));
     assert_eq!(init["rootUri"], json!(uri_of(&h.root)));
-    assert_eq!(s.notified("initialized").len(), 1);
+    eventually(|| s.notified("initialized").len() == 1).await;
     assert_eq!(h.statuses("mock"), vec![ServerStatus::Starting, ServerStatus::Ready]);
 
     let uri = uri_of(&path);
@@ -707,11 +732,13 @@ async fn gives_up_after_repeated_crashes_and_manual_restart_works() {
     let h = harness(&[("mock", incremental_caps())]);
     h.open("a.mk", "x");
     let s = h.launcher.state("mock");
+    let count = |status| h.statuses("mock").iter().filter(|s| **s == status).count();
     for n in 1..=4 {
-        eventually(|| s.launches.load(Ordering::SeqCst) == n && s.peer.lock().is_some()).await;
-        h.wait_status("mock", ServerStatus::Ready).await;
+        // Counted, not the last status: a restart may already have replaced it.
+        eventually(|| count(ServerStatus::Ready) == n).await;
+        assert_eq!(s.launches.load(Ordering::SeqCst), n);
         s.peer().kill();
-        eventually(|| h.statuses("mock").last() == Some(&ServerStatus::Crashed)).await;
+        eventually(|| count(ServerStatus::Crashed) == n).await;
     }
     tokio::time::sleep(Duration::from_millis(300)).await;
     assert_eq!(s.launches.load(Ordering::SeqCst), 4, "initial start + 3 restarts");
@@ -1526,23 +1553,27 @@ async fn cancelling_a_references_search_stops_the_candidate_search_and_closes_fi
     grep_candidates(&h);
     let slow = uri_of(&b);
     let inner = open_files_only(vec![]);
+    let gate = Arc::new(Gate::default());
+    let held = gate.clone();
     *s.responder.lock() = Some(Arc::new(move |s, method, params| {
         if s.docs.lock().contains_key(&slow) {
-            std::thread::sleep(Duration::from_millis(600));
+            held.hold();
         }
         inner(s, method, params)
     }));
     let cancel = ide_core::CancellationToken::new();
-    let later = cancel.clone();
-    tokio::spawn(async move {
-        tokio::time::sleep(Duration::from_millis(200)).await;
-        later.cancel();
+    let mgr = h.mgr.clone();
+    let search = tokio::spawn({
+        let cancel = cancel.clone();
+        async move { mgr.references(doc, Pos::new(0, 5), true, true, cancel).await }
     });
-    let started = std::time::Instant::now();
-    let err = h.mgr.references(doc, Pos::new(0, 5), true, true, cancel).await.unwrap_err();
+    eventually(|| gate.entered()).await;
+    cancel.cancel();
+    // Returns while the server still holds the answer.
+    let err = search.await.unwrap().unwrap_err();
     assert_eq!(err.kind, ide_core::error::ErrorKind::Cancelled);
-    assert!(started.elapsed() < Duration::from_millis(550));
     eventually(|| uris(&s.notified("textDocument/didClose")) == vec![uri_of(&b)]).await;
+    gate.open();
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -1586,9 +1617,11 @@ async fn opening_a_file_during_a_search_takes_it_over() {
     let (doc, _) = h.open("a.mk", "def target\n");
     let slow = uri_of(&b);
     let inner = open_files_only(vec![]);
+    let gate = Arc::new(Gate::default());
+    let held = gate.clone();
     *s.responder.lock() = Some(Arc::new(move |s, method, params| {
         if s.docs.lock().get(&slow).is_some_and(|t| t.starts_with("call")) {
-            std::thread::sleep(Duration::from_millis(400));
+            held.hold();
         }
         inner(s, method, params)
     }));
@@ -1597,9 +1630,12 @@ async fn opening_a_file_during_a_search_takes_it_over() {
 
     let mgr = h.mgr.clone();
     let search = tokio::spawn(async move { mgr.references(doc, Pos::new(0, 5), true, true, Default::default()).await });
-    eventually(|| s.docs.lock().contains_key(&uri_of(&b))).await;
+    // The server answers with the copy opened for the search.
+    eventually(|| gate.entered()).await;
     let user_doc = DocId::next();
     h.mgr.did_open(user_doc, &b, &h.lang, 7, "edited target\n");
+    eventually(|| s.docs.lock().get(&uri_of(&b)).map(String::as_str) == Some("edited target\n")).await;
+    gate.open();
     search.await.unwrap().unwrap();
 
     // The temporary copy was closed before the user's didOpen, and the user's stays open.
@@ -1633,10 +1669,15 @@ async fn an_empty_answer_while_indexing_is_retried_when_indexing_ends() {
     let peer = s.peer();
     peer.notify("$/progress", json!({ "token": "idx", "value": { "kind": "begin", "title": "Indexing" } })).unwrap();
     h.wait_status("mock", ServerStatus::Busy).await;
-    let (state, uri) = (s.clone(), uri_of(&path));
+    // Nothing found while indexing; indexing ends once that answer is out.
+    s.respond("textDocument/references", Value::Array(occurrences(&uri_of(&path), "def target\n")));
+    let answered = Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let first = answered.clone();
+    *s.responder.lock() = Some(Arc::new(move |_, method, _| {
+        (method == "textDocument/references" && !first.swap(true, Ordering::SeqCst)).then(|| json!([]))
+    }));
     tokio::spawn(async move {
-        tokio::time::sleep(Duration::from_millis(200)).await;
-        state.respond("textDocument/references", Value::Array(occurrences(&uri, "def target\n")));
+        eventually(|| answered.load(Ordering::SeqCst)).await;
         peer.notify("$/progress", json!({ "token": "idx", "value": { "kind": "end" } })).unwrap();
     });
     let refs = h.mgr.references(doc, Pos::new(0, 5), true, true, Default::default()).await.unwrap();

crates/ide-lsp/tests/tcp_server.rs

@@ -115,6 +115,14 @@ impl Harness {
     }
 }
 
+/// A loopback port where nothing listens, held (bound, not listening) by the
+/// returned socket so that no other test is handed it meanwhile.
+fn idle_port() -> (u16, tokio::net::TcpSocket) {
+    let socket = tokio::net::TcpSocket::new_v4().unwrap();
+    socket.bind(([127, 0, 0, 1], 0).into()).unwrap();
+    (socket.local_addr().unwrap().port(), socket)
+}
+
 async fn eventually(mut f: impl FnMut() -> bool) {
     for _ in 0..1000 {
         if f() {
@@ -157,7 +165,7 @@ async fn connects_to_a_running_server_requests_and_reconnects() {
 
 #[tokio::test(flavor = "multi_thread")]
 async fn nothing_listening_and_no_executable_reports_how_to_fix_it() {
-    let port = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap().local_addr().unwrap().port();
+    let (port, _reserved) = idle_port();
     let h = harness(godot_cfg(port, "openidea-no-such-godot"));
     h.open("player.gd", "extends Node\n");
     h.wait_status(ServerStatus::Missing, 1).await;
@@ -170,16 +178,18 @@ async fn nothing_listening_and_no_executable_reports_how_to_fix_it() {
 #[cfg(unix)]
 #[tokio::test(flavor = "multi_thread")]
 async fn starts_the_command_and_waits_until_it_listens() {
-    let port = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap().local_addr().unwrap().port();
-    let cfg = LanguageServerConfig { args: vec!["30".into()], ..godot_cfg(port, "sleep") };
+    let (port, socket) = idle_port();
+    let args = ["-c", "touch started; exec sleep 30"].map(String::from).to_vec();
+    let cfg = LanguageServerConfig { args, ..godot_cfg(port, "sh") };
     let root = tempfile::tempdir().unwrap();
-    // The "server" (`sleep`) never listens itself; a listener comes up a moment later.
+    // The "server" never listens itself; a listener comes up once it has been started.
     let log = Arc::new(ServerLog::default());
     let server = {
         let log = log.clone();
+        let started = root.path().join("started");
         tokio::spawn(async move {
-            tokio::time::sleep(Duration::from_millis(400)).await;
-            let listener = tokio::net::TcpListener::bind(("127.0.0.1", port)).await.unwrap();
+            eventually(|| started.exists()).await;
+            let listener = socket.listen(16).unwrap();
             let (stream, _) = listener.accept().await.unwrap();
             let (r, w) = stream.into_split();
             log.peers.lock().push(Peer::connect(r, w, Arc::new(MockHandler(log.clone()))));
@@ -205,7 +215,7 @@ async fn starts_the_command_and_waits_until_it_listens() {
 #[cfg(unix)]
 #[tokio::test(flavor = "multi_thread")]
 async fn a_command_that_exits_without_listening_fails_and_cancel_stops_waiting() {
-    let port = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap().local_addr().unwrap().port();
+    let (port, _reserved) = idle_port();
     let root = tempfile::tempdir().unwrap();
     let path = root.path().to_path_buf();
     let cfg = godot_cfg(port, "false");