Extract a scheduler crate from GPUI to enable unified integration testing of client and server code (#37326)

Extracts and cleans up GPUI's scheduler code into a new `scheduler`
crate, making it pluggable by external runtimes. This will enable
deterministic integration testing with cloud components by providing a
unified test scheduler across Zed and backend code. In Zed, it will
replace the existing GPUI scheduler for consistent async task management
across platforms.

## Changes

- **Core Implementation**: `TestScheduler` with seed-based
randomization, session tracking (`SessionId`), and foreground/background
task separation for reproducible testing.
- **Executors**: `ForegroundExecutor` (!Send, thread-local) and
`BackgroundExecutor` (Send, with blocking/timeout support) as
GPUI-compatible wrappers.
- **Clock and Timer**: Controllable `TestClock` and future-based `Timer`
for time-sensitive tests.
- **Testing APIs**: `once()`, `with_seed()`, and `many()` methods for
configurable test runs.
- **Dependencies**: Added `async-task`, `chrono`, `futures`, etc., with
updates to `Cargo.toml` and lock file.

## Benefits

- **Integration Testing**: Facilitates reliable async tests involving
cloud sessions, reducing flakiness via deterministic execution.
- **Pluggability**: Trait-based design (`Scheduler`) allows easy
integration into non-GPUI runtimes while maintaining GPUI compatibility.
- **Cleanup**: Refactors GPUI scheduler logic for clarity, correctness
(no `unwrap()`, proper error handling), and extensibility.

Follows Rust guidelines; run `./script/clippy` for verification.

- [x] Define and test a core scheduler that we think can power our cloud
code and GPUI
- [ ] Replace GPUI's scheduler


Release Notes:

- N/A

---------

Co-authored-by: Antonio Scandurra <me@as-cii.com>

crates/scheduler/src/executor.rs

@@ -0,0 +1,137 @@
+use crate::{Scheduler, SessionId, Timer};
+use std::{
+    future::Future,
+    marker::PhantomData,
+    pin::Pin,
+    rc::Rc,
+    sync::Arc,
+    task::{Context, Poll},
+    time::Duration,
+};
+
+#[derive(Clone)]
+pub struct ForegroundExecutor {
+    session_id: SessionId,
+    scheduler: Arc<dyn Scheduler>,
+    not_send: PhantomData<Rc<()>>,
+}
+
+impl ForegroundExecutor {
+    pub fn spawn<F>(&self, future: F) -> Task<F::Output>
+    where
+        F: Future + 'static,
+        F::Output: 'static,
+    {
+        let session_id = self.session_id;
+        let scheduler = Arc::clone(&self.scheduler);
+        let (runnable, task) = async_task::spawn_local(future, move |runnable| {
+            scheduler.schedule_foreground(session_id, runnable);
+        });
+        runnable.schedule();
+        Task(TaskState::Spawned(task))
+    }
+
+    pub fn timer(&self, duration: Duration) -> Timer {
+        self.scheduler.timer(duration)
+    }
+}
+
+impl ForegroundExecutor {
+    pub fn new(session_id: SessionId, scheduler: Arc<dyn Scheduler>) -> Self {
+        assert!(
+            scheduler.is_main_thread(),
+            "ForegroundExecutor must be created on the same thread as the Scheduler"
+        );
+        Self {
+            session_id,
+            scheduler,
+            not_send: PhantomData,
+        }
+    }
+}
+
+impl BackgroundExecutor {
+    pub fn new(scheduler: Arc<dyn Scheduler>) -> Self {
+        Self { scheduler }
+    }
+}
+
+pub struct BackgroundExecutor {
+    scheduler: Arc<dyn Scheduler>,
+}
+
+impl BackgroundExecutor {
+    pub fn spawn<F>(&self, future: F) -> Task<F::Output>
+    where
+        F: Future + Send + 'static,
+        F::Output: Send + 'static,
+    {
+        let scheduler = Arc::clone(&self.scheduler);
+        let (runnable, task) = async_task::spawn(future, move |runnable| {
+            scheduler.schedule_background(runnable);
+        });
+        runnable.schedule();
+        Task(TaskState::Spawned(task))
+    }
+
+    pub fn block_on<Fut: Future>(&self, future: Fut) -> Fut::Output {
+        self.scheduler.block_on(future)
+    }
+
+    pub fn block_with_timeout<Fut: Unpin + Future>(
+        &self,
+        future: &mut Fut,
+        timeout: Duration,
+    ) -> Option<Fut::Output> {
+        self.scheduler.block_with_timeout(future, timeout)
+    }
+
+    pub fn timer(&self, duration: Duration) -> Timer {
+        self.scheduler.timer(duration)
+    }
+}
+
+/// Task is a primitive that allows work to happen in the background.
+///
+/// It implements [`Future`] so you can `.await` on it.
+///
+/// If you drop a task it will be cancelled immediately. Calling [`Task::detach`] allows
+/// the task to continue running, but with no way to return a value.
+#[must_use]
+#[derive(Debug)]
+pub struct Task<T>(TaskState<T>);
+
+#[derive(Debug)]
+enum TaskState<T> {
+    /// A task that is ready to return a value
+    Ready(Option<T>),
+
+    /// A task that is currently running.
+    Spawned(async_task::Task<T>),
+}
+
+impl<T> Task<T> {
+    /// Creates a new task that will resolve with the value
+    pub fn ready(val: T) -> Self {
+        Task(TaskState::Ready(Some(val)))
+    }
+
+    /// Detaching a task runs it to completion in the background
+    pub fn detach(self) {
+        match self {
+            Task(TaskState::Ready(_)) => {}
+            Task(TaskState::Spawned(task)) => task.detach(),
+        }
+    }
+}
+
+impl<T> Future for Task<T> {
+    type Output = T;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+        match unsafe { self.get_unchecked_mut() } {
+            Task(TaskState::Ready(val)) => Poll::Ready(val.take().unwrap()),
+            Task(TaskState::Spawned(task)) => Pin::new(task).poll(cx),
+        }
+    }
+}