Client request sender and inbound response handling for ruff server (#10620)

## Summary

Fixes #10618.

This PR introduces a proper API for sending requests to the client and
handling any response sent back. Dynamic capability registration now
uses this new API, fixing an issue where a much more simplistic response
handler silently flushes a code action request that needed a response.

## Test Plan

#10618 can no longer be reproduced. No errors about unhandled responses
should appear in the extension output, and you should see this new log
when the server starts:
```
<DATE> <TIME> [info] <DURATION> INFO ruff_server::server Configuration file watcher successfully registered
```

crates/ruff_server/src/server.rs

@@ -1,9 +1,6 @@
 //! Scheduling, I/O, and API endpoints.
 
 use std::num::NonZeroUsize;
-use std::sync::atomic::AtomicI32;
-use std::sync::atomic::Ordering;
-use std::time::Duration;
 
 use lsp::Connection;
 use lsp_server as lsp;
@@ -22,6 +19,8 @@ use types::WorkDoneProgressOptions;
 use types::WorkspaceFoldersServerCapabilities;
 
 use self::schedule::event_loop_thread;
+use self::schedule::Scheduler;
+use self::schedule::Task;
 use crate::session::Session;
 use crate::PositionEncoding;
 
@@ -33,10 +32,10 @@ pub(crate) type Result<T> = std::result::Result<T, api::Error>;
 
 pub struct Server {
     conn: lsp::Connection,
+    client_capabilities: ClientCapabilities,
     threads: lsp::IoThreads,
     worker_threads: NonZeroUsize,
     session: Session,
-    next_request_id: AtomicI32,
 }
 
 impl Server {
@@ -50,12 +49,6 @@ impl Server {
         let client_capabilities = init_params.capabilities;
         let server_capabilities = Self::server_capabilities(&client_capabilities);
 
-        let dynamic_registration = client_capabilities
-            .workspace
-            .and_then(|workspace| workspace.did_change_watched_files)
-            .and_then(|watched_files| watched_files.dynamic_registration)
-            .unwrap_or_default();
-
         let workspaces = init_params
             .workspace_folders
             .map(|folders| folders.into_iter().map(|folder| folder.uri).collect())
@@ -76,56 +69,14 @@ impl Server {
             }
         });
 
-        let next_request_id = AtomicI32::from(1);
-
         conn.initialize_finish(id, initialize_data)?;
 
-        if dynamic_registration {
-            // Register capabilities
-            conn.sender
-                .send(lsp_server::Message::Request(lsp_server::Request {
-                    id: next_request_id.fetch_add(1, Ordering::Relaxed).into(),
-                    method: "client/registerCapability".into(),
-                    params: serde_json::to_value(lsp_types::RegistrationParams {
-                        registrations: vec![lsp_types::Registration {
-                            id: "ruff-server-watch".into(),
-                            method: "workspace/didChangeWatchedFiles".into(),
-                            register_options: Some(serde_json::to_value(
-                                DidChangeWatchedFilesRegistrationOptions {
-                                    watchers: vec![
-                                        FileSystemWatcher {
-                                            glob_pattern: types::GlobPattern::String(
-                                                "**/.?ruff.toml".into(),
-                                            ),
-                                            kind: None,
-                                        },
-                                        FileSystemWatcher {
-                                            glob_pattern: types::GlobPattern::String(
-                                                "**/pyproject.toml".into(),
-                                            ),
-                                            kind: None,
-                                        },
-                                    ],
-                                },
-                            )?),
-                        }],
-                    })?,
-                }))?;
-
-            // Flush response from the client (to avoid an unexpected response appearing in the event loop)
-            let _ = conn.receiver.recv_timeout(Duration::from_secs(5)).map_err(|_| {
-                tracing::error!("Timed out while waiting for client to acknowledge registration of dynamic capabilities");
-            });
-        } else {
-            tracing::warn!("LSP client does not support dynamic file watcher registration - automatic configuration reloading will not be available.");
-        }
-
         Ok(Self {
             conn,
+            client_capabilities,
             threads,
             worker_threads,
             session: Session::new(&server_capabilities, &workspaces)?,
-            next_request_id,
         })
     }
 
