diff --git a/src-tauri/src/agent/implementations/agent_runner.rs b/src-tauri/src/agent/implementations/agent_runner.rs index ec8d2ea96..744dd4bb7 100644 --- a/src-tauri/src/agent/implementations/agent_runner.rs +++ b/src-tauri/src/agent/implementations/agent_runner.rs @@ -30,6 +30,10 @@ where max_steps: u32, current_step: u32, app_handle: Arc, // 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, } impl DefaultAgentRunner @@ -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) -> 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::(); + 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(), ®istry).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::(), + 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::(); + let registry = app_state.agent_sessions(); + if registry.end_needs_input(session_id).await { + crate::agents::broadcast_sessions_updated(self.app_handle.as_ref(), ®istry).await; + } + } + /// Add tool result to memory with proper error handling async fn add_tool_result_to_memory( &mut self, diff --git a/src-tauri/src/agent/input_arbiter.rs b/src-tauri/src/agent/input_arbiter.rs new file mode 100644 index 000000000..2aae12979 --- /dev/null +++ b/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>, + /// 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>>, + cooldown: Duration, +} + +#[derive(Default)] +struct InputArbiterInner { + last_action_at: Option, +} + +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 { + 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 { + 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 { + 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, + holder: Arc>>, + held_by: Option, +} + +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")); + } +} diff --git a/src-tauri/src/agent/mod.rs b/src-tauri/src/agent/mod.rs index 6610fb672..63fe51abe 100644 --- a/src-tauri/src/agent/mod.rs +++ b/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 diff --git a/src-tauri/src/agent/providers/factory.rs b/src-tauri/src/agent/providers/factory.rs index 11f0ceee0..7220ec1f6 100644 --- a/src-tauri/src/agent/providers/factory.rs +++ b/src-tauri/src/agent/providers/factory.rs @@ -451,6 +451,18 @@ impl BrainFactory { pub async fn register_computer_use_tools( provider: &mut LocalToolProvider, app_handle: tauri::AppHandle, + ) -> Result<(), String> { + Self::register_computer_use_tools_for_session(provider, app_handle, None).await + } + + /// Register computer use tools bound to a parallel agent session + /// (LAC-1432). The session id keys the desktop overlay cursor and + /// attributes physical input in the input arbiter; `None` preserves + /// the legacy process-unique cursor identity. + pub async fn register_computer_use_tools_for_session( + provider: &mut LocalToolProvider, + app_handle: tauri::AppHandle, + session: Option, ) -> Result<(), String> { info!("🔧 Registering Computer Use tools (race-condition safe)..."); @@ -476,7 +488,17 @@ impl BrainFactory { provider.set_mcp_manager(mcp_manager); // Register the official Anthropic Computer Use tools (per-provider instance) - register_anthropic_computer_use_tools(provider, app_handle.clone()).await?; + match session { + Some(ctx) => { + crate::agent::tools::anthropic_computer_use::register_anthropic_computer_use_tools_for_session( + provider, + app_handle.clone(), + ctx, + ) + .await? + } + None => register_anthropic_computer_use_tools(provider, app_handle.clone()).await?, + } // Register additional desktop automation tools (per-provider instance) crate::agent::tools::desktop_tools::register_desktop_tools( diff --git a/src-tauri/src/agent/tools/anthropic_computer_use.rs b/src-tauri/src/agent/tools/anthropic_computer_use.rs index d6e4a0b86..e639d60e3 100644 --- a/src-tauri/src/agent/tools/anthropic_computer_use.rs +++ b/src-tauri/src/agent/tools/anthropic_computer_use.rs @@ -46,6 +46,18 @@ const AGENT_CURSOR_COLORS: &[&str] = &[ "#F97316", // orange ]; +/// Identity of the parallel agent session that owns a tool registration +/// (LAC-1432). When present, the agent's overlay cursor is keyed by the +/// session id and drawn in the session's identity color, and physical +/// input is attributed to the session in the input arbiter. When absent +/// (legacy callers), a process-unique `agent-N` cursor id and the legacy +/// palette are used instead. +#[derive(Clone, Debug)] +pub struct SessionToolContext { + pub session_id: String, + pub color: String, +} + /// Emit a cursor position update for a named agent. No-op if the app_handle cannot emit. fn emit_agent_cursor_update( app_handle: &tauri::AppHandle, @@ -70,10 +82,13 @@ fn emit_agent_cursor_update( } /// Emit cursor removal for a named agent (call on agent completion or cancellation). -#[allow(dead_code)] // Symmetric counterpart to emit_agent_cursor_update — staged for completion/cancel paths -fn emit_agent_cursor_remove(app_handle: &tauri::AppHandle, agent_id: &str) { - let state_manager = app_handle.state::(); - state_manager.remove_agent_cursor(agent_id); +/// +/// Called from `SessionHandle::drop` so every session end path — complete, +/// cancel, error, panic unwind — clears the session's overlay cursor. +pub(crate) fn emit_agent_cursor_remove(app_handle: &tauri::AppHandle, agent_id: &str) { + if let Some(state_manager) = app_handle.try_state::() { + state_manager.remove_agent_cursor(agent_id); + } let payload = serde_json::json!({ "agent_id": agent_id }); if let Err(e) = app_handle.emit(crate::constants::events::ui::AGENT_CURSOR_REMOVE, &payload) { tracing::debug!("agent cursor remove emit failed: {}", e); @@ -985,9 +1000,15 @@ macro_rules! handle_anthropic_result { // --- Main computer tool execution function --- /// Execute computer tool +/// +/// `session_id` identifies the parallel agent session issuing the action +/// (LAC-1432). It attributes physical input in the arbiter and drives the +/// session's "current action" shown in the roster/switcher UI. Legacy +/// callers without a session pass `None`. pub async fn execute_computer_tool( app_handle: &tauri::AppHandle, input: Value, + session_id: Option<&str>, ) -> Result { let action = match input["action"].as_str() { Some(action) => action, @@ -1006,6 +1027,19 @@ pub async fn execute_computer_tool( // Enhanced logging with descriptive tool name and action details info!("🖥️ Computer Use: {} → {}", descriptive_tool_name, action); + // Record this session's current action so the roster/switcher UI can + // show what each parallel agent is doing right now. + if let Some(session_id) = session_id { + let registry = state_manager.agent_sessions(); + let id = crate::agents::AgentSessionId::from(session_id.to_string()); + if let Some(session) = registry.get(&id).await { + session + .set_current_action(Some(descriptive_tool_name.clone())) + .await; + crate::agents::broadcast_sessions_updated(app_handle, ®istry).await; + } + } + // Log enhanced tool call request with descriptive name crate::agent::tool_logger::log_enhanced_tool_call_request( app_handle, @@ -1019,6 +1053,30 @@ pub async fn execute_computer_tool( // Enforce cooldown between rapid UI actions to prevent "clicked too fast" failures enforce_action_cooldown(action).await; + // Serialize coordinate-based physical input across parallel sessions + // (LAC-1432). macOS has one hardware pointer, so CGEvent-based actions + // from different sessions must not interleave. Actions listed here are + // ALWAYS physical; the click/type actions that attempt AX-grounded + // interaction first acquire the guard inside their physical fallback + // blocks instead, so AX-only actions keep running in parallel. + let always_physical = matches!( + action, + "middle_click" + | "triple_click" + | "left_click_drag" + | "mouse_move" + | "left_mouse_down" + | "left_mouse_up" + | "key" + | "hold_key" + | "scroll" + ); + let _physical_input_guard = if always_physical { + Some(state_manager.input_arbiter().acquire(session_id).await) + } else { + None + }; + // --- Safety checks --- // 1. Self-automation prevention + blocked app check if let Err(blocked_msg) = check_app_safety(action) { @@ -1143,6 +1201,8 @@ pub async fn execute_computer_tool( emit_ax_grounding_audit(app_handle, action, screen_x, screen_y, &ax_result); if !ax_result.used_ax_click { + // Physical fallback — serialize with other sessions' input. + let _guard = state_manager.input_arbiter().acquire(session_id).await; // Tier 2-4: process-targeted injection (SkyLight → CGEventPostToPid → HID-restore) // Bypasses AX; works on canvas, games, Chromium web content, and non-AX apps. let click_method = state_manager.desktop.left_click_no_warp( @@ -1208,6 +1268,8 @@ pub async fn execute_computer_tool( emit_ax_grounding_audit(app_handle, action, screen_x, screen_y, &ax_result); if !ax_result.used_ax_click { + // Physical fallback — serialize with other sessions' input. + let _guard = state_manager.input_arbiter().acquire(session_id).await; // Tier 2-4: process-targeted injection, no cursor warp let click_method = state_manager .desktop @@ -1294,6 +1356,8 @@ pub async fn execute_computer_tool( emit_ax_grounding_audit(app_handle, action, screen_x, screen_y, &ax_result); if !ax_result.used_ax_click { + // Physical fallback — serialize with other sessions' input. + let _guard = state_manager.input_arbiter().acquire(session_id).await; // Tier 2-4: process-targeted double-click, no cursor warp let click_method = state_manager.desktop.double_click_no_warp( screen_x, @@ -1604,6 +1668,9 @@ pub async fn execute_computer_tool( // keyboard simulation (clipboard paste) when AX isn't supported. let typed_via_ax = try_ax_type_focused(app_handle, text); if !typed_via_ax { + // Physical fallback (clipboard + CGEvent paste) — + // serialize with other sessions' input. + let _guard = state_manager.input_arbiter().acquire(session_id).await; handle_anthropic_result!(crate::commands::keyboard::type_text( text.to_string(), app_handle.clone(), @@ -2297,7 +2364,21 @@ pub async fn register_anthropic_computer_use_tools( provider: &mut LocalToolProvider, app_handle: tauri::AppHandle, ) -> Result<(), String> { - register_anthropic_computer_use_tools_with_version(provider, app_handle, None).await + register_anthropic_computer_use_tools_with_version(provider, app_handle, None, None).await +} + +/// Register Anthropic Computer Use tools bound to a parallel agent session. +/// +/// The session's id becomes the overlay cursor id and its identity color is +/// used for the cursor ring, so the desktop overlay and the roster UI agree +/// on which agent is which (LAC-1432). +pub async fn register_anthropic_computer_use_tools_for_session( + provider: &mut LocalToolProvider, + app_handle: tauri::AppHandle, + session: SessionToolContext, +) -> Result<(), String> { + register_anthropic_computer_use_tools_with_version(provider, app_handle, None, Some(session)) + .await } /// Register Anthropic Computer Use tools with specific API version @@ -2305,6 +2386,7 @@ pub async fn register_anthropic_computer_use_tools_with_version( provider: &mut LocalToolProvider, app_handle: tauri::AppHandle, version_config: Option, + session: Option, ) -> Result<(), String> { let version_info = version_config .as_ref() @@ -2316,12 +2398,22 @@ pub async fn register_anthropic_computer_use_tools_with_version( version_info ); - // Assign a unique cursor ID and color to this agent instance at registration time. - // The ID is captured by the closure so every tool call from this agent shares it. - let cursor_slot = NEXT_AGENT_CURSOR_ID.fetch_add(1, Ordering::Relaxed); - let agent_cursor_id = format!("agent-{}", cursor_slot); - let agent_cursor_color = - AGENT_CURSOR_COLORS[(cursor_slot as usize - 1) % AGENT_CURSOR_COLORS.len()].to_string(); + // Cursor identity: prefer the parallel-session identity (session id + + // palette color) so overlay cursors match the roster UI. Legacy callers + // without a session get a process-unique `agent-N` id and the legacy + // palette, exactly as before. + let (agent_cursor_id, agent_cursor_color, session_id) = match session { + Some(ctx) => (ctx.session_id.clone(), ctx.color, Some(ctx.session_id)), + None => { + let cursor_slot = NEXT_AGENT_CURSOR_ID.fetch_add(1, Ordering::Relaxed); + ( + format!("agent-{}", cursor_slot), + AGENT_CURSOR_COLORS[(cursor_slot as usize - 1) % AGENT_CURSOR_COLORS.len()] + .to_string(), + None, + ) + } + }; info!( "🖱️ Agent cursor ID: {} (color: {})", @@ -2340,12 +2432,19 @@ pub async fn register_anthropic_computer_use_tools_with_version( let handle = app_handle.clone(); let cursor_id = agent_cursor_id.clone(); let cursor_color = agent_cursor_color.clone(); + let session_id = session_id.clone(); move |input: Value| { let handle = handle.clone(); let cursor_id = cursor_id.clone(); let cursor_color = cursor_color.clone(); + let session_id = session_id.clone(); async move { - let result = execute_computer_tool(&handle, input.clone()).await; + let result = execute_computer_tool( + &handle, + input.clone(), + session_id.as_deref(), + ) + .await; // Emit cursor position for the agent overlay (non-blocking) if result.is_ok() { let action = input["action"].as_str().unwrap_or(""); diff --git a/src-tauri/src/agents/desktop_agent.rs b/src-tauri/src/agents/desktop_agent.rs index 5b331b4e3..afd92d35f 100644 --- a/src-tauri/src/agents/desktop_agent.rs +++ b/src-tauri/src/agents/desktop_agent.rs @@ -51,9 +51,17 @@ impl DesktopAgent { "computer" => { // Delegate to the official Anthropic Computer Use tool implementation // This handles all computer actions: click, type, scroll, screenshot, etc. + // + // TODO(LAC-3073): session_id is None because the SpecializedAgent + // path (AgentFactory → handle_task) is not wired through the + // AgentSessionRegistry — no session id exists here yet, and this + // instance is shared across runs so it must not store one. + // Consequence: roster `current_action` doesn't update during + // orchestrated DesktopAgent runs (input arbitration is unaffected). match crate::agent::tools::anthropic_computer_use::execute_computer_tool( &self.app_handle, tool_call.input.clone(), + None, ) .await { diff --git a/src-tauri/src/agents/mod.rs b/src-tauri/src/agents/mod.rs index 101e043b9..c67941f60 100644 --- a/src-tauri/src/agents/mod.rs +++ b/src-tauri/src/agents/mod.rs @@ -3,6 +3,7 @@ pub mod base_agent; pub mod browser_agent; pub mod desktop_agent; pub mod orchestrator; +pub mod session; pub mod system_agent; // Re-export key types for easier use @@ -11,4 +12,8 @@ pub use base_agent::*; pub use browser_agent::BrowserAgent; pub use desktop_agent::DesktopAgent; pub use orchestrator::{Orchestrator, OrchestratorConfig}; +pub use session::{ + broadcast_sessions_updated, color_for_slot, AgentSession, AgentSessionId, AgentSessionInfo, + AgentSessionRegistry, AgentSessionStatus, SessionHandle, SESSION_COLOR_SLOTS, +}; pub use system_agent::SystemAgent; diff --git a/src-tauri/src/agents/session.rs b/src-tauri/src/agents/session.rs new file mode 100644 index 000000000..6878ae547 --- /dev/null +++ b/src-tauri/src/agents/session.rs @@ -0,0 +1,777 @@ +use std::collections::{BTreeSet, HashMap}; +use std::sync::{Arc, Mutex as StdMutex}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use serde::{Deserialize, Serialize}; +use tauri::{AppHandle, Emitter}; +use tokio::sync::{watch, Mutex as TokioMutex}; +use tracing::{debug, info, warn}; +use uuid::Uuid; + +use crate::agent::input_arbiter::InputArbiter; +use crate::constants::events; +use crate::constants::ui::agent_session_colors; + +/// Fixed 8-slot identity palette for parallel agent sessions +/// (LAC-2830 spec section 2). Index = color slot. +pub const SESSION_COLOR_SLOTS: [&str; 8] = [ + agent_session_colors::SLOT_0, + agent_session_colors::SLOT_1, + agent_session_colors::SLOT_2, + agent_session_colors::SLOT_3, + agent_session_colors::SLOT_4, + agent_session_colors::SLOT_5, + agent_session_colors::SLOT_6, + agent_session_colors::SLOT_7, +]; + +/// Round-robin color slot allocator with slot reuse. +/// +/// Freed slots are handed out before fresh ones so long-lived fleets keep +/// stable, distinct colors. If more sessions run than palette slots, slots +/// repeat — a visual collision, not a correctness issue (LAC-2830 spec). +#[derive(Default)] +struct ColorAllocator { + next_slot: u8, + freed: BTreeSet, +} + +impl ColorAllocator { + fn allocate(&mut self) -> u8 { + if let Some(slot) = self.freed.iter().next().copied() { + self.freed.remove(&slot); + return slot; + } + let slot = self.next_slot; + self.next_slot = (self.next_slot + 1) % SESSION_COLOR_SLOTS.len() as u8; + slot + } + + fn free(&mut self, slot: u8) { + if (slot as usize) < SESSION_COLOR_SLOTS.len() { + self.freed.insert(slot); + } + } +} + +/// Hex color for a palette slot. +pub fn color_for_slot(slot: u8) -> &'static str { + SESSION_COLOR_SLOTS[slot as usize % SESSION_COLOR_SLOTS.len()] +} + +/// Unique identifier for a parallel agent session. +/// +/// Every parallel agent gets its own [`AgentSessionId`] so backend state +/// (memory, tool approvals, cancellation token, cursor overlay window) +/// stays isolated across simultaneous runs. +#[derive(Clone, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)] +pub struct AgentSessionId(String); + +impl AgentSessionId { + pub fn new() -> Self { + Self(Uuid::new_v4().to_string()) + } + + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl Default for AgentSessionId { + fn default() -> Self { + Self::new() + } +} + +impl std::fmt::Display for AgentSessionId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } +} + +impl From for AgentSessionId { + fn from(value: String) -> Self { + Self(value) + } +} + +/// Lifecycle status of an agent session, surfaced to the switcher/status-bar UI. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AgentSessionStatus { + Starting, + Running, + NeedsInput, + Cancelling, + Cancelled, + Finished, + Failed, +} + +impl AgentSessionStatus { + /// Terminal states — the session is done and will be removed shortly. + pub fn is_terminal(&self) -> bool { + matches!(self, Self::Cancelled | Self::Finished | Self::Failed) + } +} + +/// Snapshot of a session's user-facing metadata. +/// +/// Emitted to the frontend for the parallel-agents switcher and status bar. +/// This is intentionally serializable and free of any backend handles so it +/// can be sent through Tauri events without leaking Arc/Mutex internals. +#[derive(Clone, Debug, Serialize)] +pub struct AgentSessionInfo { + pub id: String, + pub agent_name: String, + pub color_slot: u8, + pub display_color: String, + pub status: AgentSessionStatus, + pub current_action: Option, + pub started_at_ms: u64, + pub last_activity_ms: u64, + pub focused: bool, +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64 +} + +/// One agent's private slice of session state. +/// +/// Each session owns its own cancellation channel — escape or the switcher +/// UI can cancel a single agent without disturbing the others. Mutable +/// metadata (status, current action) lives behind a `TokioMutex` because +/// it is updated from the async execution loop and read from Tauri command +/// handlers that render the switcher. +pub struct AgentSession { + id: AgentSessionId, + agent_name: String, + color_slot: u8, + cancel_tx: watch::Sender, + cancel_rx: watch::Receiver, + started_at_ms: u64, + inner: TokioMutex, +} + +struct AgentSessionInner { + status: AgentSessionStatus, + current_action: Option, + last_activity_ms: u64, +} + +impl AgentSession { + fn new_with_id(id: AgentSessionId, agent_name: String, color_slot: u8) -> Arc { + let (cancel_tx, cancel_rx) = watch::channel(false); + let started_at_ms = now_ms(); + Arc::new(Self { + id, + agent_name, + color_slot, + cancel_tx, + cancel_rx, + started_at_ms, + inner: TokioMutex::new(AgentSessionInner { + status: AgentSessionStatus::Starting, + current_action: None, + last_activity_ms: started_at_ms, + }), + }) + } + + pub fn id(&self) -> &AgentSessionId { + &self.id + } + + pub fn agent_name(&self) -> &str { + &self.agent_name + } + + pub fn color_slot(&self) -> u8 { + self.color_slot + } + + pub fn display_color(&self) -> &'static str { + color_for_slot(self.color_slot) + } + + /// Clone the cancellation receiver so the agent's execution loop can + /// observe cancellation requests without holding a lock on the registry. + pub fn cancel_receiver(&self) -> watch::Receiver { + self.cancel_rx.clone() + } + + /// Signal cancellation for this session only. Idempotent. + pub fn cancel(&self) { + if let Err(e) = self.cancel_tx.send(true) { + warn!( + "Failed to signal cancellation for session {}: {}", + self.id, e + ); + } + } + + pub fn is_cancelled(&self) -> bool { + *self.cancel_rx.borrow() + } + + pub async fn set_status(&self, status: AgentSessionStatus) { + let mut guard = self.inner.lock().await; + guard.status = status; + guard.last_activity_ms = now_ms(); + } + + /// Compare-and-set: transition to `next` only while the status is still + /// `expected`. Returns whether the transition happened. Used by the + /// needs-input flow so a cancellation that lands mid-wait (status + /// `Cancelling`) is never overwritten. + pub async fn set_status_if( + &self, + expected: AgentSessionStatus, + next: AgentSessionStatus, + ) -> bool { + let mut guard = self.inner.lock().await; + if guard.status != expected { + return false; + } + guard.status = next; + guard.last_activity_ms = now_ms(); + true + } + + pub async fn set_current_action(&self, action: Option) { + let mut guard = self.inner.lock().await; + guard.current_action = action; + guard.last_activity_ms = now_ms(); + } + + pub async fn snapshot(&self, focused: bool) -> AgentSessionInfo { + let guard = self.inner.lock().await; + AgentSessionInfo { + id: self.id.0.clone(), + agent_name: self.agent_name.clone(), + color_slot: self.color_slot, + display_color: self.display_color().to_string(), + status: guard.status, + current_action: guard.current_action.clone(), + started_at_ms: self.started_at_ms, + last_activity_ms: guard.last_activity_ms, + focused, + } + } +} + +/// Registry of every live agent session. +/// +/// Owns the input arbiter shared by all sessions so coordinate-based +/// physical input is serialized across the fleet (macOS has one pointer). +/// AX-grounded actions do not touch the arbiter and run in parallel. +/// +/// `sessions` uses a `TokioMutex` because `list()` calls `session.snapshot().await` +/// while iterating. `focused` uses a plain `StdMutex` — no async work happens +/// while holding it, so the lighter-weight lock is appropriate. +pub struct AgentSessionRegistry { + sessions: TokioMutex>>, + focused: StdMutex>, + colors: StdMutex, + input_arbiter: Arc, + max_parallel: usize, +} + +impl AgentSessionRegistry { + pub fn new(max_parallel: usize, input_arbiter: Arc) -> Self { + Self { + sessions: TokioMutex::new(HashMap::new()), + focused: StdMutex::new(None), + colors: StdMutex::new(ColorAllocator::default()), + input_arbiter, + max_parallel, + } + } + + pub fn max_parallel(&self) -> usize { + self.max_parallel + } + + pub fn input_arbiter(&self) -> Arc { + self.input_arbiter.clone() + } + + /// Create a new session and register it. Fails if the parallel cap is hit. + /// + /// Assigns the next free identity-color slot (LAC-2830 palette); the slot + /// is returned to the allocator when the session is removed. The first + /// session created becomes the focused session automatically; callers can + /// override focus later via [`set_focused`]. + pub async fn create(&self, agent_name: String) -> Result, String> { + let mut sessions = self.sessions.lock().await; + if sessions.len() >= self.max_parallel { + return Err(format!( + "Parallel session cap reached ({}); cancel or finish an existing session first", + self.max_parallel + )); + } + let color_slot = { + let mut colors = self.colors.lock().unwrap_or_else(|e| e.into_inner()); + colors.allocate() + }; + let id = AgentSessionId::new(); + let session = AgentSession::new_with_id(id.clone(), agent_name, color_slot); + sessions.insert(id.clone(), session.clone()); + drop(sessions); + + // Auto-focus the first session so escape has an obvious target. + let mut focused = self.focused.lock().unwrap_or_else(|e| e.into_inner()); + let is_focused = focused.is_none(); + if is_focused { + *focused = Some(id.clone()); + } + info!("Registered agent session {} (focused={})", id, is_focused); + Ok(session) + } + + pub async fn get(&self, id: &AgentSessionId) -> Option> { + self.sessions.lock().await.get(id).cloned() + } + + pub async fn remove(&self, id: &AgentSessionId) { + let mut sessions = self.sessions.lock().await; + let removed = sessions.remove(id); + // Reassign focus while still holding the sessions lock so a + // concurrent remove() cannot delete the replacement candidate + // between our map read and the focus write. `focused` is a std + // mutex held only for this block with no await inside. + { + let mut focused = self.focused.lock().unwrap_or_else(|e| e.into_inner()); + if focused.as_ref() == Some(id) { + *focused = sessions.keys().next().cloned(); + } + } + drop(sessions); + + if let Some(session) = removed { + debug!("Removed agent session {} from registry", id); + let mut colors = self.colors.lock().unwrap_or_else(|e| e.into_inner()); + colors.free(session.color_slot()); + } + } + + pub async fn cancel(&self, id: &AgentSessionId) -> Result<(), String> { + let session = self + .get(id) + .await + .ok_or_else(|| format!("Session {} not found", id))?; + session.set_status(AgentSessionStatus::Cancelling).await; + session.cancel(); + Ok(()) + } + + /// Cancel the currently focused session, if any. + /// + /// Escape handling uses this so pressing escape only kills the agent + /// the user is watching; background sessions keep running. + pub async fn cancel_focused(&self) -> Result { + let focused = self + .focused + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone(); + match focused { + Some(id) => { + self.cancel(&id).await?; + Ok(true) + } + None => Ok(false), + } + } + + pub async fn set_focused(&self, id: Option) -> Result<(), String> { + if let Some(ref candidate) = id { + let sessions = self.sessions.lock().await; + if !sessions.contains_key(candidate) { + return Err(format!("Cannot focus unknown session {}", candidate)); + } + } + *self.focused.lock().unwrap_or_else(|e| e.into_inner()) = id; + Ok(()) + } + + pub fn focused(&self) -> Option { + self.focused + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() + } + + /// Mark a session as blocked on the user (a risky tool batch is waiting + /// for approval). Guarded on `Running` so a session that is already + /// cancelling or terminal never flips back into a waiting state. + /// + /// Returns the post-transition snapshot — `focused` is captured here, in + /// the same call, so the caller's notify-if-background gate and the + /// snapshot it emits cannot disagree. `None` when the session is unknown + /// or not currently `Running`. + pub async fn begin_needs_input(&self, id: &AgentSessionId) -> Option { + let session = self.get(id).await?; + if !session + .set_status_if(AgentSessionStatus::Running, AgentSessionStatus::NeedsInput) + .await + { + return None; + } + let focused = self.focused().as_ref() == Some(id); + Some(session.snapshot(focused).await) + } + + /// Restore `Running` after the input wait resolves (approved, denied, or + /// timed out). Guarded on `NeedsInput` so a cancellation that arrived + /// mid-wait is not clobbered. Returns whether the status changed. + pub async fn end_needs_input(&self, id: &AgentSessionId) -> bool { + match self.get(id).await { + Some(session) => { + session + .set_status_if(AgentSessionStatus::NeedsInput, AgentSessionStatus::Running) + .await + } + None => false, + } + } + + /// List a snapshot of every session for the switcher/status-bar UI. + pub async fn list(&self) -> Vec { + let focused = self + .focused + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone(); + let sessions = self.sessions.lock().await.clone(); + let mut out = Vec::with_capacity(sessions.len()); + for (id, session) in sessions.iter() { + let is_focused = focused.as_ref() == Some(id); + out.push(session.snapshot(is_focused).await); + } + out.sort_by_key(|info| info.started_at_ms); + out + } + + pub async fn len(&self) -> usize { + self.sessions.lock().await.len() + } + + pub async fn is_empty(&self) -> bool { + self.sessions.lock().await.is_empty() + } +} + +/// Emit the current session list to the frontend. +/// +/// Standalone helper so background tasks and RAII cleanup can broadcast +/// updates without needing a `State` handle. +pub async fn broadcast_sessions_updated(app: &AppHandle, registry: &Arc) { + let list = registry.list().await; + if let Err(e) = app.emit(events::agent_sessions::UPDATED, &list) { + warn!("Failed to emit agent-sessions-updated: {}", e); + } +} + +/// RAII guard that removes an agent session from the registry on drop. +/// +/// `execute_agent_internal` has ~8 explicit `return Err` paths plus a +/// fall-through success path; threading manual `registry.remove()` calls +/// through every path is brittle. This guard removes the session on any +/// exit (including panic unwinds) and broadcasts an `agent-sessions-updated` +/// event so the switcher UI drops the row. +/// +/// Registry mutation is async, so cleanup is scheduled on the Tauri async +/// runtime — `Drop` itself stays cheap and synchronous. +pub struct SessionHandle { + registry: Arc, + session: Arc, + app_handle: AppHandle, + active: bool, +} + +impl SessionHandle { + pub fn new( + registry: Arc, + session: Arc, + app_handle: AppHandle, + ) -> Self { + Self { + registry, + session, + app_handle, + active: true, + } + } + + pub fn session(&self) -> &Arc { + &self.session + } + + /// Mark the session as finished/failed/cancelled and broadcast the state + /// before the RAII cleanup removes the row entirely. Callers that know + /// whether the run succeeded or failed should call this to give the + /// UI a final status snapshot instead of the row just disappearing. + /// + /// Also emits the discrete lifecycle event for the terminal state + /// (completed / cancelled / failed) so the roster UI can play its + /// pulse / shake animations and the backend can decide whether to fire + /// a system notification. + /// + /// Returns whether the session was focused at the moment the terminal + /// snapshot was taken. Callers gating notifications on focus must use + /// this value rather than re-reading `is_focused()` — focus can change + /// between the two reads, making the notification gate disagree with + /// the emitted snapshot. + pub async fn mark_terminal(&self, status: AgentSessionStatus) -> bool { + self.session.set_status(status).await; + let focused = self.registry.focused().as_ref() == Some(self.session.id()); + let snapshot = self.session.snapshot(focused).await; + let event = match status { + AgentSessionStatus::Cancelled => Some(events::agent_sessions::CANCELLED), + AgentSessionStatus::Finished => Some(events::agent_sessions::COMPLETED), + AgentSessionStatus::Failed => Some(events::agent_sessions::FAILED), + _ => None, + }; + if let Some(event) = event { + if let Err(e) = self.app_handle.emit(event, &snapshot) { + warn!( + "Failed to emit {} for session {}: {}", + event, snapshot.id, e + ); + } + } + broadcast_sessions_updated(&self.app_handle, &self.registry).await; + focused + } + + /// True when this session is the currently focused one. + pub fn is_focused(&self) -> bool { + self.registry.focused().as_ref() == Some(self.session.id()) + } +} + +/// Remove any cursor overlay identity left behind by a session. +/// +/// The desktop cursor overlay renders one cursor slot per agent id; the +/// computer-use tool registers cursors under the session id (LAC-1432), so +/// clearing that id here guarantees the overlay cursor disappears on every +/// session end path — complete, cancel, or error. +fn cleanup_session_cursor(app_handle: &AppHandle, session_id: &str) { + crate::agent::tools::anthropic_computer_use::emit_agent_cursor_remove(app_handle, session_id); +} + +impl Drop for SessionHandle { + fn drop(&mut self) { + if !self.active { + return; + } + let registry = self.registry.clone(); + let session_id = self.session.id().clone(); + let app_handle = self.app_handle.clone(); + tauri::async_runtime::spawn(async move { + cleanup_session_cursor(&app_handle, session_id.as_str()); + registry.remove(&session_id).await; + broadcast_sessions_updated(&app_handle, ®istry).await; + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + fn arbiter() -> Arc { + Arc::new(InputArbiter::new(Duration::from_millis(0))) + } + + #[tokio::test] + async fn creates_and_lists_sessions() { + let registry = AgentSessionRegistry::new(4, arbiter()); + let a = registry + .create("desktop".into()) + .await + .expect("first session created"); + let b = registry + .create("browser".into()) + .await + .expect("second session created"); + + let listed = registry.list().await; + assert_eq!(listed.len(), 2); + assert!(listed.iter().any(|s| s.id == a.id().to_string())); + assert!(listed.iter().any(|s| s.id == b.id().to_string())); + + // First session auto-focused; snapshot exposes it. + let focused_id = registry.focused().expect("focused set"); + assert_eq!(&focused_id, a.id()); + assert!(listed + .iter() + .any(|s| s.focused && s.id == a.id().to_string())); + } + + #[tokio::test] + async fn enforces_parallel_cap() { + let registry = AgentSessionRegistry::new(1, arbiter()); + registry.create("first".into()).await.expect("first ok"); + let result = registry.create("second".into()).await; + let err = match result { + Ok(_) => panic!("expected second create to fail"), + Err(e) => e, + }; + assert!(err.contains("cap reached"), "unexpected error: {err}"); + } + + #[tokio::test] + async fn cancel_focused_kills_only_focused_session() { + let registry = AgentSessionRegistry::new(4, arbiter()); + let focused_session = registry + .create("focused".into()) + .await + .expect("focused created"); + let background = registry + .create("background".into()) + .await + .expect("background created"); + + let mut focused_rx = focused_session.cancel_receiver(); + let background_rx = background.cancel_receiver(); + + let cancelled = registry.cancel_focused().await.expect("cancel_focused ok"); + assert!(cancelled); + + // The focused session sees the cancel; the background session does not. + assert!(focused_rx.has_changed().unwrap_or(false)); + focused_rx.borrow_and_update(); + assert!(*focused_rx.borrow()); + assert!(!*background_rx.borrow()); + } + + #[tokio::test] + async fn remove_clears_and_reassigns_focus() { + let registry = AgentSessionRegistry::new(4, arbiter()); + let a = registry.create("a".into()).await.unwrap(); + let b = registry.create("b".into()).await.unwrap(); + + assert_eq!(registry.focused().as_ref(), Some(a.id())); + registry.remove(a.id()).await; + // Focus falls back to the remaining session. + assert_eq!(registry.focused().as_ref(), Some(b.id())); + + registry.remove(b.id()).await; + assert!(registry.focused().is_none()); + assert_eq!(registry.len().await, 0); + } + + #[tokio::test] + async fn color_slots_assigned_round_robin_and_freed_on_remove() { + let registry = AgentSessionRegistry::new(12, arbiter()); + let a = registry.create("a".into()).await.unwrap(); + let b = registry.create("b".into()).await.unwrap(); + let c = registry.create("c".into()).await.unwrap(); + assert_eq!(a.color_slot(), 0); + assert_eq!(b.color_slot(), 1); + assert_eq!(c.color_slot(), 2); + assert_eq!(a.display_color(), SESSION_COLOR_SLOTS[0]); + + // Removing a session frees its slot; the next session reuses the + // lowest freed slot instead of advancing the round-robin counter. + registry.remove(b.id()).await; + let d = registry.create("d".into()).await.unwrap(); + assert_eq!(d.color_slot(), 1, "freed slot must be reused first"); + + let e = registry.create("e".into()).await.unwrap(); + assert_eq!(e.color_slot(), 3, "fresh slots continue round-robin"); + } + + #[tokio::test] + async fn set_focused_rejects_unknown_id() { + let registry = AgentSessionRegistry::new(4, arbiter()); + let phantom = AgentSessionId::new(); + assert!(registry.set_focused(Some(phantom)).await.is_err()); + } + + #[tokio::test] + async fn needs_input_transition_and_guarded_restore() { + let registry = AgentSessionRegistry::new(4, arbiter()); + let session = registry.create("desktop".into()).await.expect("created"); + session.set_status(AgentSessionStatus::Running).await; + + let snapshot = registry + .begin_needs_input(session.id()) + .await + .expect("running session enters needs_input"); + assert_eq!(snapshot.status, AgentSessionStatus::NeedsInput); + assert!(snapshot.focused, "first session is auto-focused"); + + assert!(registry.end_needs_input(session.id()).await); + let restored = session.snapshot(true).await; + assert_eq!(restored.status, AgentSessionStatus::Running); + + // Restore is idempotent — a second call finds Running and no-ops. + assert!(!registry.end_needs_input(session.id()).await); + } + + #[tokio::test] + async fn needs_input_reports_background_session_as_unfocused() { + let registry = AgentSessionRegistry::new(4, arbiter()); + let focused = registry.create("focused".into()).await.expect("created"); + let background = registry.create("background".into()).await.expect("created"); + focused.set_status(AgentSessionStatus::Running).await; + background.set_status(AgentSessionStatus::Running).await; + + let snapshot = registry + .begin_needs_input(background.id()) + .await + .expect("background session enters needs_input"); + assert!( + !snapshot.focused, + "background session must be reported unfocused so the caller notifies" + ); + } + + #[tokio::test] + async fn needs_input_restore_does_not_clobber_cancellation() { + let registry = AgentSessionRegistry::new(4, arbiter()); + let session = registry.create("desktop".into()).await.expect("created"); + session.set_status(AgentSessionStatus::Running).await; + + registry + .begin_needs_input(session.id()) + .await + .expect("enters needs_input"); + + // User cancels from the switcher while the approval prompt is open. + registry.cancel(session.id()).await.expect("cancel ok"); + + assert!( + !registry.end_needs_input(session.id()).await, + "restore must not overwrite Cancelling" + ); + let after = session.snapshot(true).await; + assert_eq!(after.status, AgentSessionStatus::Cancelling); + } + + #[tokio::test] + async fn begin_needs_input_requires_running_session() { + let registry = AgentSessionRegistry::new(4, arbiter()); + let session = registry.create("desktop".into()).await.expect("created"); + + // Still Starting — the run loop has not begun, so no approval wait + // can be user-visible yet. + assert!(registry.begin_needs_input(session.id()).await.is_none()); + + session.set_status(AgentSessionStatus::Cancelling).await; + assert!(registry.begin_needs_input(session.id()).await.is_none()); + + // Unknown id after removal. + registry.remove(session.id()).await; + assert!(registry.begin_needs_input(session.id()).await.is_none()); + assert!(!registry.end_needs_input(session.id()).await); + } +} diff --git a/src-tauri/src/anthropic.rs b/src-tauri/src/anthropic.rs index 131392344..eadb96036 100644 --- a/src-tauri/src/anthropic.rs +++ b/src-tauri/src/anthropic.rs @@ -402,6 +402,61 @@ pub async fn submit_query( Ok(()) } +/// Emit a session's discrete terminal lifecycle event and, when the session +/// ended in the BACKGROUND (not focused), fire a macOS notification so the +/// user learns the outcome. Must run on EVERY exit path of +/// `execute_agent_internal` — including early setup failures — otherwise the +/// roster row silently vanishes with no error animation or notification +/// (LAC-2830 §6). The RAII `SessionHandle::drop` only broadcasts the row +/// removal; it never emits the terminal event. +/// +/// The focused session's outcome is already on screen, so notifying for it +/// would be noise. Cancellations are always user-initiated and never notify. +/// Focus is read exactly once (inside `mark_terminal`) so the notification +/// gate and the emitted snapshot cannot disagree. +async fn finish_session_terminal_state( + session_handle: Option<&crate::agents::SessionHandle>, + app_handle: &tauri::AppHandle, + state: &tauri::State<'_, AppState>, + terminal_status: crate::agents::AgentSessionStatus, + outcome_message: &str, +) { + use crate::agents::AgentSessionStatus; + + let Some(handle) = session_handle else { + return; + }; + let was_focused = handle.mark_terminal(terminal_status).await; + + if !was_focused + && !crate::cli::headless::is_headless_mode() + && terminal_status != AgentSessionStatus::Cancelled + { + let agent_name = handle.session().agent_name().to_string(); + let (title, level) = if terminal_status == AgentSessionStatus::Finished { + (format!("{} — Complete", agent_name), "success") + } else { + (format!("{} — Failed", agent_name), "error") + }; + let data = crate::commands::notifications::NotificationData { + title, + message: outcome_message.chars().take(140).collect::(), + level: level.to_string(), + important: Some(matches!(terminal_status, AgentSessionStatus::Failed)), + timeout: None, + }; + if let Err(e) = crate::commands::notifications::send_notification( + app_handle.clone(), + state.clone(), + data, + ) + .await + { + warn!("Failed to send background-session notification: {}", e); + } + } +} + /// Internal agent execution function - handles the actual agent logic async fn execute_agent_internal( query: String, @@ -452,6 +507,52 @@ async fn execute_agent_internal( } }; + // Register this run in the parallel-agent registry so the switcher UI + // sees it and per-session cancel/focus commands have a live target. + // If the parallel cap is hit we log and continue — the queue guarantees + // at most one run at a time today, so the cap should never actually bite + // until LAC-1432 lifts the queue serialization. + let session_handle = { + let registry = state.agent_sessions(); + match registry.create("orchestrator".to_string()).await { + Ok(session) => { + session + .set_status(crate::agents::AgentSessionStatus::Running) + .await; + let handle = crate::agents::SessionHandle::new( + registry.clone(), + session, + app_handle.clone(), + ); + let focused = handle.is_focused(); + let snapshot = handle.session().snapshot(focused).await; + if let Err(e) = + app_handle.emit(crate::constants::events::agent_sessions::STARTED, &snapshot) + { + warn!("Failed to emit agent-session-started: {}", e); + } + crate::agents::broadcast_sessions_updated(&app_handle, ®istry).await; + Some(handle) + } + Err(e) => { + warn!( + "Failed to register agent session in parallel registry: {} — proceeding without session tracking", + e + ); + None + } + } + }; + + // Identity context handed to the computer-use tool registration so the + // overlay cursor is keyed by session id and drawn in the session color. + let session_tool_context = session_handle.as_ref().map(|handle| { + crate::agent::tools::anthropic_computer_use::SessionToolContext { + session_id: handle.session().id().to_string(), + color: handle.session().display_color().to_string(), + } + }); + // TODO: TARS Integration disabled - event system not yet implemented // let agent_run_start_event = JunoAgentEvent::AgentRunStart { // session_id: execution_id.clone(), @@ -506,7 +607,40 @@ async fn execute_agent_internal( } }; - let cancel_rx = state.cancel_rx.clone(); + // Cancellation: merge the global cancel channel (stop-all, legacy paths) + // with this session's private cancel channel (per-session cancel from the + // switcher UI / focused-escape). The runner and all specialists observe + // the merged receiver, so cancelling one session never disturbs others. + // The forwarding task exits when the run drops its receivers (merged_tx + // closes), so it cannot leak across runs. + let cancel_rx = match session_handle.as_ref() { + Some(handle) => { + let (merged_tx, merged_rx) = tokio::sync::watch::channel(false); + let mut global_rx = state.cancel_rx.clone(); + let mut session_rx = handle.session().cancel_receiver(); + tauri::async_runtime::spawn(async move { + loop { + tokio::select! { + _ = merged_tx.closed() => break, + changed = global_rx.changed() => { + if changed.is_err() || *global_rx.borrow() { + let _ = merged_tx.send(true); + break; + } + } + changed = session_rx.changed() => { + if changed.is_err() || *session_rx.borrow() { + let _ = merged_tx.send(true); + break; + } + } + } + } + }); + merged_rx + } + None => state.cancel_rx.clone(), + }; // --- Get Persistent Memory Manager (Orchestrator maintains conversation memory) --- let memory_manager_arc = state.get_memory_manager().await; @@ -670,9 +804,10 @@ async fn execute_agent_internal( // In companion mode, skip all computer use tools — agent observes only if !companion_mode { - if let Err(e) = BrainFactory::register_computer_use_tools( + if let Err(e) = BrainFactory::register_computer_use_tools_for_session( &mut single_agent_tool_provider, app_handle.clone(), + session_tool_context.clone(), ) .await { @@ -687,6 +822,14 @@ async fn execute_agent_internal( .unregister_escape_user(&app_handle, "agent_execution") .await; state.mark_agent_execution_finished(); + finish_session_terminal_state( + session_handle.as_ref(), + &app_handle, + &state, + crate::agents::AgentSessionStatus::Failed, + &err_msg, + ) + .await; return Err(err_msg); } info!("✅ Registered full Computer Use tools for single agent mode"); @@ -795,6 +938,14 @@ async fn execute_agent_internal( .unregister_escape_user(&app_handle, "agent_execution") .await; state.mark_agent_execution_finished(); + finish_session_terminal_state( + session_handle.as_ref(), + &app_handle, + &state, + crate::agents::AgentSessionStatus::Failed, + &err_msg, + ) + .await; return Err(err_msg); } }; @@ -810,7 +961,8 @@ async fn execute_agent_internal( brain, agent::config::MAX_ITERATIONS, app_handle.clone(), - ); + ) + .with_session_id(session_handle.as_ref().map(|h| h.session().id().clone())); info!("✅ Single agent runner created with direct tools (no delegation capabilities)"); info!("🚀 Starting single agent run..."); @@ -856,9 +1008,10 @@ async fn execute_agent_internal( // In companion mode, skip all computer use tools — agent observes only if !companion_mode { - if let Err(e) = BrainFactory::register_computer_use_tools( + if let Err(e) = BrainFactory::register_computer_use_tools_for_session( &mut specialist_tool_provider, app_handle.clone(), + session_tool_context.clone(), ) .await { @@ -873,6 +1026,14 @@ async fn execute_agent_internal( .unregister_escape_user(&app_handle, "agent_execution") .await; state.mark_agent_execution_finished(); + finish_session_terminal_state( + session_handle.as_ref(), + &app_handle, + &state, + crate::agents::AgentSessionStatus::Failed, + &err_msg, + ) + .await; return Err(err_msg); } info!("✅ Registered full Computer Use tools for specialist mode"); @@ -1062,6 +1223,14 @@ async fn execute_agent_internal( .unregister_escape_user(&app_handle, "agent_execution") .await; state.mark_agent_execution_finished(); + finish_session_terminal_state( + session_handle.as_ref(), + &app_handle, + &state, + crate::agents::AgentSessionStatus::Failed, + &err_msg, + ) + .await; return Err(err_msg); } }; @@ -1086,7 +1255,8 @@ async fn execute_agent_internal( orchestrator_brain, agent::config::MAX_ITERATIONS, app_handle.clone(), - ); + ) + .with_session_id(session_handle.as_ref().map(|h| h.session().id().clone())); info!("✅ Orchestrator runner created with delegation tools only"); info!("🚀 Starting multi-agent orchestrator run..."); @@ -1117,6 +1287,32 @@ async fn execute_agent_internal( execution_id ); + // --- Parallel-session terminal state (LAC-1432) --- + // Give the roster UI a final status snapshot (and lifecycle event) before + // the RAII SessionHandle removes the row, and notify the user when a + // BACKGROUND session ends. Setup-failure early returns above call the + // same helper, so every exit path emits a terminal event. + { + use crate::agents::AgentSessionStatus; + let terminal_status = match &agent_result { + Ok(_) => AgentSessionStatus::Finished, + Err(AgentError::Terminated) => AgentSessionStatus::Cancelled, + Err(_) => AgentSessionStatus::Failed, + }; + let outcome_message = match &agent_result { + Ok(message) => message.clone(), + Err(e) => e.to_string(), + }; + finish_session_terminal_state( + session_handle.as_ref(), + &app_handle, + &state, + terminal_status, + &outcome_message, + ) + .await; + } + // TODO: TARS Integration disabled - event system not yet implemented // let agent_run_end_event = JunoAgentEvent::AgentRunEnd { // session_id: execution_id.clone(), diff --git a/src-tauri/src/commands/agent_sessions.rs b/src-tauri/src/commands/agent_sessions.rs new file mode 100644 index 000000000..92398123a --- /dev/null +++ b/src-tauri/src/commands/agent_sessions.rs @@ -0,0 +1,166 @@ +//! Tauri commands that expose the parallel [`AgentSessionRegistry`] to the frontend. +//! +//! LAC-1432 introduces the ability to run multiple agents in parallel, each +//! with its own cursor overlay, cancellation, and status. The frontend +//! session-switcher and status-bar UIs need three things from the backend: +//! +//! 1. A snapshot list of every live session for the switcher. +//! 2. A way to change which session is "focused" (the one whose cursor is +//! highlighted and whose escape key cancels). +//! 3. A way to cancel a specific session, or the focused one. +//! +//! These commands are the surface for that. Every mutating command emits +//! [`events::agent_sessions::UPDATED`] with the fresh snapshot so the +//! frontend never polls; focus changes also emit +//! [`events::agent_sessions::FOCUSED`] so cursor overlays can react to +//! the focus change specifically without diffing the list. + +use serde::Serialize; +use tauri::{AppHandle, Emitter, State}; +use tracing::{debug, warn}; + +use crate::agents::{AgentSessionId, AgentSessionInfo}; +use crate::constants::events; +use crate::state::AppState; + +/// Emit the current session list to the frontend. +/// +/// Called after any mutation so the switcher/status-bar re-render without +/// polling. Failure to emit is logged and swallowed — a broken event bus +/// must not abort the underlying registry mutation. +pub(crate) async fn emit_sessions_updated(app: &AppHandle, state: &AppState) { + let list = state.agent_sessions().list().await; + if let Err(e) = app.emit(events::agent_sessions::UPDATED, &list) { + warn!("Failed to emit agent-sessions-updated: {}", e); + } else { + debug!( + "Emitted agent-sessions-updated with {} sessions", + list.len() + ); + } +} + +#[derive(Serialize)] +struct FocusedPayload { + session_id: Option, +} + +fn emit_focused(app: &AppHandle, session_id: Option) { + let payload = FocusedPayload { session_id }; + if let Err(e) = app.emit(events::agent_sessions::FOCUSED, &payload) { + warn!("Failed to emit agent-session-focused: {}", e); + } +} + +/// Return a snapshot of every live agent session. +/// +/// Ordered by `started_at_ms` ascending so newer sessions appear at the +/// bottom of the switcher regardless of HashMap iteration order. +#[tauri::command] +pub async fn list_agent_sessions( + state: State<'_, AppState>, +) -> Result, String> { + Ok(state.agent_sessions().list().await) +} + +/// Return the id of the currently focused session, if any. +/// +/// The frontend uses this on cold start to know which overlay to draw +/// with the "focused" outline before the first `UPDATED` event lands. +#[tauri::command] +pub async fn get_focused_agent_session( + state: State<'_, AppState>, +) -> Result, String> { + Ok(state.agent_sessions().focused().map(|id| id.to_string())) +} + +/// Focus a session so escape cancels it and its cursor overlay highlights. +/// +/// Pass `null`/`None` to clear focus. Rejects unknown ids so the UI can +/// surface stale focus attempts (e.g. after a session finishes between +/// the switcher render and the click). +#[tauri::command] +pub async fn focus_agent_session( + app: AppHandle, + state: State<'_, AppState>, + session_id: Option, +) -> Result<(), String> { + let registry = state.agent_sessions(); + let target = session_id.clone().map(AgentSessionId::from); + registry.set_focused(target).await?; + emit_focused(&app, session_id); + emit_sessions_updated(&app, &state).await; + Ok(()) +} + +/// Cancel the currently focused session, if any. Returns `true` if a +/// session was cancelled. +/// +/// This is the command the global escape shortcut invokes when the +/// parallel-agent switcher is active. Background sessions keep running +/// so the user can walk away from one agent and come back to another. +#[tauri::command] +pub async fn cancel_focused_agent_session( + app: AppHandle, + state: State<'_, AppState>, +) -> Result { + let registry = state.agent_sessions(); + let cancelled = registry.cancel_focused().await?; + if cancelled { + emit_sessions_updated(&app, &state).await; + } + Ok(cancelled) +} + +/// Cancel a specific session by id. +/// +/// Used by the switcher's per-row "cancel" affordance. Returns an error +/// if the session id is unknown (already finished or never existed) so +/// the UI can distinguish "cancel raced with completion" from success. +#[tauri::command] +pub async fn cancel_agent_session( + app: AppHandle, + state: State<'_, AppState>, + session_id: String, +) -> Result<(), String> { + let registry = state.agent_sessions(); + let id = AgentSessionId::from(session_id); + registry.cancel(&id).await?; + emit_sessions_updated(&app, &state).await; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::agent::input_arbiter::InputArbiter; + use crate::agents::AgentSessionRegistry; + use std::sync::Arc; + use std::time::Duration; + + fn registry() -> AgentSessionRegistry { + AgentSessionRegistry::new(4, Arc::new(InputArbiter::new(Duration::from_millis(0)))) + } + + // Command handlers require a Tauri State/AppHandle so we cover them via + // the registry directly — the commands are thin adapters and the + // registry's own tests exercise the actual state machine. + + #[tokio::test] + async fn focus_missing_session_is_rejected() { + let registry = registry(); + let phantom = AgentSessionId::from("does-not-exist".to_string()); + let result = registry.set_focused(Some(phantom)).await; + assert!(result.is_err(), "expected focus on unknown id to error"); + } + + #[tokio::test] + async fn cancel_focused_reports_false_when_empty() { + let registry = registry(); + let cancelled = registry + .cancel_focused() + .await + .expect("cancel_focused with no sessions returns Ok(false)"); + assert!(!cancelled); + } +} diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs index bf382ba38..5df9de5ac 100644 --- a/src-tauri/src/commands/mod.rs +++ b/src-tauri/src/commands/mod.rs @@ -18,6 +18,7 @@ pub mod registry; pub mod safari_tools; // Removed deprecated dictation_reset module pub mod agent_continuation; +pub mod agent_sessions; // Parallel agent-session registry commands (LAC-1432) pub mod always_listening; pub mod cloud; pub mod cloud_test; @@ -80,6 +81,7 @@ pub use self::safari_tools::{ // Removed unused dev import: pub use self::dev::*; pub use self::dictation::*; // Removed deprecated dictation_reset exports +pub use self::agent_sessions::*; pub use self::always_listening::*; pub use self::cloud::*; pub use self::cloud_test::*; diff --git a/src-tauri/src/commands/stop_coordinator.rs b/src-tauri/src/commands/stop_coordinator.rs index 6dcf7ab32..96bf5ab9b 100644 --- a/src-tauri/src/commands/stop_coordinator.rs +++ b/src-tauri/src/commands/stop_coordinator.rs @@ -203,13 +203,38 @@ impl StopCoordinator { if let Some(agent_op_id) = self.try_register_operation("agent_stop").await { info!("[StopCoordinator] Signaling agent cancellation"); - let cancel_requested = *app_state.cancel_rx.borrow(); - if !cancel_requested { - app_state.signal_cancel(); - cleanup_results.push("Agent cancellation signaled".to_string()); + // Parallel sessions (LAC-1432): escape/stop targets only the + // FOCUSED session so background agents keep working. The global + // cancel signal remains the fallback when no session is + // registered (legacy paths, headless runs). + let cancelled_focused = match app_state.agent_sessions().cancel_focused().await { + Ok(cancelled) => cancelled, + Err(e) => { + warn!("[StopCoordinator] Failed to cancel focused session: {}", e); + false + } + }; + if cancelled_focused { + cleanup_results.push("Focused agent session cancelled".to_string()); + } else { + let cancel_requested = *app_state.cancel_rx.borrow(); + if !cancel_requested { + app_state.signal_cancel(); + cleanup_results.push("Agent cancellation signaled".to_string()); + } } - app_state.mark_agent_execution_finished(); + // The global "is executing" flag serves the whole app, not just + // the focused session. Only clear it when no OTHER session is + // still running — the cancelled session itself may linger in the + // registry until its run tears down, hence <= 1 rather than == 0. + // (Its own run clears the flag again on exit.) Today the queue + // serializes runs so this gate is a no-op, but once LAC-1432 + // lifts the cap, clearing unconditionally would switch off + // execution UI while background agents are still working. + if !cancelled_focused || app_state.agent_sessions().len().await <= 1 { + app_state.mark_agent_execution_finished(); + } self.unregister_operation(&agent_op_id).await; } diff --git a/src-tauri/src/constants/events.rs b/src-tauri/src/constants/events.rs index f42fe459b..523c94b5f 100644 --- a/src-tauri/src/constants/events.rs +++ b/src-tauri/src/constants/events.rs @@ -28,6 +28,37 @@ pub mod agent { pub const QUERY_READY: &str = "agent-query-ready"; } +/// Parallel agent-session lifecycle events (LAC-1432). +/// +/// The frontend session-switcher and status bar listen to these events +/// so the UI stays in sync with the backend `AgentSessionRegistry` +/// without polling. Every registry mutation that changes visible state +/// (create / focus / status / current action / remove) fires +/// `AGENT_SESSIONS_UPDATED` with the full snapshot; `AGENT_SESSION_FOCUSED` +/// fires additionally when the focused session changes so cursor +/// overlays can key off the focus change without diffing the list. +pub mod agent_sessions { + /// Full snapshot of all live sessions. Payload is a list of + /// AgentSessionInfo values. Also serves as the action-update channel: + /// it fires whenever a session's current action or status changes. + /// NOTE: no curly braces in doc comments here — generate-ts-constants.js + /// silently drops constants that follow one. + pub const UPDATED: &str = "agent-sessions-updated"; + /// Focus changed. Payload has a nullable `session_id` string field. + pub const FOCUSED: &str = "agent-session-focused"; + /// A new session started. Payload: AgentSessionInfo snapshot. + pub const STARTED: &str = "agent-session-started"; + /// A session finished successfully. Payload: AgentSessionInfo snapshot. + pub const COMPLETED: &str = "agent-session-completed"; + /// A session was cancelled by the user. Payload: AgentSessionInfo snapshot. + pub const CANCELLED: &str = "agent-session-cancelled"; + /// A session failed with an error. Payload: AgentSessionInfo snapshot. + pub const FAILED: &str = "agent-session-failed"; + /// A session is blocked waiting on user input. Payload: AgentSessionInfo + /// snapshot. Fired while a risky tool batch is pending user approval. + pub const NEEDS_INPUT: &str = "agent-session-needs-input"; +} + /// Streaming events pub mod streaming { pub const TEXT_STREAM: &str = "agent-text-stream"; diff --git a/src-tauri/src/constants/ui.rs b/src-tauri/src/constants/ui.rs index 9a1fab57a..2c70d1549 100644 --- a/src-tauri/src/constants/ui.rs +++ b/src-tauri/src/constants/ui.rs @@ -13,6 +13,23 @@ pub mod window_labels { pub const DYNAMIC_BAR: &str = "dynamic-bar"; } +/// Agent session identity colors (LAC-1432 / LAC-2830 spec section 2). +/// Fixed 8-color palette assigned slot-by-slot as parallel agent sessions spawn. +/// Identity colors are used for cursor overlay rings, roster dots, and labels — +/// never for status dots, which keep the existing blue/yellow/green semantics. +/// NOTE: keep doc comments in this module free of curly braces — the +/// generate-ts-constants.js module parser drops constants that follow one. +pub mod agent_session_colors { + pub const SLOT_0: &str = "#3B82F6"; // blue + pub const SLOT_1: &str = "#10B981"; // emerald + pub const SLOT_2: &str = "#F59E0B"; // amber + pub const SLOT_3: &str = "#F43F5E"; // rose + pub const SLOT_4: &str = "#8B5CF6"; // violet + pub const SLOT_5: &str = "#06B6D4"; // cyan + pub const SLOT_6: &str = "#F97316"; // orange + pub const SLOT_7: &str = "#EC4899"; // pink +} + /// UI element IDs used for element targeting and interactions pub mod element_ids { pub const FLOATING_BAR: &str = "floating-bar"; diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 6e7457659..f39d05185 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -325,6 +325,12 @@ pub fn run() { commands::orchestrator::execute_optimized_workflow, commands::orchestrator::configure_enhanced_orchestrator, commands::orchestrator::benchmark_orchestrator_performance, + // Parallel Agent Sessions (LAC-1432) — per-agent cursors, switcher, escape targeting + commands::agent_sessions::list_agent_sessions, + commands::agent_sessions::get_focused_agent_session, + commands::agent_sessions::focus_agent_session, + commands::agent_sessions::cancel_focused_agent_session, + commands::agent_sessions::cancel_agent_session, // Workflow Orchestration Commands execute_mcp_task, get_workflow_templates, diff --git a/src-tauri/src/state.rs b/src-tauri/src/state.rs index 7ff9f7efd..202c7274b 100644 --- a/src-tauri/src/state.rs +++ b/src-tauri/src/state.rs @@ -30,6 +30,9 @@ use crate::cloud::{CloudClient, CloudConfig, ProductionCloudConnector}; use crate::agent::tools::mcp_integration::{MCPManager, MCPServerStatus}; // Import LocalToolProvider for tool provider registry use crate::agent::implementations::tool_provider::LocalToolProvider; +// Physical-input arbiter + parallel session registry +use crate::agent::input_arbiter::InputArbiter; +use crate::agents::session::AgentSessionRegistry; use crate::constants::{audio, errors::templates, events}; use crate::utils::rate_limiter::GlobalRateLimiters; use crate::utils::string_cache::format_error_cached; @@ -342,9 +345,15 @@ pub struct AppState { // Rate limiting for command safety pub rate_limiters: Arc, + // Parallel agent sessions (LAC-1432): each running agent gets its own + // isolated cancellation, status, and cursor identity. The registry + // also owns the shared input arbiter, so coordinate-based physical + // input is serialized across sessions while AX-grounded actions + // continue to run in parallel. + agent_sessions: Arc, + // Per-agent cursor positions for multi-agent overlay (Phase 4) pub agent_cursors: Arc>>, - // Dynamic storage for other state components state_components: Arc>>>, } @@ -421,14 +430,40 @@ impl AppState { // Use the rate limiters created above rate_limiters, + // Initialize parallel-agent session registry. DEFAULT_COOLDOWN + // matches the existing UI-action cooldown used by + // anthropic_computer_use.rs; the parallel cap of 12 mirrors the + // orchestrator's max_parallel_tasks so we don't outrun the + // higher-level scheduler. Note the cap exceeds the 8-slot + // SESSION_COLOR_SLOTS identity palette — sessions 9-12 reuse + // colors of sessions 1-4, an accepted visual collision per the + // LAC-2830 spec (ColorAllocator wraps mod palette size). + agent_sessions: Arc::new(AgentSessionRegistry::new( + 12, + Arc::new(InputArbiter::new( + crate::agent::input_arbiter::DEFAULT_COOLDOWN, + )), + )), + // Initialize per-agent cursor tracking agent_cursors: Arc::new(StdMutex::new(HashMap::new())), - // Initialize dynamic storage state_components: Arc::new(StdMutex::new(HashMap::new())), } } + /// Registry of parallel agent sessions (LAC-1432). + pub fn agent_sessions(&self) -> Arc { + self.agent_sessions.clone() + } + + /// Shared physical-input arbiter. Callers that emit CGEvent-based + /// clicks, drags, or typing must acquire a guard from this arbiter + /// so parallel agents cannot fight over the single hardware pointer. + pub fn input_arbiter(&self) -> Arc { + self.agent_sessions.input_arbiter() + } + /// Initialize rate limiter cleanup task (must be called after Tokio runtime is ready) pub async fn initialize_rate_limiter_cleanup(&self) { info!("Starting rate limiter cleanup task"); diff --git a/src/App.tsx b/src/App.tsx index 5a4529e4d..61ba12977 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -14,6 +14,7 @@ import { } from "@/components/ui/resizable"; import { ChatContainer, ChatInput } from "@/components/chat"; +import { AgentSessionSwitcher } from "@/components/AgentSessionSwitcher"; import ClickVisualizer from "@/components/ClickVisualizer"; import CommandOverlay from "@/components/CommandOverlay"; import KeyPressOverlay from "@/components/KeyPressOverlay"; @@ -29,6 +30,7 @@ import { useMenuEvents } from "@/hooks/useMenuEvents"; import { useSound, useVoiceSounds } from "@/hooks/useSound"; import { useShortcutEvents } from "@/hooks/useShortcutEvents"; import { useDictationStateEvents } from "@/hooks/useDictationStateEvents"; +import { useAgentSessions } from "@/hooks/useAgentSessions"; import { useUpdater } from "@/hooks/useUpdater"; function App() { @@ -38,6 +40,7 @@ function App() { const { isOnboardingActive } = useOnboardingState(); const audioPlayback = useAudioPlayback(); const { playError } = useSound(); + const agentSessions = useAgentSessions(); const { checkForUpdates, installUpdate } = useUpdater(); @@ -452,6 +455,12 @@ function App() { onContinuationUpdate={handleContinuationUpdate} /> + +
0 + ? Math.min(agentSessionCount * 32 + 60, 160) + : 0), + }; case "chat": return { width: CHAT_WIDTH, height: CHAT_HEIGHT }; case "settings": @@ -42,6 +53,7 @@ export default function FloatingPanel() { const [panelMode, setPanelMode] = useState< "compact" | "expanded" | "chat" | "settings" >("compact"); + const { sessions: agentSessions } = useAgentSessions(); useEffect(() => { let mounted = true; @@ -98,7 +110,10 @@ export default function FloatingPanel() { const resizeWindow = async () => { try { const appWindow = getCurrentWindow(); - const dimensions = getWindowDimensionsForMode(panelMode); + const dimensions = getWindowDimensionsForMode( + panelMode, + agentSessions.length + ); if (panelMode === "compact") { // Compact state - delay to allow CSS transitions to complete @@ -143,7 +158,7 @@ export default function FloatingPanel() { clearTimeout(timeoutId); } }; - }, [panelMode, windowReady]); + }, [panelMode, windowReady, agentSessions.length]); // Listen for Rust-based window hover events (same as floating bar) useEffect(() => { diff --git a/src/components/AgentRosterStrip.tsx b/src/components/AgentRosterStrip.tsx new file mode 100644 index 000000000..07ecabbcd --- /dev/null +++ b/src/components/AgentRosterStrip.tsx @@ -0,0 +1,141 @@ +import { useState } from "react"; +import { cn } from "@/lib/utils"; +import type { AgentSessionInfo, AgentSessionStatus } from "@/hooks/useAgentSessions"; + +interface AgentRosterStripProps { + sessions: AgentSessionInfo[]; + onFocus: (sessionId: string) => void; + className?: string; +} + +/** Max dots shown before collapsing into a "+N" overflow label (palette size). */ +const MAX_VISIBLE_DOTS = 8; + +const STATUS_LABELS: Record = { + starting: "Starting", + running: "Working", + needs_input: "Needs input", + cancelling: "Stopping", + cancelled: "Cancelled", + finished: "Done", + failed: "Failed", +}; + +/** Mini status overlay color for the bottom-right badge on each dot. */ +const STATUS_OVERLAY_CLASS: Record = { + starting: "bg-gray-400", + running: "bg-yellow-400 animate-pulse", + needs_input: "bg-white", + cancelling: "bg-gray-400", + cancelled: "bg-gray-400", + finished: "bg-green-400", + failed: "bg-red-400", +}; + +/** Ring/animation treatment for notification states (LAC-2830 spec section 6). */ +function dotStateClass(session: AgentSessionInfo): string { + switch (session.status) { + case "finished": + return "ring-2 ring-green-400 animate-agent-completion-pulse"; + case "failed": + return "ring-2 ring-red-400 animate-agent-error-shake"; + case "needs_input": + return "animate-agent-needs-input-blink"; + default: + return ""; + } +} + +/** + * Compact roster of parallel agent sessions (LAC-2830 spec section 3). + * + * One colored dot per live session, shown beneath the floating bar whenever + * two or more agents run. The dot color is the agent's identity color; the + * small overlay badge communicates status. Clicking a dot focuses that + * session — background sessions keep running. Renders nothing with fewer + * than two sessions so the single-agent experience is unchanged. + */ +export function AgentRosterStrip({ sessions, onFocus, className }: AgentRosterStripProps) { + const [hoveredId, setHoveredId] = useState(null); + + if (sessions.length < 2) return null; + + const visible = sessions.slice(0, MAX_VISIBLE_DOTS); + const overflow = sessions.length - visible.length; + + return ( +
+ {visible.map((session) => { + const label = `Switch to ${session.agent_name} — ${STATUS_LABELS[session.status]}`; + const tooltip = + session.status === "running" && session.current_action + ? `${session.agent_name} — ${session.current_action}` + : `${session.agent_name} — ${STATUS_LABELS[session.status]}`; + return ( +
+ {hoveredId === session.id && ( +
+ {tooltip} +
+ )} + +
+ ); + })} + {overflow > 0 && ( + + +{overflow} + + )} +
+ ); +} diff --git a/src/components/AgentSessionRows.tsx b/src/components/AgentSessionRows.tsx new file mode 100644 index 000000000..37db1c045 --- /dev/null +++ b/src/components/AgentSessionRows.tsx @@ -0,0 +1,97 @@ +import { AlertCircle, Bell, Check, Loader2, X } from "lucide-react"; +import { cn } from "@/lib/utils"; +import type { AgentSessionInfo, AgentSessionStatus } from "@/hooks/useAgentSessions"; + +interface AgentSessionRowsProps { + sessions: AgentSessionInfo[]; + onFocus: (sessionId: string) => void; + className?: string; +} + +const STATUS_LABELS: Record = { + starting: "Starting", + running: "Working", + needs_input: "Needs input", + cancelling: "Stopping", + cancelled: "Cancelled", + finished: "Done", + failed: "Failed", +}; + +function StatusIcon({ status }: { status: AgentSessionStatus }) { + switch (status) { + case "starting": + case "running": + case "cancelling": + return