Merge remote-tracking branch 'origin/main' into feat/remote-development

# Conflicts:
#	crates/ide-jsonrpc/src/lib.rs

.github/workflows/ci.yml

@@ -7,7 +7,8 @@ on:
 
 concurrency:
   group: ci-${{ github.ref }}
-  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+  # A newer push supersedes the run in progress, on main too.
+  cancel-in-progress: true
 
 env:
   CARGO_TERM_COLOR: always
@@ -29,7 +30,10 @@ jobs:
         with: { node-version: 22, cache: pnpm }
       - uses: dtolnay/rust-toolchain@stable
         with: { components: "clippy, rustfmt" }
+      # Only main saves the cache (saving takes minutes on Windows); PRs restore main's.
       - uses: Swatinem/rust-cache@v2
+        with:
+          save-if: ${{ github.ref == 'refs/heads/main' }}
       - uses: taiki-e/install-action@nextest
       - name: Linux deps
         if: runner.os == 'Linux'
@@ -45,9 +49,10 @@ jobs:
       # Test fixtures write LF files; keep Git from converting them on checkout.
       - run: git config --global core.autocrlf false
       - run: cargo nextest run --workspace --no-fail-fast
+      # nextest ran export_bindings, which rewrites src/bindings.ts.
       - name: Bindings are up to date
         if: runner.os == 'Linux'
-        run: pnpm gen:bindings && git diff --exit-code src/bindings.ts
+        run: git diff --exit-code src/bindings.ts
 
   # The remote development server ships as a static Linux binary (release.yml):
   # keep that build working.
@@ -78,8 +83,22 @@ jobs:
       - run: pnpm lint
       - run: pnpm test
       - run: pnpm build
-      - run: pnpm exec playwright install --with-deps chromium
+
+  e2e:
+    name: E2E
+    runs-on: ubuntu-24.04
+    # Browsers and their system libraries are preinstalled; keep the tag in step with @playwright/test.
+    container: mcr.microsoft.com/playwright:v1.64.1-noble
+    steps:
+      - uses: actions/checkout@v4
+      - uses: pnpm/action-setup@v4
+        with: { version: 11 }
+      - uses: actions/setup-node@v4
+        with: { node-version: 22, cache: pnpm }
+      - run: pnpm install --frozen-lockfile
       - run: pnpm e2e
+        env:
+          HOME: /root
       - uses: actions/upload-artifact@v4
         if: failure()
         with:

crates/ide-agent/src/lib.rs

@@ -191,6 +191,8 @@ struct Task {
     dirty: Mutex<BTreeSet<usize>>,
     flush_scheduled: AtomicBool,
     stderr: Mutex<VecDeque<String>>,
+    /// Set by `delete`; held while saving, so a late save can't bring the file back.
+    deleted: Mutex<bool>,
 }
 
 impl AgentManager {
@@ -519,6 +521,7 @@ impl AgentManager {
             blocking(move || worktree::remove(&root, &cwd)).await?;
         }
         self.tasks.write().remove(id);
+        *task.deleted.lock() = true;
         let _ = std::fs::remove_file(self.shared.paths.store_dir.join(format!("{id}.json")));
         self.shared.emit(AgentEvent::Removed { id: id.to_owned() });
         Ok(())
@@ -568,6 +571,7 @@ impl Task {
             dirty: Mutex::new(BTreeSet::new()),
             flush_scheduled: AtomicBool::new(false),
             stderr: Mutex::new(VecDeque::new()),
+            deleted: Mutex::new(false),
         })
     }
 
@@ -604,6 +608,10 @@ impl Task {
             })
         };
         let path = self.shared.paths.store_dir.join(format!("{}.json", self.id));
+        let deleted = self.deleted.lock();
+        if *deleted {
+            return;
+        }
         if let Err(e) = bytes.map_err(|e| e.to_string()).and_then(|b| write_file(&path, &b)) {
             tracing::warn!("saving agent task {}: {e}", self.id);
         }
@@ -934,15 +942,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,10 @@ 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. So `on_notification`
+/// must not wait for the remote side (a response to our request may depend on
+/// its request to us, which would then never start); hand such work to a task.
 pub trait Handler: Send + Sync + 'static {
     fn on_notification(&self, method: &str, params: Value);
     /// Respond to a server→client request (e.g. `workspace/configuration`).
@@ -221,14 +224,6 @@ impl Handler for NoopHandler {
 
 type Pending = HashMap<i64, oneshot::Sender<Result<Value, RpcError>>>;
 
-/// What the delivery thread hands on, in arrival order: a response is not seen
-/// before the notifications the peer sent ahead of it (a streamed result's last
-/// values before its request completes).
-enum Incoming {
-    Notification(String, Value),
-    Response(oneshot::Sender<Result<Value, RpcError>>, Result<Value, RpcError>),
-}
-
 struct Inner {
     out: mpsc::UnboundedSender<Vec<u8>>,
     pending: Mutex<Pending>,
@@ -506,27 +501,35 @@ impl Peer {
             });
         }
 
