Add WaiterGuard RAII for Session::send_and_wait (cancel-safety)

Closes RFD-400 review finding #2 (idle_waiter slot leak on external
cancellation). See cancel-safety review session db4b1ac8-... for the
full report.

`Session::send_and_wait` installs an `IdleWaiter` slot under
`self.idle_waiter`, then awaits the response. Internal failure paths
(send_inner error, oneshot closure, internal timeout) clean up the
slot. External cancellation — caller wraps `send_and_wait` in
`tokio::time::timeout`, races it in `select!`, or aborts the
JoinHandle — does not. The cleanup at the event loop's `else` branch
only fires when the channel itself closes (i.e., the entire session
is going away), not when an individual call is cancelled.

Effect: any external cancellation between "install waiter" and
"drain response" leaves `idle_waiter = Some(...)` forever. All
subsequent `send` and `send_and_wait` calls return
`SendWhileWaiting`, permanently bricking the session.

Fix: `WaiterGuard` RAII helper that takes the slot on Drop. Construct
right after install, let RAII handle every exit path. The explicit
`self.idle_waiter.lock().take()` calls inside the function disappear,
and the slot is cleared on every cancellation path automatically.

Mutex conversion (folded in from finding #5):

- `Session::idle_waiter`: `tokio::sync::Mutex<Option<IdleWaiter>>` ->
  `parking_lot::Mutex<Option<IdleWaiter>>`. Lock is never held across
  `.await` in any code path; the conversion is what makes
  WaiterGuard's synchronous Drop work without needing an async-spawn
  fallback.

`event_loop` mutex stays `tokio::sync::Mutex` for now — commit C
converts it as part of cooperative shutdown (so the conversion lands
alongside the matching change to Drop for Session).

Tests: two new regression tests in tests/session_test.rs

- `send_and_wait_outer_cancellation_clears_waiter`: outer
  `tokio::time::timeout(50ms, ...)` around `send_and_wait` with a
  60-second inner timeout. Outer fires, dropping the future. Verify
  the next `send` succeeds (no SendWhileWaiting).
- `send_and_wait_drop_clears_waiter`: explicit `JoinHandle::abort` of
  an in-flight send_and_wait. Verify the next `send` succeeds.

Existing 211 tests continue to pass; total now 213.

Validation: cargo test --all-features, cargo doc -D warnings, cargo
+nightly fmt --check, cargo clippy -- -D warnings all clean.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: tclem

rust/src/session.rs

@@ -3,6 +3,7 @@ use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Duration;
 
+use parking_lot::Mutex as ParkingLotMutex;
 use serde_json::Value;
 use tokio::sync::{Mutex, oneshot};
 use tokio::task::JoinHandle;
@@ -42,6 +43,27 @@ struct IdleWaiter {
     last_assistant_message: Option<SessionEvent>,
 }
 
+/// RAII guard that clears the [`Session::idle_waiter`] slot on drop. Used
+/// by [`Session::send_and_wait`] to ensure the slot doesn't leak if the
+/// caller's future is cancelled (outer `tokio::time::timeout` / `select!`
+/// / dropped JoinHandle). Synchronous clear via `parking_lot::Mutex` —
+/// no async drop needed.
+///
+/// Without this, an outer cancellation between "install waiter" and
+/// "drain channel" would leave the slot occupied, causing all subsequent
+/// `send` and `send_and_wait` calls on the session to return
+/// [`SendWhileWaiting`](SessionError::SendWhileWaiting). Closes RFD-400
+/// review finding #2.
+struct WaiterGuard {
+    slot: Arc<ParkingLotMutex<Option<IdleWaiter>>>,
+}
+
+impl Drop for WaiterGuard {
+    fn drop(&mut self) {
+        self.slot.lock().take();
+    }
+}
+
 /// A session on a GitHub Copilot CLI server.
 ///
 /// Created via [`Client::create_session`] or [`Client::resume_session`].
@@ -61,7 +83,12 @@ pub struct Session {
     client: Client,
     event_loop: Mutex<Option<JoinHandle<()>>>,
     /// Only populated while a `send_and_wait` call is in flight.
-    idle_waiter: Arc<Mutex<Option<IdleWaiter>>>,
+    ///
+    /// Sync `parking_lot::Mutex` because the lock is never held across an
+    /// `.await`, and synchronous access lets the `WaiterGuard` RAII helper
+    /// in `send_and_wait` clear the slot from a `Drop` impl on caller-side
+    /// cancellation. See RFD-400 review (cancel-safety hardening).
+    idle_waiter: Arc<ParkingLotMutex<Option<IdleWaiter>>>,
     /// Capabilities negotiated with the CLI, updated on `capabilities.changed` events.
     capabilities: Arc<parking_lot::RwLock<SessionCapabilities>>,
     /// Broadcast channel for runtime event subscribers — see [`Session::subscribe`].
@@ -166,7 +193,7 @@ impl Session {
             let _ = handle.await;
         }
         // Fail any pending send_and_wait so it returns immediately.
-        if let Some(waiter) = self.idle_waiter.lock().await.take() {
+        if let Some(waiter) = self.idle_waiter.lock().take() {
             let _ = waiter
                 .tx
                 .send(Err(Error::Session(SessionError::EventLoopClosed)));
@@ -187,7 +214,7 @@ impl Session {
     /// Returns an error if a [`send_and_wait`](Self::send_and_wait) call is
     /// currently in flight, since the plain send would race with the waiter.
     pub async fn send(&self, opts: impl Into<MessageOptions>) -> Result<String, Error> {
-        if self.idle_waiter.lock().await.is_some() {
+        if self.idle_waiter.lock().is_some() {
             return Err(Error::Session(SessionError::SendWhileWaiting));
         }
         self.send_inner(opts.into()).await
@@ -250,6 +277,14 @@ impl Session {
     /// Only one `send_and_wait` call may be active per session at a time.
     /// Calling [`send`](Self::send) while a `send_and_wait`
     /// is in flight will also return an error.
+    ///
+    /// # Cancel safety
+    ///
+    /// **Cancel-safe.** A `WaiterGuard` clears the in-flight slot on every
+    /// exit path (success, internal failure, internal timeout, *and*
+    /// external cancellation via `tokio::time::timeout` / `select!` /
+    /// dropped JoinHandle). Subsequent `send` and `send_and_wait` calls on
+    /// this session will succeed normally — the slot is never leaked.
     pub async fn send_and_wait(
         &self,
         opts: impl Into<MessageOptions>,
@@ -259,7 +294,7 @@ impl Session {
         let (tx, rx) = oneshot::channel();
 
         {
-            let mut guard = self.idle_waiter.lock().await;
+            let mut guard = self.idle_waiter.lock();
             if guard.is_some() {
                 return Err(Error::Session(SessionError::SendWhileWaiting));
             }
@@ -269,28 +304,26 @@ impl Session {
             });
         }
 
-        let result = tokio::time::timeout(timeout_duration, async {
-            if let Err(e) = self.send_inner(opts).await {
-                self.idle_waiter.lock().await.take();
-                return Err(e);
-            }
+        // RAII: clears the idle_waiter slot on every exit path, including
+        // external cancellation (caller's outer `select!` / `timeout` /
+        // dropped future). Without this, an outer cancellation would leak
+        // the slot and brick subsequent `send`/`send_and_wait` calls.
+        let _waiter_guard = WaiterGuard {
+            slot: self.idle_waiter.clone(),
+        };
 
+        let result = tokio::time::timeout(timeout_duration, async {
+            self.send_inner(opts).await?;
             match rx.await {
                 Ok(result) => result,
-                Err(_) => {
-                    self.idle_waiter.lock().await.take();
-                    Err(Error::Session(SessionError::EventLoopClosed))
-                }
+                Err(_) => Err(Error::Session(SessionError::EventLoopClosed)),
             }
         })
         .await;
 
         match result {
             Ok(inner) => inner,
-            Err(_) => {
-                self.idle_waiter.lock().await.take();
-                Err(Error::Session(SessionError::Timeout(timeout_duration)))
-            }
+            Err(_) => Err(Error::Session(SessionError::Timeout(timeout_duration))),
         }
     }
 
@@ -751,7 +784,7 @@ impl Client {
         ));
         let channels = self.register_session(&session_id);
 
-        let idle_waiter = Arc::new(Mutex::new(None));
+        let idle_waiter = Arc::new(ParkingLotMutex::new(None));
         let (event_tx, _) = tokio::sync::broadcast::channel(512);
         let event_loop = spawn_event_loop(
             session_id.clone(),
@@ -851,7 +884,7 @@ impl Client {
         ));
         let channels = self.register_session(&cli_session_id);
 
-        let idle_waiter = Arc::new(Mutex::new(None));
+        let idle_waiter = Arc::new(ParkingLotMutex::new(None));
         let (event_tx, _) = tokio::sync::broadcast::channel(512);
         let event_loop = spawn_event_loop(
             cli_session_id.clone(),
@@ -905,7 +938,7 @@ fn spawn_event_loop(
     command_handlers: Arc<CommandHandlerMap>,
     session_fs_provider: Option<Arc<dyn SessionFsProvider>>,
     channels: crate::router::SessionChannels,
-    idle_waiter: Arc<Mutex<Option<IdleWaiter>>>,
+    idle_waiter: Arc<ParkingLotMutex<Option<IdleWaiter>>>,
     capabilities: Arc<parking_lot::RwLock<SessionCapabilities>>,
     event_tx: tokio::sync::broadcast::Sender<SessionEvent>,
 ) -> JoinHandle<()> {
@@ -933,7 +966,7 @@ fn spawn_event_loop(
                 }
             }
             // Channels closed — fail any pending send_and_wait.
-            if let Some(waiter) = idle_waiter.lock().await.take() {
+            if let Some(waiter) = idle_waiter.lock().take() {
                 let _ = waiter
                     .tx
                     .send(Err(Error::Session(SessionError::EventLoopClosed)));
@@ -1022,7 +1055,7 @@ async fn handle_notification(
     handler: &Arc<dyn SessionHandler>,
     command_handlers: &Arc<CommandHandlerMap>,
     notification: SessionEventNotification,
-    idle_waiter: &Arc<Mutex<Option<IdleWaiter>>>,
+    idle_waiter: &Arc<ParkingLotMutex<Option<IdleWaiter>>>,
     capabilities: &Arc<parking_lot::RwLock<SessionCapabilities>>,
     event_tx: &tokio::sync::broadcast::Sender<SessionEvent>,
 ) {
@@ -1035,7 +1068,7 @@ async fn handle_notification(
         SessionEventType::AssistantMessage
         | SessionEventType::SessionIdle
         | SessionEventType::SessionError => {
-            let mut guard = idle_waiter.lock().await;
+            let mut guard = idle_waiter.lock();
             if let Some(waiter) = guard.as_mut() {
                 match event_type {
                     SessionEventType::AssistantMessage => {

rust/tests/session_test.rs

@@ -1773,6 +1773,119 @@ async fn send_and_wait_times_out() {
     ));
 }
 
+/// Cancel-safety regression: an outer `tokio::time::timeout` around
+/// `send_and_wait` must NOT leak the `idle_waiter` slot. After the outer
+/// timeout fires and drops the future, subsequent `send` and
+/// `send_and_wait` calls must succeed without `SendWhileWaiting`.
+///
+/// Closes RFD-400 review finding #2.
+#[tokio::test]
+async fn send_and_wait_outer_cancellation_clears_waiter() {
+    let (session, mut server) = create_session_pair(Arc::new(NoopHandler)).await;
+    let session = Arc::new(session);
+
+    // First call: wrap in outer timeout much shorter than the inner
+    // wait_timeout. The outer timeout expires, dropping the
+    // send_and_wait future before the idle/error event arrives.
+    let handle = tokio::spawn({
+        let session = session.clone();
+        async move {
+            tokio::time::timeout(
+                Duration::from_millis(50),
+                session.send_and_wait(
+                    MessageOptions::new("first").with_wait_timeout(Duration::from_secs(60)),
+                ),
+            )
+            .await
+        }
+    });
+
+    let request = server.read_request().await;
+    server.respond(&request, serde_json::json!({})).await;
+
+    // Outer timeout fires → Err(Elapsed) returned, future is dropped.
+    let outer_result = timeout(Duration::from_secs(2), handle)
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(outer_result.is_err(), "outer timeout should have elapsed");
+
+    // The WaiterGuard's Drop should have cleared the slot. A subsequent
+    // `send` must NOT return SendWhileWaiting.
+    let send_handle = tokio::spawn({
+        let session = session.clone();
+        async move { session.send("second").await }
+    });
+
+    let request = server.read_request().await;
+    assert_eq!(request["method"], "session.send");
+    assert_eq!(request["params"]["prompt"], "second");
+    server
+        .respond(
+            &request,
+            serde_json::json!({ "messageId": "msg-after-cancel" }),
+        )
+        .await;
+
+    let result = timeout(TIMEOUT, send_handle).await.unwrap().unwrap();
+    assert_eq!(result.unwrap(), "msg-after-cancel");
+}
+
+/// Cancel-safety regression: explicitly dropping the JoinHandle of an
+/// in-flight `send_and_wait` must clear the waiter slot via WaiterGuard's
+/// Drop. The next `send` must succeed.
+///
+/// Closes RFD-400 review finding #2.
+#[tokio::test]
+async fn send_and_wait_drop_clears_waiter() {
+    let (session, mut server) = create_session_pair(Arc::new(NoopHandler)).await;
+    let session = Arc::new(session);
+
+    // Start a send_and_wait, let it install the waiter, then abort the
+    // task before any idle/error event arrives.
+    let handle = tokio::spawn({
+        let session = session.clone();
+        async move {
+            session
+                .send_and_wait(
+                    MessageOptions::new("aborted").with_wait_timeout(Duration::from_secs(60)),
+                )
+                .await
+        }
+    });
+
+    // Drain the session.send RPC so we know the waiter is installed.
+    let request = server.read_request().await;
+    server.respond(&request, serde_json::json!({})).await;
+
+    // Now abort the in-flight send_and_wait. The WaiterGuard drops as
+    // the future unwinds, clearing the slot.
+    handle.abort();
+    let _ = handle.await;
+
+    // Give the runtime a moment to run the drop.
+    tokio::task::yield_now().await;
+
+    // Next `send` must succeed — no SendWhileWaiting.
+    let send_handle = tokio::spawn({
+        let session = session.clone();
+        async move { session.send("after-abort").await }
+    });
+
+    let request = server.read_request().await;
+    assert_eq!(request["method"], "session.send");
+    assert_eq!(request["params"]["prompt"], "after-abort");
+    server
+        .respond(
+            &request,
+            serde_json::json!({ "messageId": "msg-after-abort" }),
+        )
+        .await;
+
+    let result = timeout(TIMEOUT, send_handle).await.unwrap().unwrap();
+    assert_eq!(result.unwrap(), "msg-after-abort");
+}
+
 #[tokio::test]
 async fn elicitation_requested_dispatches_to_handler_and_responds() {
     use github_copilot_sdk::types::ElicitationResult;