feat(LAC-1432): parallel background agent sessions — registry wiring, input arbiter, per-session cursors, roster UI, notifications

src-tauri/src/agent/implementations/agent_runner.rs

@@ -30,6 +30,10 @@ where
     max_steps: u32,
     current_step: u32,
     app_handle: Arc<AppHandle>, // Added AppHandle for logging
+    /// Parallel-session registry row this run belongs to, when session
+    /// tracking is active. Lets the approval wait surface `NeedsInput` in
+    /// the switcher UI and notify for background sessions (LAC-1432).
+    session_id: Option<crate::agents::AgentSessionId>,
 }
 
 impl<M, T> DefaultAgentRunner<M, T>
@@ -56,6 +60,7 @@ where
             max_steps,
             current_step: 0,
             app_handle: Arc::new(app_handle), // Store AppHandle
+            session_id: None,
         }
     }
 
@@ -76,9 +81,18 @@ where
             max_steps,
             current_step: 0,
             app_handle: Arc::new(app_handle), // Store AppHandle
+            session_id: None,
         }
     }
 
+    /// Associate this runner with a parallel-session registry row so the
+    /// tool-approval wait can flip the session to `NeedsInput` and notify
+    /// the user when the session is running in the background (LAC-1432).
+    pub fn with_session_id(mut self, session_id: Option<crate::agents::AgentSessionId>) -> Self {
+        self.session_id = session_id;
+        self
+    }
+
     /// Filter tools based on brain type to prevent access to inappropriate tools
     fn filter_tools_for_brain(
         &self,
@@ -523,6 +537,11 @@ where
             max_risk
         );
 
+        // Surface the wait in the parallel-session registry: the switcher
+        // row flips to "Needs input" and a background (unfocused) session
+        // fires a notification so the user knows to come look (LAC-1432).
+        let marked_needs_input = self.mark_session_needs_input(&batch_description).await;
+
         // Poll for up to timeout_seconds at 50 ms intervals.
         let poll_iterations = (approval_request.timeout_seconds * 1000 / 50) as i64;
         let mut remaining = poll_iterations;
@@ -532,6 +551,11 @@ where
             if *cancel_rx.borrow() {
                 log::info!("Cancellation detected during approval wait");
                 app_state.remove_tool_approval(&batch_id).await;
+                if marked_needs_input {
+                    // Guarded restore: no-ops when the cancel already moved
+                    // the session to Cancelling via the registry.
+                    self.clear_session_needs_input().await;
+                }
                 return Err(AgentError::Terminated);
             }
 
@@ -553,6 +577,9 @@ where
         }
 
         app_state.remove_tool_approval(&batch_id).await;