@@ -133,9 +84,9 @@ impl Server {
         let result = event_loop_thread(move || {
             Self::event_loop(
                 &self.conn,
+                &self.client_capabilities,
                 self.session,
                 self.worker_threads,
-                self.next_request_id,
             )
         })?
         .join();
@@ -146,11 +97,14 @@ impl Server {
     #[allow(clippy::needless_pass_by_value)] // this is because we aren't using `next_request_id` yet.
     fn event_loop(
         connection: &Connection,
-        session: Session,
+        client_capabilities: &ClientCapabilities,
+        mut session: Session,
         worker_threads: NonZeroUsize,
-        _next_request_id: AtomicI32,
     ) -> crate::Result<()> {
-        let mut scheduler = schedule::Scheduler::new(session, worker_threads, &connection.sender);
+        let mut scheduler =
+            schedule::Scheduler::new(&mut session, worker_threads, &connection.sender);
+
+        Self::try_register_capabilities(client_capabilities, &mut scheduler);
         for msg in &connection.receiver {
             let task = match msg {
                 lsp::Message::Request(req) => {
@@ -160,18 +114,69 @@ impl Server {
                     api::request(req)
                 }
                 lsp::Message::Notification(notification) => api::notification(notification),
-                lsp::Message::Response(response) => {
-                    tracing::error!(
-                        "Expected request or notification, got response instead: {response:?}"
-                    );
-                    continue;
-                }
+                lsp::Message::Response(response) => scheduler.response(response),
             };
             scheduler.dispatch(task);
         }
         Ok(())
     }
 
+    fn try_register_capabilities(
+        client_capabilities: &ClientCapabilities,
+        scheduler: &mut Scheduler,
+    ) {
+        let dynamic_registration = client_capabilities
+            .workspace
+            .as_ref()
+            .and_then(|workspace| workspace.did_change_watched_files)
+            .and_then(|watched_files| watched_files.dynamic_registration)
+            .unwrap_or_default();
+        if dynamic_registration {
+            // Register all dynamic capabilities here
+
+            // `workspace/didChangeWatchedFiles`
+            // (this registers the configuration file watcher)
+            let params = lsp_types::RegistrationParams {
+                registrations: vec![lsp_types::Registration {
+                    id: "ruff-server-watch".into(),
+                    method: "workspace/didChangeWatchedFiles".into(),
+                    register_options: Some(
+                        serde_json::to_value(DidChangeWatchedFilesRegistrationOptions {
+                            watchers: vec![
+                                FileSystemWatcher {
+                                    glob_pattern: types::GlobPattern::String(
+                                        "**/.?ruff.toml".into(),
+                                    ),
+                                    kind: None,
+                                },
+                                FileSystemWatcher {
+                                    glob_pattern: types::GlobPattern::String(
+                                        "**/pyproject.toml".into(),
+                                    ),
+                                    kind: None,
+                                },
+                            ],
+                        })
+                        .unwrap(),
+                    ),
+                }],
+            };
+
+            let response_handler = |()| {
+                tracing::info!("Configuration file watcher successfully registered");
+                Task::nothing()
+            };
+
+            if let Err(err) = scheduler
+                .request::<lsp_types::request::RegisterCapability>(params, response_handler)
+            {
+                tracing::error!("An error occurred when trying to register the configuration file watcher: {err}");
+            }
+        } else {
+            tracing::warn!("LSP client does not support dynamic capability registration - automatic configuration reloading will not be available.");
+        }
+    }
+
     fn server_capabilities(client_capabilities: &ClientCapabilities) -> types::ServerCapabilities {
         let position_encoding = client_capabilities
             .general