From 4d12e53741794022c4d46906fbf4518756cacd3d Mon Sep 17 00:00:00 2001 From: Lacy Morrow Date: Sun, 19 Jul 2026 01:07:58 -0400 Subject: [PATCH 01/12] wip(LAC-1432): session registry and input arbiter scaffolding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Salvaged uncommitted work from the stale LAC-1400 worktree during LAC-2841 workspace repair. Not yet compiled or reviewed — resume on this branch when LAC-1432 work restarts. Co-Authored-By: Paperclip --- src-tauri/src/agent/input_arbiter.rs | 191 ++++++++++++ src-tauri/src/agent/mod.rs | 1 + src-tauri/src/agents/mod.rs | 4 + src-tauri/src/agents/session.rs | 423 +++++++++++++++++++++++++++ src-tauri/src/state.rs | 32 ++ 5 files changed, 651 insertions(+) create mode 100644 src-tauri/src/agent/input_arbiter.rs create mode 100644 src-tauri/src/agents/session.rs diff --git a/src-tauri/src/agent/input_arbiter.rs b/src-tauri/src/agent/input_arbiter.rs new file mode 100644 index 000000000..586b89a99 --- /dev/null +++ b/src-tauri/src/agent/input_arbiter.rs @@ -0,0 +1,191 @@ +use std::sync::Arc; +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. +pub struct InputArbiter { + inner: Arc>, + cooldown: Duration, +} + +#[derive(Default)] +struct InputArbiterInner { + last_action_at: Option, + held_by: Option, +} + +impl InputArbiter { + pub fn new(cooldown: Duration) -> Self { + Self { + inner: Arc::new(TokioMutex::new(InputArbiterInner::default())), + cooldown, + } + } + + pub fn cooldown(&self) -> Duration { + self.cooldown + } + + /// 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 mut 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; + } + } + guard.held_by = session_id.map(|s| s.to_string()); + debug!("InputArbiter acquired by session {:?}", session_id); + PhysicalInputGuard { guard } + } + + /// 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(mut guard) => { + guard.held_by = session_id.map(|s| s.to_string()); + Some(PhysicalInputGuard { guard }) + } + Err(_) => None, + } + } + + /// Session id currently holding the arbiter, if any. For observability only. + pub async fn held_by(&self) -> Option { + self.inner.lock().await.held_by.clone() + } +} + +impl Default for InputArbiter { + fn default() -> Self { + Self::new(Duration::from_millis(50)) + } +} + +/// 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, +} + +impl PhysicalInputGuard { + pub fn held_by(&self) -> Option<&str> { + self.guard.held_by.as_deref() + } +} + +impl Drop for PhysicalInputGuard { + fn drop(&mut self) { + self.guard.last_action_at = Some(Instant::now()); + self.guard.held_by = 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()); + assert_eq!(arbiter.held_by().await.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().await.as_deref(), Some("first")); + } + // 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 3cc2a2d6b..6868be202 100644 --- a/src-tauri/src/agent/mod.rs +++ b/src-tauri/src/agent/mod.rs @@ -9,6 +9,7 @@ pub mod core; // Core agent traits and types for orchestration pub mod multi_agent; // Multi-agent orchestration system pub mod error_recovery; // Enhanced error recovery with checkpoint and rollback pub mod intelligence; // Tool choice intelligence system +pub mod input_arbiter; // Physical input serialization across parallel agent sessions // Re-export commonly used items pub use core::*; diff --git a/src-tauri/src/agents/mod.rs b/src-tauri/src/agents/mod.rs index 840a519fb..e9655d9a2 100644 --- a/src-tauri/src/agents/mod.rs +++ b/src-tauri/src/agents/mod.rs @@ -4,6 +4,7 @@ pub mod browser_agent; pub mod desktop_agent; pub mod system_agent; pub mod orchestrator; +pub mod session; // Re-export key types for easier use pub use base_agent::*; @@ -12,3 +13,6 @@ pub use browser_agent::BrowserAgent; pub use desktop_agent::DesktopAgent; pub use system_agent::SystemAgent; pub use orchestrator::{Orchestrator, OrchestratorConfig}; +pub use session::{ + AgentSession, AgentSessionId, AgentSessionInfo, AgentSessionRegistry, AgentSessionStatus, +}; diff --git a/src-tauri/src/agents/session.rs b/src-tauri/src/agents/session.rs new file mode 100644 index 000000000..232e14ca7 --- /dev/null +++ b/src-tauri/src/agents/session.rs @@ -0,0 +1,423 @@ +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; + +use serde::{Deserialize, Serialize}; +use tokio::sync::{watch, Mutex as TokioMutex}; +use tracing::{debug, info, warn}; +use uuid::Uuid; + +use crate::agent::input_arbiter::InputArbiter; + +/// 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 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, + Finished, + 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 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, + display_color: String, + 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, + display_color: String, + ) -> Arc { + let (cancel_tx, cancel_rx) = watch::channel(false); + let started_at_ms = now_ms(); + Arc::new(Self { + id, + agent_name, + display_color, + 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 display_color(&self) -> &str { + &self.display_color + } + + /// 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(); + } + + 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(), + display_color: self.display_color.clone(), + 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. +pub struct AgentSessionRegistry { + sessions: TokioMutex>>, + focused: TokioMutex>, + 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: TokioMutex::new(None), + 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. + /// + /// 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, + display_color: 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 id = AgentSessionId::new(); + let session = AgentSession::new_with_id(id.clone(), agent_name, display_color); + 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().await; + if focused.is_none() { + *focused = Some(id.clone()); + } + info!( + "Registered agent session {} (focused={})", + id, + focused.as_ref().map(|f| f == &id).unwrap_or(false) + ); + 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; + if sessions.remove(id).is_some() { + debug!("Removed agent session {} from registry", id); + } + drop(sessions); + + let mut focused = self.focused.lock().await; + if focused.as_ref() == Some(id) { + *focused = None; + let sessions = self.sessions.lock().await; + if let Some(next) = sessions.keys().next().cloned() { + *focused = Some(next); + } + } + } + + 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().await.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().await = id; + Ok(()) + } + + pub async fn focused(&self) -> Option { + self.focused.lock().await.clone() + } + + /// List a snapshot of every session for the switcher/status-bar UI. + pub async fn list(&self) -> Vec { + let focused = self.focused.lock().await.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(|a, b| a.started_at_ms.cmp(&b.started_at_ms)); + out + } + + pub async fn len(&self) -> usize { + self.sessions.lock().await.len() + } +} + +#[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(), "#ff00aa".into()) + .await + .expect("first session created"); + let b = registry + .create("browser".into(), "#00aaff".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().await.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(), "#111111".into()) + .await + .expect("first ok"); + let result = registry.create("second".into(), "#222222".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(), "#f".into()) + .await + .expect("focused created"); + let background = registry + .create("background".into(), "#b".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(), "#a".into()).await.unwrap(); + let b = registry.create("b".into(), "#b".into()).await.unwrap(); + + assert_eq!(registry.focused().await.as_ref(), Some(a.id())); + registry.remove(a.id()).await; + // Focus falls back to the remaining session. + assert_eq!(registry.focused().await.as_ref(), Some(b.id())); + + registry.remove(b.id()).await; + assert!(registry.focused().await.is_none()); + assert_eq!(registry.len().await, 0); + } + + #[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()); + } +} diff --git a/src-tauri/src/state.rs b/src-tauri/src/state.rs index 898d65ecf..f492c2ba2 100644 --- a/src-tauri/src/state.rs +++ b/src-tauri/src/state.rs @@ -31,6 +31,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, events, errors::templates}; use crate::utils::string_cache::format_error_cached; use crate::utils::rate_limiter::GlobalRateLimiters; @@ -271,6 +274,13 @@ 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, + // Dynamic storage for other state components state_components: Arc>>>, } @@ -345,11 +355,33 @@ impl AppState { // Use the rate limiters created above rate_limiters, + // Initialize parallel-agent session registry. The 500ms cooldown + // matches the existing UI-action cooldown used by + // anthropic_computer_use.rs; the parallel cap mirrors the + // orchestrator's max_parallel_tasks so we don't outrun the + // higher-level scheduler. + agent_sessions: Arc::new(AgentSessionRegistry::new( + 12, + Arc::new(InputArbiter::new(Duration::from_millis(500))), + )), + // 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"); From d891dc2b64ee835a518d40639982120eb9df1b10 Mon Sep 17 00:00:00 2001 From: Lacy Morrow Date: Tue, 21 Jul 2026 09:18:58 -0400 Subject: [PATCH 02/12] feat(LAC-1432): expose parallel agent-session registry to frontend Wire the salvaged session registry + input arbiter scaffolding (commit 4d12e537) into a real Tauri command surface so the frontend session-switcher and status bar can render, focus, and cancel parallel agent sessions without polling. - `commands/agent_sessions.rs`: list_agent_sessions, get_focused_agent_session, focus_agent_session, cancel_focused_agent_session, cancel_agent_session - New event constants `agent-sessions-updated` (full snapshot on every mutation) and `agent-session-focused` (focus change payload so cursor overlays can react without diffing the list) - Register commands module + wire the five commands into `generate_handler!` - Scaffolding + commands compile clean under `cargo check`; 5/5 session-registry tests and 2/2 command tests pass Follow-up on this branch will: - Wire session lifecycle into `submit_query` / orchestrator so sessions are created and removed as agents start and finish - Ship the frontend switcher + per-session cursor overlay windows Co-Authored-By: Paperclip --- src-tauri/src/commands/agent_sessions.rs | 170 +++++++++++++++++++++++ src-tauri/src/commands/mod.rs | 2 + src-tauri/src/constants/events.rs | 16 +++ src-tauri/src/lib.rs | 6 + 4 files changed, 194 insertions(+) create mode 100644 src-tauri/src/commands/agent_sessions.rs diff --git a/src-tauri/src/commands/agent_sessions.rs b/src-tauri/src/commands/agent_sessions.rs new file mode 100644 index 000000000..89766fe5d --- /dev/null +++ b/src-tauri/src/commands/agent_sessions.rs @@ -0,0 +1,170 @@ +//! 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() + .await + .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 86647a476..acb20ff59 100644 --- a/src-tauri/src/commands/mod.rs +++ b/src-tauri/src/commands/mod.rs @@ -19,6 +19,7 @@ pub mod dev; pub mod dictation; // 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; @@ -75,6 +76,7 @@ pub use self::core::*; // 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/constants/events.rs b/src-tauri/src/constants/events.rs index 67b67e03f..aa674f8f8 100644 --- a/src-tauri/src/constants/events.rs +++ b/src-tauri/src/constants/events.rs @@ -28,6 +28,22 @@ 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 (`Vec` payload). + pub const UPDATED: &str = "agent-sessions-updated"; + /// Focus changed. Payload is `{ "session_id": Option }`. + pub const FOCUSED: &str = "agent-session-focused"; +} + /// Streaming events pub mod streaming { pub const TEXT_STREAM: &str = "agent-text-stream"; diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 8a0be7c71..74666d2bb 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -297,6 +297,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, From 6fc7d6060423b08e9800392fe9bb2e2b22e4e12c Mon Sep 17 00:00:00 2001 From: Lacy Morrow Date: Tue, 21 Jul 2026 09:49:51 -0400 Subject: [PATCH 03/12] feat(LAC-1432): register agent runs in parallel-session registry Hook execute_agent_internal into the AgentSessionRegistry via a RAII SessionHandle so every agent run appears in list_agent_sessions and can be targeted by focus_agent_session / cancel_agent_session commands. - Add next_session_color() round-robin picker (8 distinct colors) - Add SessionHandle: RAII guard that removes the session on drop and broadcasts agent-sessions-updated so the switcher UI drops the row on any exit path (success, error, panic unwind) - Add broadcast_sessions_updated() helper for callers without AppState - Wire execute_agent_internal to create + track a session per run Sessions today are 1:1 with agent runs because the execution queue still serializes; lifting that cap is separate work and unblocks true parallel execution. This delta makes the registry surface authoritative for the frontend switcher. Co-Authored-By: Paperclip --- src-tauri/src/agents/mod.rs | 3 +- src-tauri/src/agents/session.rs | 100 ++++++++++++++++++++++++++++++++ src-tauri/src/anthropic.rs | 32 ++++++++++ 3 files changed, 134 insertions(+), 1 deletion(-) diff --git a/src-tauri/src/agents/mod.rs b/src-tauri/src/agents/mod.rs index e9655d9a2..2c967e081 100644 --- a/src-tauri/src/agents/mod.rs +++ b/src-tauri/src/agents/mod.rs @@ -14,5 +14,6 @@ pub use desktop_agent::DesktopAgent; pub use system_agent::SystemAgent; pub use orchestrator::{Orchestrator, OrchestratorConfig}; pub use session::{ - AgentSession, AgentSessionId, AgentSessionInfo, AgentSessionRegistry, AgentSessionStatus, + broadcast_sessions_updated, next_session_color, AgentSession, AgentSessionId, AgentSessionInfo, + AgentSessionRegistry, AgentSessionStatus, SessionHandle, }; diff --git a/src-tauri/src/agents/session.rs b/src-tauri/src/agents/session.rs index 232e14ca7..7ec581567 100644 --- a/src-tauri/src/agents/session.rs +++ b/src-tauri/src/agents/session.rs @@ -1,13 +1,42 @@ use std::collections::HashMap; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; 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; + +/// Distinct display colors for parallel agent sessions. +/// +/// Assigned round-robin as sessions are created so each session's cursor +/// overlay and switcher row is visually distinguishable. Kept small and +/// perceptually distinct — sequential runs of the same color are only a +/// concern if the user exceeds the parallel cap (typically 12), and even +/// then the collision is not a correctness issue. +const SESSION_COLORS: &[&str] = &[ + "#ff5c8a", // rose + "#5cc8ff", // cyan + "#ffb45c", // amber + "#5cff9d", // mint + "#c85cff", // violet + "#ffe45c", // yellow + "#5c7dff", // indigo + "#ff5c5c", // red +]; + +static NEXT_COLOR_INDEX: AtomicUsize = AtomicUsize::new(0); + +/// Pick the next round-robin session color. +pub fn next_session_color() -> String { + let i = NEXT_COLOR_INDEX.fetch_add(1, Ordering::Relaxed); + SESSION_COLORS[i % SESSION_COLORS.len()].to_string() +} /// Unique identifier for a parallel agent session. /// @@ -323,6 +352,77 @@ impl AgentSessionRegistry { } } +/// 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 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. + pub async fn mark_terminal(&self, status: AgentSessionStatus) { + self.session.set_status(status).await; + broadcast_sessions_updated(&self.app_handle, &self.registry).await; + } +} + +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 { + registry.remove(&session_id).await; + broadcast_sessions_updated(&app_handle, ®istry).await; + }); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src-tauri/src/anthropic.rs b/src-tauri/src/anthropic.rs index a8b3d56e1..18e0a56cf 100644 --- a/src-tauri/src/anthropic.rs +++ b/src-tauri/src/anthropic.rs @@ -362,6 +362,38 @@ 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(); + let color = crate::agents::next_session_color(); + match registry + .create("orchestrator".to_string(), color.clone()) + .await + { + Ok(session) => { + session.set_status(crate::agents::AgentSessionStatus::Running).await; + let handle = crate::agents::SessionHandle::new( + registry.clone(), + session, + app_handle.clone(), + ); + 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 + } + } + }; + // TODO: TARS Integration disabled - event system not yet implemented // let agent_run_start_event = JunoAgentEvent::AgentRunStart { // session_id: execution_id.clone(), From b54eac106928fd0a9dca707760cc8413759839a2 Mon Sep 17 00:00:00 2001 From: Lacy Morrow Date: Fri, 24 Jul 2026 04:30:29 -0400 Subject: [PATCH 04/12] feat(LAC-1432): agent session switcher UI + fix dropped event constant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - useAgentSessions hook: initial snapshot via list_agent_sessions, then event-driven sync from agent-sessions-updated (no polling) - AgentSessionSwitcher: pill strip above chat input showing each live session's cursor color, name, and current action; click focuses, X cancels one session without disturbing the rest - events.rs: strip curly braces from FOCUSED doc comment — the TS constants codegen silently dropped AGENT_SESSIONS_FOCUSED after it; regenerated constants now include both session events Co-Authored-By: Claude Fable 5 --- src-tauri/src/constants/events.rs | 4 +- src/App.tsx | 9 ++ src/components/AgentSessionSwitcher.tsx | 91 +++++++++++++++++++ .../__tests__/AgentSessionSwitcher.test.tsx | 73 +++++++++++++++ src/hooks/useAgentSessions.ts | 72 +++++++++++++++ src/lib/constants.generated.ts | 2 + 6 files changed, 250 insertions(+), 1 deletion(-) create mode 100644 src/components/AgentSessionSwitcher.tsx create mode 100644 src/components/__tests__/AgentSessionSwitcher.test.tsx create mode 100644 src/hooks/useAgentSessions.ts diff --git a/src-tauri/src/constants/events.rs b/src-tauri/src/constants/events.rs index c7ecd37eb..c0e449cae 100644 --- a/src-tauri/src/constants/events.rs +++ b/src-tauri/src/constants/events.rs @@ -40,7 +40,9 @@ pub mod agent { pub mod agent_sessions { /// Full snapshot of all live sessions (`Vec` payload). pub const UPDATED: &str = "agent-sessions-updated"; - /// Focus changed. Payload is `{ "session_id": Option }`. + /// Focus changed. Payload has a nullable `session_id` string field. + /// NOTE: no curly braces in doc comments here — generate-ts-constants.js + /// silently drops constants that follow one. pub const FOCUSED: &str = "agent-session-focused"; } 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} /> + +
void; + onCancel: (sessionId: string) => void; + className?: string; +} + +const STATUS_LABELS: Record = { + starting: "Starting", + running: "Working", + needs_input: "Needs input", + cancelling: "Stopping", + finished: "Done", + failed: "Failed", +}; + +const cancellableStatuses: AgentSessionStatus[] = ["starting", "running", "needs_input"]; + +/** + * Horizontal strip of live agent sessions (LAC-1432 parallel agents). + * + * Each pill shows the session's cursor color, name, and current action. + * Clicking a pill focuses that session (escape then cancels it, and its + * cursor overlay highlights); the X cancels just that session while the + * others keep running. Renders nothing when no sessions are live. + */ +export function AgentSessionSwitcher({ + sessions, + onFocus, + onCancel, + className, +}: AgentSessionSwitcherProps) { + if (sessions.length === 0) return null; + + return ( +
+ {sessions.map((session) => { + const detail = + session.status === "running" && session.current_action + ? session.current_action + : STATUS_LABELS[session.status]; + return ( +
+ + {cancellableStatuses.includes(session.status) && ( + + )} +
+ ); + })} +
+ ); +} diff --git a/src/components/__tests__/AgentSessionSwitcher.test.tsx b/src/components/__tests__/AgentSessionSwitcher.test.tsx new file mode 100644 index 000000000..d4d3229c5 --- /dev/null +++ b/src/components/__tests__/AgentSessionSwitcher.test.tsx @@ -0,0 +1,73 @@ +import { render, screen, fireEvent } from "@testing-library/react"; +import { describe, expect, it, vi } from "vitest"; +import { AgentSessionSwitcher } from "../AgentSessionSwitcher"; +import type { AgentSessionInfo } from "@/hooks/useAgentSessions"; + +const makeSession = (overrides: Partial = {}): AgentSessionInfo => ({ + id: "session-1", + agent_name: "orchestrator", + display_color: "#22c55e", + status: "running", + current_action: "Clicking Submit button", + started_at_ms: 1000, + last_activity_ms: 2000, + focused: false, + ...overrides, +}); + +describe("AgentSessionSwitcher", () => { + it("renders nothing when no sessions are live", () => { + const { container } = render( + , + ); + expect(container).toBeEmptyDOMElement(); + }); + + it("shows each session with its name and current action", () => { + const sessions = [ + makeSession(), + makeSession({ + id: "session-2", + agent_name: "browser", + status: "needs_input", + current_action: null, + }), + ]; + render(); + + expect(screen.getByText("orchestrator")).toBeInTheDocument(); + expect(screen.getByText("Clicking Submit button")).toBeInTheDocument(); + expect(screen.getByText("browser")).toBeInTheDocument(); + expect(screen.getByText("Needs input")).toBeInTheDocument(); + }); + + it("focuses a session when its pill is clicked", () => { + const onFocus = vi.fn(); + render( + , + ); + fireEvent.click(screen.getByRole("tab")); + expect(onFocus).toHaveBeenCalledWith("session-1"); + }); + + it("cancels a session via its stop button", () => { + const onCancel = vi.fn(); + render( + , + ); + fireEvent.click(screen.getByRole("button", { name: "Stop orchestrator" })); + expect(onCancel).toHaveBeenCalledWith("session-1"); + }); + + it("hides the stop button for sessions already terminal or cancelling", () => { + render( + , + ); + expect(screen.queryByRole("button", { name: "Stop orchestrator" })).not.toBeInTheDocument(); + expect(screen.getByText("Stopping")).toBeInTheDocument(); + }); +}); diff --git a/src/hooks/useAgentSessions.ts b/src/hooks/useAgentSessions.ts new file mode 100644 index 000000000..b05f46b0e --- /dev/null +++ b/src/hooks/useAgentSessions.ts @@ -0,0 +1,72 @@ +import { useCallback, useEffect, useState } from "react"; +import { invoke } from "@tauri-apps/api/core"; +import { useEventListener } from "@/hooks/useEventListener"; +import { EVENTS } from "@/lib/constants.generated"; + +/** Mirrors `AgentSessionStatus` in src-tauri/src/agents/session.rs (snake_case serde). */ +export type AgentSessionStatus = + | "starting" + | "running" + | "needs_input" + | "cancelling" + | "finished" + | "failed"; + +/** Mirrors `AgentSessionInfo` in src-tauri/src/agents/session.rs. */ +export interface AgentSessionInfo { + id: string; + agent_name: string; + display_color: string; + status: AgentSessionStatus; + current_action: string | null; + started_at_ms: number; + last_activity_ms: number; + focused: boolean; +} + +/** + * Live view of the backend's parallel-agent session registry (LAC-1432). + * + * Loads the initial snapshot via `list_agent_sessions`, then stays in sync + * through `agent-sessions-updated` events — the backend broadcasts the full + * session list on every registry mutation, so no polling or diffing is needed. + */ +export function useAgentSessions() { + const [sessions, setSessions] = useState([]); + + useEffect(() => { + let mounted = true; + invoke("list_agent_sessions") + .then((snapshot) => { + if (mounted) setSessions(snapshot); + }) + .catch((error) => { + console.error("Failed to load agent sessions:", error); + }); + return () => { + mounted = false; + }; + }, []); + + useEventListener(EVENTS.AGENT_SESSIONS_UPDATED, setSessions); + + const focusSession = useCallback(async (sessionId: string | null) => { + try { + await invoke("focus_agent_session", { sessionId }); + } catch (error) { + console.error("Failed to focus agent session:", error); + } + }, []); + + const cancelSession = useCallback(async (sessionId: string) => { + try { + await invoke("cancel_agent_session", { sessionId }); + } catch (error) { + console.error("Failed to cancel agent session:", error); + } + }, []); + + const focusedSession = sessions.find((session) => session.focused) ?? null; + + return { sessions, focusedSession, focusSession, cancelSession }; +} diff --git a/src/lib/constants.generated.ts b/src/lib/constants.generated.ts index 5f26de5c0..bb35b2a79 100644 --- a/src/lib/constants.generated.ts +++ b/src/lib/constants.generated.ts @@ -21,6 +21,8 @@ export const EVENTS = { AGENT_FORCE_STOP: 'agent-force-stop', AGENT_FORCE_CLEANUP: 'agent-force-cleanup', AGENT_QUERY_READY: 'agent-query-ready', + AGENT_SESSIONS_UPDATED: 'agent-sessions-updated', + AGENT_SESSIONS_FOCUSED: 'agent-session-focused', STREAMING_TEXT_STREAM: 'agent-text-stream', STREAMING_STREAM_START: 'agent-stream-start', STREAMING_STREAM_END: 'agent-stream-end', From 2ed1069bdf49fcf00ec405d0ea840f8f01363d39 Mon Sep 17 00:00:00 2001 From: Lacy Morrow Date: Fri, 24 Jul 2026 04:34:36 -0400 Subject: [PATCH 05/12] =?UTF-8?q?refactor(LAC-1432):=20address=20Gemini=20?= =?UTF-8?q?code=20review=20=E2=80=94=20sync=20mutex=20for=20focused,=20opt?= =?UTF-8?q?imize=20remove,=20DEFAULT=5FCOOLDOWN?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace `TokioMutex>` on `focused` with `std::sync::Mutex` — no async work is done while this lock is held, so the lighter-weight sync variant is appropriate and avoids unnecessary async scheduler overhead - `focused()` is now a sync fn (no `.await` needed at call sites) - Optimize `AgentSessionRegistry::remove()`: capture `next_id` from the sessions map before dropping it, eliminating the second `self.sessions.lock().await` acquire inside the focused critical section (was: drop sessions → lock focused → re-lock sessions) - Simplify auto-focus logic in `create()`: use a local bool instead of re-borrowing the guard after mutation - Add `InputArbiter::DEFAULT_COOLDOWN` public constant (500 ms) so callers share one canonical default instead of repeating the literal Co-Authored-By: Claude Sonnet 4.6 --- src-tauri/src/agent/input_arbiter.rs | 8 ++++ src-tauri/src/agents/session.rs | 50 ++++++++++++------------ src-tauri/src/commands/agent_sessions.rs | 1 - 3 files changed, 33 insertions(+), 26 deletions(-) diff --git a/src-tauri/src/agent/input_arbiter.rs b/src-tauri/src/agent/input_arbiter.rs index 586b89a99..28b0fa430 100644 --- a/src-tauri/src/agent/input_arbiter.rs +++ b/src-tauri/src/agent/input_arbiter.rs @@ -16,6 +16,14 @@ use tracing::{debug, trace}; /// 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>, cooldown: Duration, diff --git a/src-tauri/src/agents/session.rs b/src-tauri/src/agents/session.rs index 7ec581567..487b10098 100644 --- a/src-tauri/src/agents/session.rs +++ b/src-tauri/src/agents/session.rs @@ -1,5 +1,5 @@ use std::collections::HashMap; -use std::sync::Arc; +use std::sync::{Arc, Mutex as StdMutex}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::{SystemTime, UNIX_EPOCH}; @@ -214,9 +214,13 @@ impl AgentSession { /// 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: TokioMutex>, + focused: StdMutex>, input_arbiter: Arc, max_parallel: usize, } @@ -225,7 +229,7 @@ impl AgentSessionRegistry { pub fn new(max_parallel: usize, input_arbiter: Arc) -> Self { Self { sessions: TokioMutex::new(HashMap::new()), - focused: TokioMutex::new(None), + focused: StdMutex::new(None), input_arbiter, max_parallel, } @@ -261,15 +265,12 @@ impl AgentSessionRegistry { drop(sessions); // Auto-focus the first session so escape has an obvious target. - let mut focused = self.focused.lock().await; - if focused.is_none() { + 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, - focused.as_ref().map(|f| f == &id).unwrap_or(false) - ); + info!("Registered agent session {} (focused={})", id, is_focused); Ok(session) } @@ -278,19 +279,18 @@ impl AgentSessionRegistry { } pub async fn remove(&self, id: &AgentSessionId) { + // Capture the next candidate before releasing the sessions lock so we + // don't need to re-acquire it inside the focused critical section. let mut sessions = self.sessions.lock().await; if sessions.remove(id).is_some() { debug!("Removed agent session {} from registry", id); } + let next_id = sessions.keys().next().cloned(); drop(sessions); - let mut focused = self.focused.lock().await; + let mut focused = self.focused.lock().unwrap_or_else(|e| e.into_inner()); if focused.as_ref() == Some(id) { - *focused = None; - let sessions = self.sessions.lock().await; - if let Some(next) = sessions.keys().next().cloned() { - *focused = Some(next); - } + *focused = next_id; } } @@ -309,7 +309,7 @@ impl AgentSessionRegistry { /// 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().await.clone(); + let focused = self.focused.lock().unwrap_or_else(|e| e.into_inner()).clone(); match focused { Some(id) => { self.cancel(&id).await?; @@ -326,17 +326,17 @@ impl AgentSessionRegistry { return Err(format!("Cannot focus unknown session {}", candidate)); } } - *self.focused.lock().await = id; + *self.focused.lock().unwrap_or_else(|e| e.into_inner()) = id; Ok(()) } - pub async fn focused(&self) -> Option { - self.focused.lock().await.clone() + pub fn focused(&self) -> Option { + self.focused.lock().unwrap_or_else(|e| e.into_inner()).clone() } /// List a snapshot of every session for the switcher/status-bar UI. pub async fn list(&self) -> Vec { - let focused = self.focused.lock().await.clone(); + 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() { @@ -450,7 +450,7 @@ mod tests { assert!(listed.iter().any(|s| s.id == b.id().to_string())); // First session auto-focused; snapshot exposes it. - let focused_id = registry.focused().await.expect("focused set"); + 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())); } @@ -504,13 +504,13 @@ mod tests { let a = registry.create("a".into(), "#a".into()).await.unwrap(); let b = registry.create("b".into(), "#b".into()).await.unwrap(); - assert_eq!(registry.focused().await.as_ref(), Some(a.id())); + 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().await.as_ref(), Some(b.id())); + assert_eq!(registry.focused().as_ref(), Some(b.id())); registry.remove(b.id()).await; - assert!(registry.focused().await.is_none()); + assert!(registry.focused().is_none()); assert_eq!(registry.len().await, 0); } diff --git a/src-tauri/src/commands/agent_sessions.rs b/src-tauri/src/commands/agent_sessions.rs index 89766fe5d..ee4815c0f 100644 --- a/src-tauri/src/commands/agent_sessions.rs +++ b/src-tauri/src/commands/agent_sessions.rs @@ -71,7 +71,6 @@ pub async fn get_focused_agent_session( Ok(state .agent_sessions() .focused() - .await .map(|id| id.to_string())) } From 116d0d59dd2ac10f1b671de6a7ee1d98c90c80a5 Mon Sep 17 00:00:00 2001 From: Lacy Morrow Date: Fri, 24 Jul 2026 16:48:11 -0400 Subject: [PATCH 06/12] =?UTF-8?q?feat(LAC-1432):=20parallel=20sessions=20?= =?UTF-8?q?=E2=80=94=20submit=5Fquery=20wiring,=20arbiter=20routing,=20ses?= =?UTF-8?q?sion=20cursors,=20roster=20UI,=20notifications?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backend: - submit_query runs register in the session registry; merged global+session cancel channel gives per-session cancellation isolation; terminal states (finished/cancelled/failed) marked + lifecycle events emitted before RAII cleanup removes the row - Physical CGEvent input serialized through the InputArbiter: always-physical actions guard at dispatch, AX-attempt actions guard only in their physical fallback blocks so AX-grounded actions stay parallel across sessions - Cursor overlay keyed by session id + LAC-2830 identity palette (8 slots, round-robin with freed-slot reuse); SessionHandle::drop clears the cursor on every end path (complete, cancel, error, panic) - Escape/stop cancels only the FOCUSED session when parallel sessions exist; global cancel remains the legacy/headless fallback - Background (non-focused) sessions fire macOS notifications on completion/ failure via send_notification (respects user notification settings) Frontend (display-only): - AgentRosterStrip: identity dots below floating bar when 2+ agents run, status badges, completion-pulse/error-shake/needs-input-blink animations, +N overflow, tooltips, tablist a11y; bar window grows to fit - AgentSessionRows: named rows w/ current action + status icon in floating panel expanded mode; click to focus without pausing background sessions - Palette + lifecycle event constants generated from Rust (ui.rs/events.rs) Tests: session registry slot allocator + cancel isolation (Rust), 6 new AgentRosterStrip tests (Vitest). cargo check clean, tsc clean, 47/47 FE tests pass. Co-Authored-By: Claude Fable 5 --- src-tauri/src/agent/providers/factory.rs | 24 ++- .../src/agent/tools/anthropic_computer_use.rs | 119 ++++++++++- src-tauri/src/agents/desktop_agent.rs | 1 + src-tauri/src/agents/mod.rs | 4 +- src-tauri/src/agents/session.rs | 202 +++++++++++++----- src-tauri/src/anthropic.rs | 118 +++++++++- src-tauri/src/commands/stop_coordinator.rs | 23 +- src-tauri/src/constants/events.rs | 19 +- src-tauri/src/constants/ui.rs | 17 ++ src/FloatingPanel.tsx | 23 +- src/components/AgentRosterStrip.tsx | 141 ++++++++++++ src/components/AgentSessionRows.tsx | 97 +++++++++ src/components/AgentSessionSwitcher.tsx | 1 + src/components/FloatingBar.tsx | 26 ++- src/components/TransparentFloatingPanel.tsx | 18 +- .../__tests__/AgentRosterStrip.test.tsx | 74 +++++++ .../__tests__/AgentSessionSwitcher.test.tsx | 1 + src/hooks/useAgentSessions.ts | 3 + src/lib/constants.generated.ts | 13 ++ src/styles/globals.css | 56 +++++ 20 files changed, 888 insertions(+), 92 deletions(-) create mode 100644 src/components/AgentRosterStrip.tsx create mode 100644 src/components/AgentSessionRows.tsx create mode 100644 src/components/__tests__/AgentRosterStrip.test.tsx diff --git a/src-tauri/src/agent/providers/factory.rs b/src-tauri/src/agent/providers/factory.rs index bbf22a47e..00d65f3c3 100644 --- a/src-tauri/src/agent/providers/factory.rs +++ b/src-tauri/src/agent/providers/factory.rs @@ -440,6 +440,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)..."); @@ -465,7 +477,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 989acab69..e4ba2e3fb 100644 --- a/src-tauri/src/agent/tools/anthropic_computer_use.rs +++ b/src-tauri/src/agent/tools/anthropic_computer_use.rs @@ -49,6 +49,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, @@ -73,10 +85,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); @@ -910,9 +925,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, @@ -927,6 +948,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, @@ -939,6 +973,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) { @@ -1038,6 +1096,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( @@ -1082,6 +1142,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.right_click_no_warp(screen_x, screen_y); match click_method { @@ -1135,6 +1197,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, @@ -1342,6 +1406,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(), @@ -1943,7 +2010,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 @@ -1951,6 +2032,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() @@ -1959,11 +2041,22 @@ pub async fn register_anthropic_computer_use_tools_with_version( info!("Registering official Anthropic Computer Use tools (API 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: {})", agent_cursor_id, agent_cursor_color); @@ -1978,12 +2071,16 @@ 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 db00ee603..b4cad2c70 100644 --- a/src-tauri/src/agents/desktop_agent.rs +++ b/src-tauri/src/agents/desktop_agent.rs @@ -54,6 +54,7 @@ impl DesktopAgent { match crate::agent::tools::anthropic_computer_use::execute_computer_tool( &self.app_handle, tool_call.input.clone(), + None, ).await { Ok(result) => Ok(ToolResult { call_id: tool_call.id.clone(), diff --git a/src-tauri/src/agents/mod.rs b/src-tauri/src/agents/mod.rs index 2c967e081..c349d4bbd 100644 --- a/src-tauri/src/agents/mod.rs +++ b/src-tauri/src/agents/mod.rs @@ -14,6 +14,6 @@ pub use desktop_agent::DesktopAgent; pub use system_agent::SystemAgent; pub use orchestrator::{Orchestrator, OrchestratorConfig}; pub use session::{ - broadcast_sessions_updated, next_session_color, AgentSession, AgentSessionId, AgentSessionInfo, - AgentSessionRegistry, AgentSessionStatus, SessionHandle, + broadcast_sessions_updated, color_for_slot, AgentSession, AgentSessionId, AgentSessionInfo, + AgentSessionRegistry, AgentSessionStatus, SessionHandle, SESSION_COLOR_SLOTS, }; diff --git a/src-tauri/src/agents/session.rs b/src-tauri/src/agents/session.rs index 487b10098..dd824be9b 100644 --- a/src-tauri/src/agents/session.rs +++ b/src-tauri/src/agents/session.rs @@ -1,6 +1,5 @@ -use std::collections::HashMap; +use std::collections::{BTreeSet, HashMap}; use std::sync::{Arc, Mutex as StdMutex}; -use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::{SystemTime, UNIX_EPOCH}; use serde::{Deserialize, Serialize}; @@ -11,31 +10,53 @@ 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, +]; -/// Distinct display colors for parallel agent sessions. +/// Round-robin color slot allocator with slot reuse. /// -/// Assigned round-robin as sessions are created so each session's cursor -/// overlay and switcher row is visually distinguishable. Kept small and -/// perceptually distinct — sequential runs of the same color are only a -/// concern if the user exceeds the parallel cap (typically 12), and even -/// then the collision is not a correctness issue. -const SESSION_COLORS: &[&str] = &[ - "#ff5c8a", // rose - "#5cc8ff", // cyan - "#ffb45c", // amber - "#5cff9d", // mint - "#c85cff", // violet - "#ffe45c", // yellow - "#5c7dff", // indigo - "#ff5c5c", // red -]; +/// 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, +} -static NEXT_COLOR_INDEX: AtomicUsize = AtomicUsize::new(0); +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); + } + } +} -/// Pick the next round-robin session color. -pub fn next_session_color() -> String { - let i = NEXT_COLOR_INDEX.fetch_add(1, Ordering::Relaxed); - SESSION_COLORS[i % SESSION_COLORS.len()].to_string() +/// 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. @@ -76,10 +97,18 @@ pub enum AgentSessionStatus { 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. @@ -89,6 +118,7 @@ pub enum AgentSessionStatus { 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, @@ -114,7 +144,7 @@ fn now_ms() -> u64 { pub struct AgentSession { id: AgentSessionId, agent_name: String, - display_color: String, + color_slot: u8, cancel_tx: watch::Sender, cancel_rx: watch::Receiver, started_at_ms: u64, @@ -128,17 +158,13 @@ struct AgentSessionInner { } impl AgentSession { - fn new_with_id( - id: AgentSessionId, - agent_name: String, - display_color: String, - ) -> Arc { + 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, - display_color, + color_slot, cancel_tx, cancel_rx, started_at_ms, @@ -158,8 +184,12 @@ impl AgentSession { &self.agent_name } - pub fn display_color(&self) -> &str { - &self.display_color + 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 @@ -199,7 +229,8 @@ impl AgentSession { AgentSessionInfo { id: self.id.0.clone(), agent_name: self.agent_name.clone(), - display_color: self.display_color.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, @@ -221,6 +252,7 @@ impl AgentSession { pub struct AgentSessionRegistry { sessions: TokioMutex>>, focused: StdMutex>, + colors: StdMutex, input_arbiter: Arc, max_parallel: usize, } @@ -230,6 +262,7 @@ impl AgentSessionRegistry { Self { sessions: TokioMutex::new(HashMap::new()), focused: StdMutex::new(None), + colors: StdMutex::new(ColorAllocator::default()), input_arbiter, max_parallel, } @@ -245,13 +278,11 @@ impl AgentSessionRegistry { /// Create a new session and register it. Fails if the parallel cap is hit. /// - /// 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, - display_color: String, - ) -> Result, String> { + /// 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!( @@ -259,8 +290,12 @@ impl AgentSessionRegistry { 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, display_color); + let session = AgentSession::new_with_id(id.clone(), agent_name, color_slot); sessions.insert(id.clone(), session.clone()); drop(sessions); @@ -282,12 +317,16 @@ impl AgentSessionRegistry { // Capture the next candidate before releasing the sessions lock so we // don't need to re-acquire it inside the focused critical section. let mut sessions = self.sessions.lock().await; - if sessions.remove(id).is_some() { - debug!("Removed agent session {} from registry", id); - } + let removed = sessions.remove(id); let next_id = 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()); + } + let mut focused = self.focused.lock().unwrap_or_else(|e| e.into_inner()); if focused.as_ref() == Some(id) { *focused = next_id; @@ -398,14 +437,47 @@ impl SessionHandle { &self.session } - /// Mark the session as finished/failed and broadcast the state before - /// the RAII cleanup removes the row entirely. Callers that know + /// 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. pub async fn mark_terminal(&self, status: AgentSessionStatus) { 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; } + + /// 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 { @@ -417,6 +489,7 @@ impl Drop for SessionHandle { 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; }); @@ -436,11 +509,11 @@ mod tests { async fn creates_and_lists_sessions() { let registry = AgentSessionRegistry::new(4, arbiter()); let a = registry - .create("desktop".into(), "#ff00aa".into()) + .create("desktop".into()) .await .expect("first session created"); let b = registry - .create("browser".into(), "#00aaff".into()) + .create("browser".into()) .await .expect("second session created"); @@ -459,10 +532,10 @@ mod tests { async fn enforces_parallel_cap() { let registry = AgentSessionRegistry::new(1, arbiter()); registry - .create("first".into(), "#111111".into()) + .create("first".into()) .await .expect("first ok"); - let result = registry.create("second".into(), "#222222".into()).await; + let result = registry.create("second".into()).await; let err = match result { Ok(_) => panic!("expected second create to fail"), Err(e) => e, @@ -474,11 +547,11 @@ mod tests { async fn cancel_focused_kills_only_focused_session() { let registry = AgentSessionRegistry::new(4, arbiter()); let focused_session = registry - .create("focused".into(), "#f".into()) + .create("focused".into()) .await .expect("focused created"); let background = registry - .create("background".into(), "#b".into()) + .create("background".into()) .await .expect("background created"); @@ -501,8 +574,8 @@ mod tests { #[tokio::test] async fn remove_clears_and_reassigns_focus() { let registry = AgentSessionRegistry::new(4, arbiter()); - let a = registry.create("a".into(), "#a".into()).await.unwrap(); - let b = registry.create("b".into(), "#b".into()).await.unwrap(); + 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; @@ -514,6 +587,27 @@ mod tests { 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()); diff --git a/src-tauri/src/anthropic.rs b/src-tauri/src/anthropic.rs index 293a8f919..ea1cf0f41 100644 --- a/src-tauri/src/anthropic.rs +++ b/src-tauri/src/anthropic.rs @@ -378,13 +378,9 @@ async fn execute_agent_internal( // 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 session_handle = { let registry = state.agent_sessions(); - let color = crate::agents::next_session_color(); - match registry - .create("orchestrator".to_string(), color.clone()) - .await - { + match registry.create("orchestrator".to_string()).await { Ok(session) => { session.set_status(crate::agents::AgentSessionStatus::Running).await; let handle = crate::agents::SessionHandle::new( @@ -392,6 +388,14 @@ async fn execute_agent_internal( 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) } @@ -405,6 +409,15 @@ async fn execute_agent_internal( } }; + // 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(), @@ -456,7 +469,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; @@ -600,9 +646,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 { @@ -764,9 +811,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 { @@ -996,6 +1044,58 @@ 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 — the focused session's outcome is already on + // screen, so notifying for it would be noise. Cancellations are always + // user-initiated, so they never notify. + if let Some(handle) = session_handle.as_ref() { + use crate::agents::AgentSessionStatus; + let terminal_status = match &agent_result { + Ok(_) => AgentSessionStatus::Finished, + Err(AgentError::Terminated) => AgentSessionStatus::Cancelled, + Err(_) => AgentSessionStatus::Failed, + }; + let was_focused = handle.is_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, message, level) = match &agent_result { + Ok(message) => ( + format!("{} — Complete", agent_name), + message.chars().take(140).collect::(), + "success".to_string(), + ), + Err(e) => ( + format!("{} — Failed", agent_name), + e.to_string().chars().take(140).collect::(), + "error".to_string(), + ), + }; + let data = crate::commands::notifications::NotificationData { + title, + message, + level, + 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); + } + } + } + // 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/stop_coordinator.rs b/src-tauri/src/commands/stop_coordinator.rs index 2a54af9f3..662a2aa0c 100644 --- a/src-tauri/src/commands/stop_coordinator.rs +++ b/src-tauri/src/commands/stop_coordinator.rs @@ -168,10 +168,25 @@ 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(); diff --git a/src-tauri/src/constants/events.rs b/src-tauri/src/constants/events.rs index c0e449cae..67d95b23f 100644 --- a/src-tauri/src/constants/events.rs +++ b/src-tauri/src/constants/events.rs @@ -38,12 +38,25 @@ pub mod agent { /// 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 (`Vec` payload). - pub const UPDATED: &str = "agent-sessions-updated"; - /// Focus changed. Payload has a nullable `session_id` string field. + /// 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. Reserved: no backend path triggers this yet. + pub const NEEDS_INPUT: &str = "agent-session-needs-input"; } /// Streaming events diff --git a/src-tauri/src/constants/ui.rs b/src-tauri/src/constants/ui.rs index 8959d6116..e624ff25d 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/FloatingPanel.tsx b/src/FloatingPanel.tsx index cadf05bba..bc41fe749 100644 --- a/src/FloatingPanel.tsx +++ b/src/FloatingPanel.tsx @@ -3,6 +3,7 @@ import { listen } from "@tauri-apps/api/event"; import { getCurrentWindow, LogicalSize } from "@tauri-apps/api/window"; import { useEffect, useState } from "react"; import { useDragWindow } from "@/hooks/useDragWindow"; +import { useAgentSessions } from "@/hooks/useAgentSessions"; import { UI } from "@/lib/constants.generated"; import "./styles/globals.css"; @@ -19,13 +20,23 @@ const SETTINGS_HEIGHT = 180 + 24; // 204 // Helper function to get window dimensions for each panel mode function getWindowDimensionsForMode( - mode: "compact" | "expanded" | "chat" | "settings" + mode: "compact" | "expanded" | "chat" | "settings", + agentSessionCount = 0 ) { switch (mode) { case "compact": return { width: COMPACT_WIDTH, height: COMPACT_HEIGHT }; case "expanded": - return { width: EXPANDED_WIDTH, height: EXPANDED_HEIGHT }; + // Grow with the agent summary rows (LAC-2830 §4) — mirrors the + // content growth inside TransparentFloatingPanel, same 160px cap. + return { + width: EXPANDED_WIDTH, + height: + EXPANDED_HEIGHT + + (agentSessionCount > 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