fix(cortex-collab): track pending reservations in Guards spawn slot limit (#477)

The test_guards_reservation test was failing because reserve_spawn_slot()
only checked threads_set.len() for the current count, but that set only
contains committed threads. This allowed multiple pending reservations to
bypass the limit.

Added pending_count atomic counter to track reservations that haven't been
committed yet. The check now includes both active threads and pending
reservations: current_count + pending >= max_threads.

- Added pending_count field to Guards struct
- Updated reserve_spawn_slot to check pending count when enforcing limit
- Updated commit() to decrement pending_count when reservation becomes active
- Updated cancel() and Drop to decrement pending_count when releasing

Co-authored-by: Droid Agent <droid@factory.ai>

Author: echobt

cortex-collab/src/guards.rs

@@ -20,6 +20,9 @@ pub struct Guards {
     /// Total count of spawned threads (includes completed).
     total_count: AtomicUsize,
 
+    /// Count of pending reservations (not yet committed).
+    pending_count: AtomicUsize,
+
     /// Maximum allowed concurrent threads.
     max_threads: usize,
 }
@@ -30,6 +33,7 @@ impl Guards {
         Self {
             threads_set: Mutex::new(HashSet::new()),
             total_count: AtomicUsize::new(0),
+            pending_count: AtomicUsize::new(0),
             max_threads: DEFAULT_MAX_THREADS,
         }
     }
@@ -39,6 +43,7 @@ impl Guards {
         Self {
             threads_set: Mutex::new(HashSet::new()),
             total_count: AtomicUsize::new(0),
+            pending_count: AtomicUsize::new(0),
             max_threads,
         }
     }
@@ -48,12 +53,14 @@ impl Guards {
     pub async fn reserve_spawn_slot(self: &Arc<Self>) -> Option<SpawnReservation> {
         let threads = self.threads_set.lock().await;
         let current_count = threads.len();
+        let pending = self.pending_count.load(Ordering::Acquire);
 
-        if current_count >= self.max_threads {
+        if current_count + pending >= self.max_threads {
             return None;
         }
 
-        // Increment total count
+        // Increment pending and total count
+        self.pending_count.fetch_add(1, Ordering::AcqRel);
         self.total_count.fetch_add(1, Ordering::AcqRel);
 
         Some(SpawnReservation {
@@ -109,12 +116,15 @@ impl SpawnReservation {
     /// This transfers ownership of the slot to the spawned thread.
     pub async fn commit(mut self, thread_id: ThreadId) {
         self.state.register_spawned_thread(thread_id).await;
+        // Decrement pending count since it's now committed
+        self.state.pending_count.fetch_sub(1, Ordering::AcqRel);
         self.active = false;
     }
 
     /// Cancel the reservation without spawning.
     pub fn cancel(mut self) {
         if self.active {
+            self.state.pending_count.fetch_sub(1, Ordering::AcqRel);
             self.state.total_count.fetch_sub(1, Ordering::AcqRel);
             self.active = false;
         }
@@ -124,7 +134,8 @@ impl SpawnReservation {
 impl Drop for SpawnReservation {
     fn drop(&mut self) {
         if self.active {
-            // If not committed, decrement the count
+            // If not committed, decrement the counts
+            self.state.pending_count.fetch_sub(1, Ordering::AcqRel);
             self.state.total_count.fetch_sub(1, Ordering::AcqRel);
         }
     }