+        if marked_needs_input {
+            self.clear_session_needs_input().await;
+        }
 
         if !approved {
             let reason = if remaining <= 0 {
@@ -578,6 +605,72 @@ where
         Ok(approved)
     }
 
+    /// Flip this runner's session row to `NeedsInput` while a tool-approval
+    /// prompt is pending. Emits the discrete needs-input lifecycle event,
+    /// rebroadcasts the session list for the switcher UI, and — when the
+    /// session is unfocused (running in the background) — sends a
+    /// notification so the user learns an agent is waiting on them
+    /// (LAC-1432 criterion 5b). Returns whether the transition happened;
+    /// the caller must call `clear_session_needs_input` once the wait
+    /// resolves.
+    async fn mark_session_needs_input(&self, description: &str) -> bool {
+        let Some(session_id) = self.session_id.as_ref() else {
+            return false;
+        };
+        let app_state = self.app_handle.state::<crate::state::AppState>();
+        let registry = app_state.agent_sessions();
+        let Some(snapshot) = registry.begin_needs_input(session_id).await else {
+            return false;
+        };
+        let was_focused = snapshot.focused;
+        let agent_name = snapshot.agent_name.clone();
+
+        if let Err(e) = self.app_handle.emit(
+            crate::constants::events::agent_sessions::NEEDS_INPUT,
+            &snapshot,
+        ) {
+            log::error!("Failed to emit agent-session-needs-input: {}", e);
+        }
+        crate::agents::broadcast_sessions_updated(self.app_handle.as_ref(), &registry).await;
+
+        // The focused session's approval dialog is already on screen;
+        // notifying for it would be noise (same gate as terminal-state
+        // notifications in anthropic.rs).
+        if !was_focused && !crate::cli::headless::is_headless_mode() {
+            let data = crate::commands::notifications::NotificationData {
+                title: format!("{} — Needs input", agent_name),
+                message: description.chars().take(140).collect::<String>(),
+                level: "warning".to_string(),
+                important: Some(true),
+                timeout: None,
+            };
+            if let Err(e) = crate::commands::notifications::send_notification(
+                self.app_handle.as_ref().clone(),
+                app_state.clone(),
+                data,
+            )
+            .await
+            {
+                log::warn!("Failed to send needs-input notification: {}", e);
+            }
+        }
+        true
+    }
+
+    /// Restore the session to `Running` after the approval wait resolves.
+    /// The registry guards the transition, so a cancellation that landed
+    /// mid-wait (status `Cancelling`) is never overwritten.
+    async fn clear_session_needs_input(&self) {
+        let Some(session_id) = self.session_id.as_ref() else {
+            return;
+        };
+        let app_state = self.app_handle.state::<crate::state::AppState>();
+        let registry = app_state.agent_sessions();
+        if registry.end_needs_input(session_id).await {
+            crate::agents::broadcast_sessions_updated(self.app_handle.as_ref(), &registry).await;
+        }
+    }
+
     /// Add tool result to memory with proper error handling
     async fn add_tool_result_to_memory(
         &mut self,

src-tauri/src/agent/input_arbiter.rs

@@ -0,0 +1,234 @@
+use std::sync::{Arc, Mutex as StdMutex};
+use std::time::{Duration, Instant};
+
+use tokio::sync::{Mutex as TokioMutex, OwnedMutexGuard};
+use tracing::{debug, trace};
+
+/// Serializes coordinate-based physical input across all agent sessions.
+///
+/// macOS exposes exactly one hardware pointer, so N parallel agents cannot
+/// execute CGEvent-based clicks, drags, or typing simultaneously. Every
+/// call site that emits a physical input event must acquire a
+/// [`PhysicalInputGuard`] first — the guard blocks other sessions until it
+/// is dropped, and enforces a small cooldown between actions so we never
+/// fire pointer events faster than macOS reliably delivers them.
+///
+/// AX-grounded actions (`AXPress` via the accessibility API) do NOT go
+/// through this arbiter. They do not move the physical pointer, so multiple
+/// agents can invoke them concurrently — that is Juno's parallelism moat.
+/// Default cooldown between coordinate-based input actions.
+///
+/// 500 ms gives macOS time to process one event before the next lands.
+/// Callers that need tighter pacing can construct an [`InputArbiter`] with
+/// a custom [`Duration`], but this constant should be preferred for
+/// production agent sessions.
+pub const DEFAULT_COOLDOWN: Duration = Duration::from_millis(500);
+
+pub struct InputArbiter {
+    inner: Arc<TokioMutex<InputArbiterInner>>,
+    /// Observable holder id, kept OUTSIDE the input mutex so observers can
+    /// ask "who holds the arbiter?" while a guard is held. Storing it inside
+    /// `inner` would deadlock any `held_by()` call made during a hold.
+    holder: Arc<StdMutex<Option<String>>>,
+    cooldown: Duration,
+}
+
+#[derive(Default)]
+struct InputArbiterInner {
+    last_action_at: Option<Instant>,
+}
+
+impl InputArbiter {
+    pub fn new(cooldown: Duration) -> Self {
+        Self {
+            inner: Arc::new(TokioMutex::new(InputArbiterInner::default())),
+            holder: Arc::new(StdMutex::new(None)),
+            cooldown,
+        }
+    }
+
+    pub fn cooldown(&self) -> Duration {
+        self.cooldown
+    }
+
+    fn set_holder(&self, session_id: Option<&str>) -> Option<String> {
+        let held = session_id.map(|s| s.to_string());
+        *self
+            .holder
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) = held.clone();
+        held
+    }
+
+    /// Acquire exclusive access to physical input.
+    ///
+    /// Blocks until any current holder releases, then sleeps for the
+    /// remainder of the cooldown if the previous action was too recent.
+    /// The returned guard tracks the caller's session id for observability;
+    /// pass `None` for internal/system callers that are not agent-scoped.
+    pub async fn acquire(&self, session_id: Option<&str>) -> PhysicalInputGuard {
+        let guard = self.inner.clone().lock_owned().await;
+        if let Some(last) = guard.last_action_at {
+            let elapsed = last.elapsed();
+            if elapsed < self.cooldown {
+                let sleep_for = self.cooldown - elapsed;
+                trace!(
+                    "InputArbiter cooldown sleep {:?} for session {:?}",
+                    sleep_for,
+                    session_id
+                );
+                tokio::time::sleep(sleep_for).await;
+            }
+        }
+        let held_by = self.set_holder(session_id);
+        debug!("InputArbiter acquired by session {:?}", session_id);
+        PhysicalInputGuard {
+            guard,
+            holder: self.holder.clone(),
+            held_by,
+        }
+    }
+
+    /// Try to acquire without blocking. Returns `None` if another session holds it.
+    ///
+    /// Does NOT enforce the cooldown — callers using try_acquire opt into
+    /// firing as soon as they win the lock. Prefer [`acquire`] for normal
+    /// agent input paths.
+    pub async fn try_acquire(&self, session_id: Option<&str>) -> Option<PhysicalInputGuard> {
+        match self.inner.clone().try_lock_owned() {
+            Ok(guard) => {
+                let held_by = self.set_holder(session_id);
+                Some(PhysicalInputGuard {
+                    guard,
+                    holder: self.holder.clone(),
+                    held_by,
+                })
+            }
+            Err(_) => None,
+        }
+    }
+
+    /// Session id currently holding the arbiter, if any. For observability
+    /// only. Safe to call while a guard is held — the holder id lives
+    /// outside the input mutex.
+    pub fn held_by(&self) -> Option<String> {
+        self.holder
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .clone()
+    }
+}
+
+impl Default for InputArbiter {
+    fn default() -> Self {
+        Self::new(DEFAULT_COOLDOWN)
+    }
+}
+
+/// RAII guard for exclusive physical input access.
+///
+/// Records the release time on drop so the cooldown applies to the next
+/// caller regardless of exit path (success, error via `?`, panic).
+pub struct PhysicalInputGuard {
+    guard: OwnedMutexGuard<InputArbiterInner>,
+    holder: Arc<StdMutex<Option<String>>>,
+    held_by: Option<String>,
+}
+
+impl PhysicalInputGuard {
+    pub fn held_by(&self) -> Option<&str> {
+        self.held_by.as_deref()
+    }
+}
+
+impl Drop for PhysicalInputGuard {
+    fn drop(&mut self) {
+        self.guard.last_action_at = Some(Instant::now());
+        *self
+            .holder
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::time::Duration;
+
+    #[tokio::test]
+    async fn serializes_concurrent_acquire() {
+        let arbiter = Arc::new(InputArbiter::new(Duration::from_millis(0)));
+        let counter = Arc::new(AtomicUsize::new(0));
+        let observed_max = Arc::new(AtomicUsize::new(0));
+
+        let mut handles = Vec::new();
+        for i in 0..8 {
+            let arbiter = arbiter.clone();
+            let counter = counter.clone();
+            let observed_max = observed_max.clone();
+            handles.push(tokio::spawn(async move {
+                let _guard = arbiter.acquire(Some(&format!("s{i}"))).await;
+                let current = counter.fetch_add(1, Ordering::SeqCst) + 1;
+                observed_max.fetch_max(current, Ordering::SeqCst);
+                tokio::time::sleep(Duration::from_millis(2)).await;
+                counter.fetch_sub(1, Ordering::SeqCst);
+            }));
+        }
+        for h in handles {
+            h.await.expect("task ok");
+        }
+
+        assert_eq!(
+            observed_max.load(Ordering::SeqCst),
+            1,
+            "arbiter must serialize physical input across sessions"
+        );
+    }
+
+    #[tokio::test]
+    async fn enforces_cooldown_between_actions() {
+        let cooldown = Duration::from_millis(30);
+        let arbiter = InputArbiter::new(cooldown);
+
+        {
+            let _g = arbiter.acquire(None).await;
+        }
+        let start = Instant::now();
+        {
+            let _g = arbiter.acquire(None).await;
+        }
+        let elapsed = start.elapsed();
+        assert!(
+            elapsed >= cooldown,
+            "second acquire completed too fast ({:?}, cooldown {:?})",
+            elapsed,
+            cooldown
+        );
+    }
+
+    #[tokio::test]
+    async fn try_acquire_returns_none_while_held() {
+        let arbiter = Arc::new(InputArbiter::new(Duration::from_millis(0)));
+        let _held = arbiter.acquire(Some("holder")).await;
+        assert!(arbiter.try_acquire(Some("other")).await.is_none());
+        // held_by() must not deadlock while the guard is held — the holder
+        // id lives outside the input mutex precisely for this.
+        assert_eq!(arbiter.held_by().as_deref(), Some("holder"));
+    }
+
+    #[tokio::test]
+    async fn guard_drop_releases_and_clears_holder() {
+        let arbiter = Arc::new(InputArbiter::new(Duration::from_millis(0)));
+        {
+            let _g = arbiter.acquire(Some("first")).await;
+            assert_eq!(arbiter.held_by().as_deref(), Some("first"));
+        }
+        // Holder is cleared once the guard drops.
+        assert_eq!(arbiter.held_by(), None);
+        // Second caller wins immediately, and holder is reset.
+        let second = arbiter.acquire(Some("second")).await;
+        assert_eq!(second.held_by(), Some("second"));
+    }
+}

src-tauri/src/agent/mod.rs

@@ -1,6 +1,7 @@
 pub mod core; // Core agent traits and types for orchestration
 pub mod error_recovery; // Enhanced error recovery with checkpoint and rollback
 pub mod implementations;
+pub mod input_arbiter; // Physical input serialization across parallel agent sessions
 pub mod intelligence;
 pub mod multi_agent; // Multi-agent orchestration system
 pub mod prompts; // Centralized prompt management system