diff --git a/Cargo.lock b/Cargo.lock index 5cd44d9..28ded22 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1465,7 +1465,7 @@ dependencies = [ [[package]] name = "ras-agent" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "chrono", @@ -1497,7 +1497,7 @@ dependencies = [ [[package]] name = "ras-browser" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-cdp", @@ -1515,7 +1515,7 @@ dependencies = [ [[package]] name = "ras-cdp" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "chromiumoxide", @@ -1534,7 +1534,7 @@ dependencies = [ [[package]] name = "ras-cli" -version = "3.5.0" +version = "3.6.0" dependencies = [ "anyhow", "clap", @@ -1564,7 +1564,7 @@ dependencies = [ [[package]] name = "ras-cloud" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1581,7 +1581,7 @@ dependencies = [ [[package]] name = "ras-config" -version = "3.5.0" +version = "3.6.0" dependencies = [ "dotenvy", "once_cell", @@ -1595,7 +1595,7 @@ dependencies = [ [[package]] name = "ras-cosmium" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-cdp", @@ -1612,7 +1612,7 @@ dependencies = [ [[package]] name = "ras-daemon" -version = "3.5.0" +version = "3.6.0" dependencies = [ "anyhow", "dotenvy", @@ -1632,7 +1632,7 @@ dependencies = [ [[package]] name = "ras-dom" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "chromiumoxide", @@ -1652,7 +1652,7 @@ dependencies = [ [[package]] name = "ras-errors" -version = "3.5.0" +version = "3.6.0" dependencies = [ "serde", "thiserror", @@ -1660,7 +1660,7 @@ dependencies = [ [[package]] name = "ras-events" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-broadcast", "async-trait", @@ -1678,7 +1678,7 @@ dependencies = [ [[package]] name = "ras-filesystem" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1693,7 +1693,7 @@ dependencies = [ [[package]] name = "ras-judge" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "image", @@ -1708,7 +1708,7 @@ dependencies = [ [[package]] name = "ras-llm" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1722,7 +1722,7 @@ dependencies = [ [[package]] name = "ras-llm-anthropic" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "chrono", @@ -1745,7 +1745,7 @@ dependencies = [ [[package]] name = "ras-llm-bedrock" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1762,7 +1762,7 @@ dependencies = [ [[package]] name = "ras-llm-cerebras" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1780,7 +1780,7 @@ dependencies = [ [[package]] name = "ras-llm-cloud" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1797,7 +1797,7 @@ dependencies = [ [[package]] name = "ras-llm-deepseek" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1815,7 +1815,7 @@ dependencies = [ [[package]] name = "ras-llm-google" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1832,7 +1832,7 @@ dependencies = [ [[package]] name = "ras-llm-groq" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1850,7 +1850,7 @@ dependencies = [ [[package]] name = "ras-llm-langchain" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1867,7 +1867,7 @@ dependencies = [ [[package]] name = "ras-llm-mistral" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1885,7 +1885,7 @@ dependencies = [ [[package]] name = "ras-llm-oci" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1902,7 +1902,7 @@ dependencies = [ [[package]] name = "ras-llm-ollama" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1919,7 +1919,7 @@ dependencies = [ [[package]] name = "ras-llm-openai" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1936,7 +1936,7 @@ dependencies = [ [[package]] name = "ras-llm-openrouter" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1954,7 +1954,7 @@ dependencies = [ [[package]] name = "ras-llm-vercel" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1972,7 +1972,7 @@ dependencies = [ [[package]] name = "ras-mcp" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -1989,7 +1989,7 @@ dependencies = [ [[package]] name = "ras-recording" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "image", @@ -2004,7 +2004,7 @@ dependencies = [ [[package]] name = "ras-sandbox" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -2015,9 +2015,27 @@ dependencies = [ "tokio", ] +[[package]] +name = "ras-session" +version = "3.6.0" +dependencies = [ + "async-trait", + "ras-agent", + "ras-cdp", + "ras-dom", + "ras-errors", + "ras-events", + "ras-llm", + "ras-tools", + "ras-types", + "serde_json", + "tokio", + "url", +] + [[package]] name = "ras-skills" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -2034,7 +2052,7 @@ dependencies = [ [[package]] name = "ras-telemetry" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -2048,7 +2066,7 @@ dependencies = [ [[package]] name = "ras-tokens" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "chrono", @@ -2066,7 +2084,7 @@ dependencies = [ [[package]] name = "ras-tools" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "indexmap", @@ -2088,7 +2106,7 @@ dependencies = [ [[package]] name = "ras-types" -version = "3.5.0" +version = "3.6.0" dependencies = [ "chrono", "indexmap", @@ -2104,7 +2122,7 @@ dependencies = [ [[package]] name = "ras-validation" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-errors", @@ -2119,7 +2137,7 @@ dependencies = [ [[package]] name = "ras-watchdogs" -version = "3.5.0" +version = "3.6.0" dependencies = [ "async-trait", "ras-browser", diff --git a/Cargo.toml b/Cargo.toml index 2b31b31..88ee34d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,6 +32,7 @@ members = [ "ras-browser", "ras-dom", "ras-tools", + "ras-session", "ras-watchdogs", "ras-filesystem", "ras-tokens", @@ -51,7 +52,7 @@ exclude = ["examples", "tests", "docs", "scripts"] default-members = ["ras-cli", "ras-daemon"] [workspace.package] -version = "3.5.0" +version = "3.6.0" edition = "2024" rust-version = "1.95.0" license = "MIT" @@ -91,6 +92,7 @@ ras-llm-langchain = { path = "ras-llm-langchain", version = "3.0.0" } ras-browser = { path = "ras-browser", version = "3.0.0" } ras-dom = { path = "ras-dom", version = "3.0.0" } ras-tools = { path = "ras-tools", version = "3.0.0" } +ras-session = { path = "ras-session", version = "3.5.0" } ras-watchdogs = { path = "ras-watchdogs", version = "3.0.0" } ras-filesystem = { path = "ras-filesystem", version = "3.0.0" } ras-tokens = { path = "ras-tokens", version = "3.0.0" } diff --git a/ras-session/Cargo.toml b/ras-session/Cargo.toml new file mode 100644 index 0000000..3e927f0 --- /dev/null +++ b/ras-session/Cargo.toml @@ -0,0 +1,30 @@ +[package] +name = "ras-session" +description = "Tenant session lifecycle, config, and browser-provider abstraction" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +authors.workspace = true + +[lints] +workspace = true + +[dependencies] +ras-errors = { workspace = true } +ras-types = { workspace = true } +ras-cdp = { workspace = true } +ras-events = { workspace = true } +ras-agent = { workspace = true } +ras-tools = { workspace = true } +ras-llm = { workspace = true } +ras-dom = { workspace = true } +async-trait = { workspace = true } +tokio = { workspace = true } +url = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +serde_json = { workspace = true } +url = { workspace = true } diff --git a/ras-session/src/config.rs b/ras-session/src/config.rs new file mode 100644 index 0000000..020c016 --- /dev/null +++ b/ras-session/src/config.rs @@ -0,0 +1,120 @@ +//! Session configuration types and error definitions. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use ras_errors::AppError; + +/// Tenant session id (distinct from CDP `ras_types::SessionId`). +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct AgentSessionId(pub String); + +impl AgentSessionId { + /// Allocate a process-unique session id. + pub fn next() -> Self { + static COUNTER: AtomicU64 = AtomicU64::new(1); + Self(format!("sess-{}", COUNTER.fetch_add(1, Ordering::Relaxed))) + } +} + +/// Policy when the session pool is at capacity. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OnFull { + /// Reject new sessions with [`SessionError::AtCapacity`]. + Reject, + /// Evict the least-recently-used session to make room. + EvictOldest, +} + +/// Runtime configuration for the session manager. +#[derive(Debug, Clone)] +pub struct SessionConfig { + /// Maximum number of concurrent sessions. + pub max_sessions: usize, + /// Duration after which an idle session is reaped. + pub idle_timeout: Duration, + /// What to do when the pool is full. + pub on_full: OnFull, + /// Whether one owner may hold more than one active session. + pub allow_multi_per_owner: bool, +} + +impl Default for SessionConfig { + fn default() -> Self { + Self { + max_sessions: 100, + idle_timeout: Duration::from_secs(600), + on_full: OnFull::Reject, + allow_multi_per_owner: false, + } + } +} + +/// Errors produced by the session manager. +#[derive(Debug)] +pub enum SessionError { + /// Pool has reached `max_sessions` and the policy is `Reject`. + AtCapacity, + /// No session with the given id exists. + NotFound, + /// Session is currently executing a task and cannot be preempted. + Busy, + /// An underlying browser operation failed. + Browser(AppError), +} + +impl std::fmt::Display for SessionError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::AtCapacity => write!(f, "session manager at capacity"), + Self::NotFound => write!(f, "session not found"), + Self::Busy => write!(f, "session already running a task"), + Self::Browser(e) => write!(f, "browser error: {e}"), + } + } +} + +impl std::error::Error for SessionError {} + +impl From for SessionError { + fn from(e: AppError) -> Self { + Self::Browser(e) + } +} + +#[cfg(test)] +mod config_tests { + use super::{AgentSessionId, OnFull, SessionConfig, SessionError}; + use ras_errors::AppError; + + #[test] + fn agent_session_id_increments() { + let a = AgentSessionId::next(); + let b = AgentSessionId::next(); + assert!(a.0.starts_with("sess-")); + assert_ne!(a, b); + } + + #[test] + fn session_config_defaults_are_sane() { + let cfg = SessionConfig::default(); + assert_eq!(cfg.max_sessions, 100); + assert_eq!(cfg.on_full, OnFull::Reject); + assert!(!cfg.allow_multi_per_owner); + } + + #[test] + fn session_error_display_covers_variants() { + assert!(SessionError::AtCapacity.to_string().contains("capacity")); + assert!(SessionError::NotFound.to_string().contains("not found")); + assert!(SessionError::Busy.to_string().contains("running")); + let inner = AppError::ActionFailed("oops".into()); + assert!(SessionError::Browser(inner).to_string().contains("browser")); + } + + #[test] + fn session_error_from_app_error() { + let e: SessionError = AppError::ActionFailed("x".into()).into(); + assert!(matches!(e, SessionError::Browser(_))); + } +} diff --git a/ras-session/src/entry.rs b/ras-session/src/entry.rs new file mode 100644 index 0000000..29c4d77 --- /dev/null +++ b/ras-session/src/entry.rs @@ -0,0 +1,41 @@ +//! Shared per-session record and the owner-erasing registry trait. + +use std::sync::atomic::AtomicBool; +use std::sync::{Arc, Mutex}; +use std::time::Instant; + +use async_trait::async_trait; +use ras_cdp::BrowserPort; +use ras_events::EventBus; +use ras_types::{ContextId, TargetId}; + +use crate::config::AgentSessionId; +use crate::spawn_params::SpawnParams; + +/// Cloneable (via `Arc`) state for a single live session. +pub(crate) struct SessionEntry { + pub(crate) id: AgentSessionId, + pub(crate) ctx: ContextId, + pub(crate) tab: TargetId, + pub(crate) browser: Arc, + pub(crate) bus: Arc, + pub(crate) params: SpawnParams, + pub(crate) running: AtomicBool, + pub(crate) last_active: Mutex, +} + +impl SessionEntry { + /// Update `last_active` to now, ignoring a poisoned lock. + pub(crate) fn touch(&self) { + if let Ok(mut g) = self.last_active.lock() { + *g = Instant::now(); + } + } +} + +/// Owner-erasing removal hook so `SessionHandle` need not be generic. +#[async_trait] +pub(crate) trait SessionRegistry: Send + Sync { + /// Remove a session id from the manager's maps. + async fn forget(&self, id: &AgentSessionId); +} diff --git a/ras-session/src/handle.rs b/ras-session/src/handle.rs new file mode 100644 index 0000000..24fd2df --- /dev/null +++ b/ras-session/src/handle.rs @@ -0,0 +1,87 @@ +//! Non-generic handle to a single live session. + +use std::sync::Arc; +use std::sync::atomic::Ordering::SeqCst; + +use ras_agent::AgentHistoryList; +use ras_agent::application::run_agent::RunAgent; +use ras_events::EventReceiver; + +use crate::config::{AgentSessionId, SessionError}; +use crate::entry::{SessionEntry, SessionRegistry}; +use crate::provider::BrowserProvider; + +/// Cloneable handle that can run tasks, observe events, and close a session. +/// +/// Intentionally non-generic over `Owner`: removal goes through an +/// owner-erasing [`SessionRegistry`] trait object. +#[derive(Clone)] +pub struct SessionHandle { + entry: Arc, + provider: Arc, + registry: Arc, +} + +impl SessionHandle { + pub(crate) fn new( + entry: Arc, + provider: Arc, + registry: Arc, + ) -> Self { + Self { + entry, + provider, + registry, + } + } + + /// This session's id. + #[must_use] + pub fn id(&self) -> AgentSessionId { + self.entry.id.clone() + } + + /// Subscribe to this session's browser-event stream. + #[must_use] + pub fn events(&self) -> EventReceiver { + self.entry.bus.subscribe() + } + + /// Run an agent task on this session. + /// + /// Returns [`SessionError::Busy`] if a task is already running. + pub async fn run(&self, task: impl Into) -> Result { + if self.entry.running.swap(true, SeqCst) { + return Err(SessionError::Busy); + } + self.entry.touch(); + let result = self.execute(task.into()).await; + self.entry.running.store(false, SeqCst); + self.entry.touch(); + result.map_err(SessionError::Browser) + } + + async fn execute(&self, task: String) -> Result { + let p = &self.entry.params; + let mut agent = RunAgent::new( + task, + p.llm.clone(), + p.registry.clone(), + self.entry.browser.clone(), + self.entry.bus.clone(), + ) + .with_target(self.entry.tab.clone()) + .with_max_steps(p.max_steps); + if let Some(dom) = p.dom_extractor.clone() { + agent = agent.with_dom_extractor(dom); + } + agent.execute().await + } + + /// Release the browser context and forget this session. Idempotent. + pub async fn close(self) -> Result<(), SessionError> { + self.provider.release(&self.entry.ctx).await?; + self.registry.forget(&self.entry.id).await; + Ok(()) + } +} diff --git a/ras-session/src/lib.rs b/ras-session/src/lib.rs new file mode 100644 index 0000000..9e8204c --- /dev/null +++ b/ras-session/src/lib.rs @@ -0,0 +1,16 @@ +//! `ras-session` — tenant session lifecycle, config, and browser-provider abstraction. + +pub mod config; +pub mod provider; + +mod entry; +pub mod handle; +pub mod manager; +mod manager_reaper; +pub mod spawn_params; + +pub use config::{AgentSessionId, OnFull, SessionConfig, SessionError}; +pub use handle::SessionHandle; +pub use manager::SessionManager; +pub use provider::{BrowserProvider, SharedBrowserProvider}; +pub use spawn_params::SpawnParams; diff --git a/ras-session/src/manager.rs b/ras-session/src/manager.rs new file mode 100644 index 0000000..216be8a --- /dev/null +++ b/ras-session/src/manager.rs @@ -0,0 +1,144 @@ +//! `SessionManager`: spawns, tracks, and reaps tenant sessions. + +use std::collections::HashMap; +use std::hash::Hash; +use std::sync::atomic::AtomicBool; +use std::sync::{Arc, Mutex}; +use std::time::Instant; + +use async_trait::async_trait; +use ras_errors::AppError; +use ras_events::{BroadcastBus, EventBus}; +use tokio::sync::Mutex as AsyncMutex; +use url::Url; + +use crate::config::{AgentSessionId, OnFull, SessionConfig, SessionError}; +use crate::entry::{SessionEntry, SessionRegistry}; +use crate::handle::SessionHandle; +use crate::provider::BrowserProvider; +use crate::spawn_params::SpawnParams; + +pub(crate) struct Inner { + pub(crate) sessions: HashMap>, + pub(crate) by_owner: HashMap, +} + +/// Manages the pool of live tenant sessions for one `Owner` key type. +pub struct SessionManager { + provider: Arc, + cfg: SessionConfig, + inner: Arc>>, +} + +impl SessionManager +where + Owner: Eq + Hash + Clone + Send + Sync + 'static, +{ + /// Create a manager and launch its background idle-reaper task. + pub fn new(provider: Arc, cfg: SessionConfig) -> Self { + let inner = Arc::new(AsyncMutex::new(Inner { + sessions: HashMap::new(), + by_owner: HashMap::new(), + })); + crate::manager_reaper::spawn_reaper(inner.clone(), provider.clone(), cfg.idle_timeout); + Self { + provider, + cfg, + inner, + } + } + + fn handle_for(&self, entry: Arc) -> SessionHandle { + let registry: Arc = self.inner.clone(); + SessionHandle::new(entry, self.provider.clone(), registry) + } + + /// Spawn a new session for `owner`, honoring capacity and reuse policy. + pub async fn spawn( + &self, + owner: Owner, + params: SpawnParams, + ) -> Result { + let mut guard = self.inner.lock().await; + + if !self.cfg.allow_multi_per_owner + && let Some(existing) = guard.by_owner.get(&owner) + && let Some(entry) = guard.sessions.get(existing) + { + return Ok(self.handle_for(entry.clone())); + } + + if guard.sessions.len() >= self.cfg.max_sessions { + match self.cfg.on_full { + OnFull::Reject => return Err(SessionError::AtCapacity), + OnFull::EvictOldest => self.evict_oldest(&mut guard).await?, + } + } + + let (browser, ctx) = self.provider.acquire().await?; + let url = Url::parse("about:blank") + .map_err(|e| SessionError::Browser(AppError::ActionFailed(e.to_string())))?; + let tab = browser.new_target_in(&ctx, &url).await?; + let bus: Arc = Arc::new(BroadcastBus::default()); + browser.attach_events(&tab, bus.clone()).await?; + + let entry = Arc::new(SessionEntry { + id: AgentSessionId::next(), + ctx, + tab, + browser, + bus, + params, + running: AtomicBool::new(false), + last_active: Mutex::new(Instant::now()), + }); + guard.sessions.insert(entry.id.clone(), entry.clone()); + guard.by_owner.insert(owner, entry.id.clone()); + Ok(self.handle_for(entry)) + } + + async fn evict_oldest(&self, guard: &mut Inner) -> Result<(), SessionError> { + let victim = guard + .sessions + .values() + .filter(|e| !e.running.load(std::sync::atomic::Ordering::SeqCst)) + .min_by_key(|e| { + e.last_active + .lock() + .map(|g| *g) + .unwrap_or_else(|_| Instant::now()) + }) + .map(|e| (e.id.clone(), e.ctx.clone())); + let Some((id, ctx)) = victim else { + return Err(SessionError::AtCapacity); + }; + self.provider.release(&ctx).await?; + guard.sessions.remove(&id); + guard.by_owner.retain(|_, v| v != &id); + Ok(()) + } + + /// Look up a live session by id. + pub async fn get(&self, id: &AgentSessionId) -> Option { + let guard = self.inner.lock().await; + guard.sessions.get(id).cloned().map(|e| self.handle_for(e)) + } + + /// List ids of all live sessions. + pub async fn list(&self) -> Vec { + let guard = self.inner.lock().await; + guard.sessions.keys().cloned().collect() + } +} + +#[async_trait] +impl SessionRegistry for AsyncMutex> +where + Owner: Eq + Hash + Clone + Send + Sync + 'static, +{ + async fn forget(&self, id: &AgentSessionId) { + let mut guard = self.lock().await; + guard.sessions.remove(id); + guard.by_owner.retain(|_, v| v != id); + } +} diff --git a/ras-session/src/manager_reaper.rs b/ras-session/src/manager_reaper.rs new file mode 100644 index 0000000..f9d9d44 --- /dev/null +++ b/ras-session/src/manager_reaper.rs @@ -0,0 +1,74 @@ +//! Background reaper that releases idle, non-running sessions. + +use std::hash::Hash; +use std::sync::Arc; +use std::sync::atomic::Ordering::SeqCst; +use std::time::Duration; + +use ras_types::ContextId; +use tokio::sync::Mutex as AsyncMutex; +use tokio::time::interval; + +use crate::config::AgentSessionId; +use crate::manager::Inner; +use crate::provider::BrowserProvider; + +const MAX_TICK: Duration = Duration::from_secs(30); + +/// Launch the periodic reaper task. Ticks every `min(idle_timeout, 30s)`. +pub(crate) fn spawn_reaper( + inner: Arc>>, + provider: Arc, + idle_timeout: Duration, +) where + Owner: Eq + Hash + Clone + Send + Sync + 'static, +{ + let tick = idle_timeout.min(MAX_TICK).max(Duration::from_millis(10)); + tokio::spawn(async move { + let mut ticker = interval(tick); + loop { + ticker.tick().await; + reap_once(&inner, &provider, idle_timeout).await; + } + }); +} + +async fn reap_once( + inner: &Arc>>, + provider: &Arc, + idle_timeout: Duration, +) where + Owner: Eq + Hash + Clone + Send + Sync + 'static, +{ + let victims = collect_victims(inner, idle_timeout).await; + for (id, ctx) in victims { + let _ = provider.release(&ctx).await; + let mut guard = inner.lock().await; + guard.sessions.remove(&id); + guard.by_owner.retain(|_, v| v != &id); + } +} + +async fn collect_victims( + inner: &Arc>>, + idle_timeout: Duration, +) -> Vec<(AgentSessionId, ContextId)> +where + Owner: Eq + Hash + Clone + Send + Sync + 'static, +{ + let guard = inner.lock().await; + guard + .sessions + .values() + .filter(|e| !e.running.load(SeqCst)) + .filter(|e| { + let elapsed = e + .last_active + .lock() + .map(|g| g.elapsed()) + .unwrap_or_else(|_| Duration::ZERO); + elapsed > idle_timeout + }) + .map(|e| (e.id.clone(), e.ctx.clone())) + .collect() +} diff --git a/ras-session/src/provider.rs b/ras-session/src/provider.rs new file mode 100644 index 0000000..fd1a211 --- /dev/null +++ b/ras-session/src/provider.rs @@ -0,0 +1,45 @@ +//! Browser-provider abstraction for tenant session isolation. + +use std::sync::Arc; + +use async_trait::async_trait; +use ras_cdp::BrowserPort; +use ras_errors::AppError; +use ras_types::ContextId; + +/// Supplies an isolated browser context and the port to drive it. +/// +/// Swap implementations to choose the isolation strategy: +/// - `SharedBrowserProvider` — one Chromium process, one CDP context per tenant. +/// - Future: process-per-tenant via a pool of browser launchers. +#[async_trait] +pub trait BrowserProvider: Send + Sync { + /// Allocate a fresh browser context and return the port + context id. + async fn acquire(&self) -> Result<(Arc, ContextId), AppError>; + /// Release a context obtained via [`acquire`][Self::acquire]. + async fn release(&self, ctx: &ContextId) -> Result<(), AppError>; +} + +/// One shared browser; each session gets its own CDP BrowserContext. +pub struct SharedBrowserProvider { + browser: Arc, +} + +impl SharedBrowserProvider { + /// Wrap an existing browser adapter. + pub fn new(browser: Arc) -> Self { + Self { browser } + } +} + +#[async_trait] +impl BrowserProvider for SharedBrowserProvider { + async fn acquire(&self) -> Result<(Arc, ContextId), AppError> { + let ctx = self.browser.create_context().await?; + Ok((self.browser.clone(), ctx)) + } + + async fn release(&self, ctx: &ContextId) -> Result<(), AppError> { + self.browser.close_context(ctx).await + } +} diff --git a/ras-session/src/spawn_params.rs b/ras-session/src/spawn_params.rs new file mode 100644 index 0000000..dfc17cf --- /dev/null +++ b/ras-session/src/spawn_params.rs @@ -0,0 +1,19 @@ +//! Parameters carried into each spawned agent session. + +use std::sync::Arc; + +use ras_dom::DomExtractor; +use ras_llm::LlmClient; +use ras_tools::domain::registry::ActionRegistry; + +/// Per-session wiring used to build a `RunAgent` when a task runs. +pub struct SpawnParams { + /// LLM client driving the agent. + pub llm: Arc, + /// Action registry exposed to the agent. + pub registry: Arc, + /// Optional DOM extractor for richer page state. + pub dom_extractor: Option>, + /// Maximum agent steps per task. + pub max_steps: u32, +} diff --git a/ras-session/tests/manager_test.rs b/ras-session/tests/manager_test.rs new file mode 100644 index 0000000..a374288 --- /dev/null +++ b/ras-session/tests/manager_test.rs @@ -0,0 +1,179 @@ +//! Integration tests for `SessionManager` and `SessionHandle`. + +mod mock_browser; + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use async_trait::async_trait; +use mock_browser::MockBrowser; +use ras_errors::AppError; +use ras_llm::{ChatMessage, ChatResponse, InvokeOptions, LlmClient, ProviderName}; +use ras_session::provider::BrowserProvider; +use ras_session::{OnFull, SessionConfig, SessionManager, SpawnParams}; +use ras_tools::domain::registry::ActionRegistry; +use ras_types::ContextId; + +/// LLM stub; `ainvoke` is never called in these unit tests. +struct StubLlm; + +#[async_trait] +impl LlmClient for StubLlm { + fn provider(&self) -> ProviderName { + ProviderName("stub".into()) + } + + fn model(&self) -> &str { + "stub" + } + + async fn ainvoke( + &self, + _messages: Vec, + _options: InvokeOptions, + ) -> Result { + Err(AppError::ActionFailed("stub".into())) + } +} + +/// Provider backed by a single `MockBrowser`, recording releases and acquires. +struct MockProvider { + browser: Arc, + released: Arc>>, + acquires: Arc, + counter: AtomicU64, +} + +impl MockProvider { + fn new() -> Arc { + Arc::new(Self { + browser: Arc::new(MockBrowser::new("ctx")), + released: Arc::new(Mutex::new(Vec::new())), + acquires: Arc::new(AtomicU64::new(0)), + counter: AtomicU64::new(0), + }) + } +} + +#[async_trait] +impl BrowserProvider for MockProvider { + async fn acquire(&self) -> Result<(Arc, ContextId), AppError> { + self.acquires.fetch_add(1, Ordering::SeqCst); + let n = self.counter.fetch_add(1, Ordering::SeqCst); + Ok((self.browser.clone(), ContextId(format!("ctx-{n}").into()))) + } + + async fn release(&self, ctx: &ContextId) -> Result<(), AppError> { + self.released.lock().expect("lock").push(ctx.clone()); + Ok(()) + } +} + +fn params() -> SpawnParams { + SpawnParams { + llm: Arc::new(StubLlm), + registry: Arc::new(ActionRegistry::new()), + dom_extractor: None, + max_steps: 5, + } +} + +#[tokio::test] +async fn spawn_under_max_returns_ok_and_is_listed() { + let provider = MockProvider::new(); + let mgr: SessionManager = SessionManager::new(provider, SessionConfig::default()); + + let handle = mgr.spawn("owner-a".into(), params()).await.expect("spawn"); + let id = handle.id(); + + assert!(mgr.get(&id).await.is_some()); + assert_eq!(mgr.list().await.len(), 1); +} + +#[tokio::test] +async fn reject_policy_returns_at_capacity_on_second_owner() { + let provider = MockProvider::new(); + let cfg = SessionConfig { + max_sessions: 1, + on_full: OnFull::Reject, + allow_multi_per_owner: true, + ..SessionConfig::default() + }; + let mgr: SessionManager = SessionManager::new(provider, cfg); + + mgr.spawn("a".into(), params()).await.expect("first"); + let err = mgr.spawn("b".into(), params()).await; + + assert!(matches!(err, Err(ras_session::SessionError::AtCapacity))); +} + +#[tokio::test] +async fn evict_oldest_releases_and_removes_first() { + let provider = MockProvider::new(); + let released = provider.released.clone(); + let cfg = SessionConfig { + max_sessions: 1, + on_full: OnFull::EvictOldest, + allow_multi_per_owner: true, + ..SessionConfig::default() + }; + let mgr: SessionManager = SessionManager::new(provider, cfg); + + let first = mgr.spawn("a".into(), params()).await.expect("first"); + let first_id = first.id(); + mgr.spawn("b".into(), params()).await.expect("second"); + + assert!(mgr.get(&first_id).await.is_none()); + assert_eq!(released.lock().expect("lock").len(), 1); +} + +#[tokio::test] +async fn one_per_owner_reuses_session() { + let provider = MockProvider::new(); + let acquires = provider.acquires.clone(); + let cfg = SessionConfig { + allow_multi_per_owner: false, + ..SessionConfig::default() + }; + let mgr: SessionManager = SessionManager::new(provider, cfg); + + let a = mgr.spawn("owner".into(), params()).await.expect("a"); + let b = mgr.spawn("owner".into(), params()).await.expect("b"); + + assert_eq!(a.id(), b.id()); + assert_eq!(acquires.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn close_releases_ctx_and_forgets() { + let provider = MockProvider::new(); + let released = provider.released.clone(); + let mgr: SessionManager = SessionManager::new(provider, SessionConfig::default()); + + let handle = mgr.spawn("a".into(), params()).await.expect("spawn"); + let id = handle.id(); + handle.close().await.expect("close"); + + assert!(mgr.get(&id).await.is_none()); + assert_eq!(released.lock().expect("lock").len(), 1); +} + +#[tokio::test] +async fn idle_reaper_releases_idle_session() { + let provider = MockProvider::new(); + let released = provider.released.clone(); + let cfg = SessionConfig { + idle_timeout: Duration::from_millis(50), + ..SessionConfig::default() + }; + let mgr: SessionManager = SessionManager::new(provider, cfg); + + let handle = mgr.spawn("a".into(), params()).await.expect("spawn"); + let id = handle.id(); + + tokio::time::sleep(Duration::from_millis(300)).await; + + assert!(mgr.get(&id).await.is_none()); + assert_eq!(released.lock().expect("lock").len(), 1); +} diff --git a/ras-session/tests/mock_browser.rs b/ras-session/tests/mock_browser.rs new file mode 100644 index 0000000..b238b55 --- /dev/null +++ b/ras-session/tests/mock_browser.rs @@ -0,0 +1,139 @@ +//! Minimal `BrowserPort` stub for integration tests. + +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use ras_cdp::{BrowserPort, ScreenshotFormat, Viewport}; +use ras_errors::AppError; +use ras_events::EventBus; +use ras_types::{BackendNodeId, ContextId, TargetId}; +use url::Url; + +/// Records which CDP context operations were called. +#[derive(Debug, Default, Clone)] +pub struct CallLog(pub Arc>>); + +impl CallLog { + pub fn push(&self, s: impl Into) { + self.0.lock().expect("lock").push(s.into()); + } + + #[allow(dead_code)] + pub fn snapshot(&self) -> Vec { + self.0.lock().expect("lock").clone() + } +} + +/// Stub browser that captures `create_context` / `close_context` calls. +pub struct MockBrowser { + pub log: CallLog, + pub context_id: ContextId, +} + +impl MockBrowser { + pub fn new(ctx: &str) -> Self { + Self { + log: CallLog::default(), + context_id: ContextId(ctx.into()), + } + } +} + +#[async_trait] +impl BrowserPort for MockBrowser { + async fn cdp_url(&self) -> Result { + Url::parse("http://localhost:9222").map_err(|e| AppError::ActionFailed(e.to_string())) + } + + async fn list_targets(&self) -> Result, AppError> { + Ok(vec![]) + } + + async fn focused_target(&self) -> Result { + Ok(TargetId("stub-target".into())) + } + + async fn navigate(&self, _t: &TargetId, _url: &Url) -> Result<(), AppError> { + Ok(()) + } + + async fn evaluate(&self, _t: &TargetId, _expr: &str) -> Result { + Ok(serde_json::Value::Null) + } + + async fn click_at(&self, _t: &TargetId, _x: i32, _y: i32) -> Result<(), AppError> { + Ok(()) + } + + async fn click_node(&self, _t: &TargetId, _n: BackendNodeId) -> Result<(), AppError> { + Ok(()) + } + + async fn mouse_down(&self, _t: &TargetId, _x: i32, _y: i32) -> Result<(), AppError> { + Ok(()) + } + + async fn mouse_up(&self, _t: &TargetId, _x: i32, _y: i32) -> Result<(), AppError> { + Ok(()) + } + + async fn mouse_move( + &self, + _t: &TargetId, + _x: i32, + _y: i32, + _buttons: i64, + ) -> Result<(), AppError> { + Ok(()) + } + + async fn mouse_hold(&self, _t: &TargetId, _x: i32, _y: i32, _ms: u64) -> Result<(), AppError> { + Ok(()) + } + + async fn type_text(&self, _t: &TargetId, _text: &str) -> Result<(), AppError> { + Ok(()) + } + + async fn screenshot(&self, _t: &TargetId, _fmt: ScreenshotFormat) -> Result, AppError> { + Ok(vec![]) + } + + async fn set_viewport(&self, _t: &TargetId, _v: Viewport) -> Result<(), AppError> { + Ok(()) + } + + async fn block_urls(&self, _t: &TargetId, _patterns: Vec) -> Result<(), AppError> { + Ok(()) + } + + async fn clear_cookies(&self, _t: &TargetId, _origin: &str) -> Result<(), AppError> { + Ok(()) + } + + async fn close_target(&self, _t: &TargetId) -> Result<(), AppError> { + Ok(()) + } + + async fn create_target(&self, _url: &Url) -> Result { + Ok(TargetId("stub-target".into())) + } + + async fn new_target_in(&self, _ctx: &ContextId, _url: &Url) -> Result { + Ok(TargetId("tab".into())) + } + + async fn create_context(&self) -> Result { + self.log.push("create_context"); + Ok(self.context_id.clone()) + } + + async fn close_context(&self, ctx: &ContextId) -> Result<(), AppError> { + self.log.push(format!("close_context:{}", ctx.0)); + Ok(()) + } + + async fn attach_events(&self, _t: &TargetId, _bus: Arc) -> Result<(), AppError> { + Ok(()) + } +} diff --git a/ras-session/tests/provider_test.rs b/ras-session/tests/provider_test.rs new file mode 100644 index 0000000..7e869a9 --- /dev/null +++ b/ras-session/tests/provider_test.rs @@ -0,0 +1,45 @@ +//! Integration tests for `SharedBrowserProvider`. + +mod mock_browser; + +use std::sync::Arc; + +use mock_browser::MockBrowser; +use ras_session::provider::{BrowserProvider, SharedBrowserProvider}; + +#[tokio::test] +async fn acquire_returns_context_id_from_create_context() { + let mock = Arc::new(MockBrowser::new("ctx-1")); + let provider = SharedBrowserProvider::new(mock.clone()); + + let (_, ctx) = provider.acquire().await.expect("acquire"); + + assert_eq!(ctx.0.as_str(), "ctx-1"); + let calls = mock.log.snapshot(); + assert_eq!(calls, vec!["create_context"]); +} + +#[tokio::test] +async fn acquire_returns_same_browser_arc() { + let mock = Arc::new(MockBrowser::new("ctx-2")); + let provider = SharedBrowserProvider::new(mock.clone()); + + let (browser_arc, _) = provider.acquire().await.expect("acquire"); + + assert!(Arc::ptr_eq( + &(browser_arc as Arc), + &(mock.clone() as Arc), + )); +} + +#[tokio::test] +async fn release_calls_close_context_with_ctx() { + let mock = Arc::new(MockBrowser::new("ctx-3")); + let provider = SharedBrowserProvider::new(mock.clone()); + + let (_, ctx) = provider.acquire().await.expect("acquire"); + provider.release(&ctx).await.expect("release"); + + let calls = mock.log.snapshot(); + assert_eq!(calls, vec!["create_context", "close_context:ctx-3"]); +}