-        // Notifications (and the responses after them) are delivered in order on a dedicated thread.
+        // Notifications are delivered in order on a dedicated thread. Requests start from the same
+        // queue, so a request handler sees every notification that arrived before it, and responses
+        // complete from it, so a caller sees the notifications sent before its response (a streamed
+        // result's last values).
         let (notif_tx, notif_rx) = std::sync::mpsc::channel::<Incoming>();
         {
             let handler = handler.clone();
+            let inner = inner.clone();
+            let runtime = tokio::runtime::Handle::current();
             std::thread::Builder::new()
                 .name("jsonrpc-notify".into())
                 .spawn(move || {
-                    while let Ok(incoming) = notif_rx.recv() {
-                        let (method, params) = match incoming {
-                            Incoming::Notification(method, params) => (method, params),
+                    while let Ok(msg) = notif_rx.recv() {
+                        match msg {
+                            Incoming::Notification(method, params) => {
+                                let h = handler.clone();
+                                let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                                    h.on_notification(&method, params)
+                                }));
+                                if result.is_err() {
+                                    tracing::error!("jsonrpc notification handler panicked on {method}");
+                                }
+                            }
+                            Incoming::Request(id, method, params) => {
+                                runtime.spawn(handle_request(inner.clone(), handler.clone(), id, method, params));
+                            }
                             Incoming::Response(tx, result) => {
                                 let _ = tx.send(result);
-                                continue;
                             }
-                        };
-                        let h = handler.clone();
-                        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
-                            h.on_notification(&method, params)
-                        }));
-                        if result.is_err() {
-                            tracing::error!("jsonrpc notification handler panicked on {method}");
                         }
                     }
                 })
@@ -558,7 +561,7 @@ impl Peer {
                     match msg {
                         Ok(Some(body)) => {
                             seen_message.store(true, Ordering::Relaxed);
-                            dispatch(&inner, &handler, &notif_tx, &body);
+                            dispatch(&inner, &notif_tx, &body);
                         }
                         Ok(None) => break,
                         Err(e) => {
@@ -695,7 +698,29 @@ fn parse_id(id: &Value) -> Option<i64> {
     }
 }
 
-fn dispatch(inner: &Arc<Inner>, handler: &Arc<dyn Handler>, notif_tx: &std::sync::mpsc::Sender<Incoming>, body: &[u8]) {
+enum Incoming {
+    Notification(String, Value),
+    Request(Value, String, Value),
+    Response(oneshot::Sender<Result<Value, RpcError>>, Result<Value, RpcError>),
+}
+
+async fn handle_request(inner: Arc<Inner>, handler: Arc<dyn Handler>, id: Value, method: String, params: Value) {
+    let m = method.clone();
+    let result = tokio::task::spawn_blocking(move || handler.on_request(&m, params)).await;
+    let reply = match result {
+        Ok(Ok(result)) => json!({ "jsonrpc": "2.0", "id": id, "result": result }),
+        Ok(Err((code, message))) => {
+            json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })
+        }
+        Err(_) => json!({
+            "jsonrpc": "2.0", "id": id,
+            "error": { "code": codes::INTERNAL_ERROR, "message": format!("handler for {method} panicked") }
+        }),
+    };
+    let _ = inner.send_value(&reply);
+}
+
+fn dispatch(inner: &Arc<Inner>, notif_tx: &std::sync::mpsc::Sender<Incoming>, 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,
@@ -714,23 +739,7 @@ fn dispatch(inner: &Arc<Inner>, handler: &Arc<dyn Handler>, notif_tx: &std::sync
             let params = obj.remove("params").unwrap_or(Value::Null);
             match id {
                 Some(id) if !id.is_null() => {
-                    let handler = handler.clone();
-                    let inner = inner.clone();
-                    tokio::spawn(async move {
-                        let m = method.clone();
-                        let result = tokio::task::spawn_blocking(move || handler.on_request(&m, params)).await;
-                        let reply = match result {
-                            Ok(Ok(result)) => json!({ "jsonrpc": "2.0", "id": id, "result": result }),
-                            Ok(Err((code, message))) => {
-                                json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })
-                            }
-                            Err(_) => json!({
-                                "jsonrpc": "2.0", "id": id,
-                                "error": { "code": codes::INTERNAL_ERROR, "message": format!("handler for {method} panicked") }
-                            }),
-                        };
-                        let _ = inner.send_value(&reply);
-                    });
+                    let _ = notif_tx.send(Incoming::Request(id, method, params));
                 }
                 _ => {
                     let _ = notif_tx.send(Incoming::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");