From 3553c83ffac8fd84d8fa544e1408dca355094a31 Mon Sep 17 00:00:00 2001 From: Chili Palmer Date: Mon, 17 Aug 2026 21:56:15 +0000 Subject: [PATCH] feat(grid-agent): supervise grid sessions (#121) --- config/grid-agent.example.json | 7 + crates/metacrate-grid-agent/Cargo.toml | 3 + crates/metacrate-grid-agent/README.md | 10 +- crates/metacrate-grid-agent/src/backend.rs | 247 +++- crates/metacrate-grid-agent/src/config.rs | 91 +- crates/metacrate-grid-agent/src/lib.rs | 13 +- crates/metacrate-grid-agent/src/main.rs | 106 +- crates/metacrate-grid-agent/src/session.rs | 1061 +++++++++++++++++ .../metacrate-grid-agent/src/session_tests.rs | 526 ++++++++ .../tests/dependency_policy.rs | 21 +- .../tests/session_supervisor.rs | 227 ++++ docs/grid-agent-architecture.md | 26 +- docs/grid-agent-session.md | 75 ++ 13 files changed, 2377 insertions(+), 36 deletions(-) create mode 100644 crates/metacrate-grid-agent/src/session.rs create mode 100644 crates/metacrate-grid-agent/src/session_tests.rs create mode 100644 crates/metacrate-grid-agent/tests/session_supervisor.rs create mode 100644 docs/grid-agent-session.md diff --git a/config/grid-agent.example.json b/config/grid-agent.example.json index 01f1a7b..3f113e4 100644 --- a/config/grid-agent.example.json +++ b/config/grid-agent.example.json @@ -25,5 +25,12 @@ "storage_path": "data/grid-agent", "behavior": { "heartbeat_seconds": 30 + }, + "reconnect": { + "initial_delay_milliseconds": 1000, + "maximum_delay_seconds": 60, + "stable_reset_seconds": 120, + "jitter_basis_points": 2000, + "offline_work_capacity": 128 } } diff --git a/crates/metacrate-grid-agent/Cargo.toml b/crates/metacrate-grid-agent/Cargo.toml index cf1f6bb..8e7f50b 100644 --- a/crates/metacrate-grid-agent/Cargo.toml +++ b/crates/metacrate-grid-agent/Cargo.toml @@ -18,6 +18,9 @@ sha2 = "0.11" tokio = { version = "1.53.1", features = ["macros", "rt", "sync", "time"] } url = "2.5.8" +[dev-dependencies] +tokio = { version = "1.53.1", features = ["test-util"] } + [target.'cfg(any(unix, windows))'.dependencies] tokio = { version = "1.53.1", features = ["io-util", "net", "rt-multi-thread", "signal"] } diff --git a/crates/metacrate-grid-agent/README.md b/crates/metacrate-grid-agent/README.md index b5554ba..ce5c616 100644 --- a/crates/metacrate-grid-agent/README.md +++ b/crates/metacrate-grid-agent/README.md @@ -6,9 +6,10 @@ OpenSim grid agent. It contains a reusable library and the offline: it publishes a deterministic ready event, accepts control commands, and shuts down both owned tasks without contacting a grid or LLM. The library also provides the bounded exact-endpoint LLM transport and tool loop for live -adapters. The `live-grid` feature exposes the side-effect-free owner for the existing -`libremetaverse::GridClient`; later live adapters must extend that manager graph -instead of adding a protocol client. +adapters. The `live-grid` feature exposes the owner for the existing +`libremetaverse::GridClient`. Its supervised session adapter reuses that +client's native `NetworkManager` for login, event-queue readiness, disconnect +notifications, and logout instead of adding a protocol client. The LLM connection identity has exactly two resolved fields: `llm.endpoint_url` and `llm.api_key`. The endpoint is used exactly as supplied; @@ -44,3 +45,6 @@ compatibility envelope, retry rules, and tool-loop safety contract. The central origin matrix, approval binding, budgets, prompt-data boundary, and opaque backend authorization are specified in [`../../docs/grid-agent-policy.md`](../../docs/grid-agent-policy.md). +The reconnect state machine, generation fencing, offline-work contract, and +shutdown deadline are specified in +[`../../docs/grid-agent-session.md`](../../docs/grid-agent-session.md). diff --git a/crates/metacrate-grid-agent/src/backend.rs b/crates/metacrate-grid-agent/src/backend.rs index 7f04f66..3802585 100644 --- a/crates/metacrate-grid-agent/src/backend.rs +++ b/crates/metacrate-grid-agent/src/backend.rs @@ -7,6 +7,10 @@ use std::error::Error; use std::fmt; use std::future::Future; use std::pin::Pin; +#[cfg(feature = "live-grid")] +use std::sync::Arc; +#[cfg(feature = "live-grid")] +use std::sync::atomic::{AtomicBool, Ordering}; use tokio::sync::mpsc; pub type BackendFuture<'a, T> = Pin + Send + 'a>>; @@ -82,11 +86,45 @@ impl OfflineGridBackend { /// Live-feature composition owner that guarantees production adapters reuse /// the existing `libremetaverse` manager/client graph. #[cfg(feature = "live-grid")] -#[derive(Debug)] pub struct LibremetaverseClientOwner { client: libremetaverse::GridClient, } +#[cfg(feature = "live-grid")] +impl fmt::Debug for LibremetaverseClientOwner { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LibremetaverseClientOwner") + .field("client", &"[NATIVE CLIENT REDACTED]") + .finish() + } +} + +/// Feature-gated session adapter that reuses the existing native +/// `NetworkManager` login, event, disconnect, and logout lifecycle. +#[cfg(feature = "live-grid")] +#[derive(Clone)] +pub struct LibremetaverseSessionBackend { + network: libremetaverse::NetworkManager, + login_url: String, + first_name: String, + last_name: String, + password: crate::config::SecretString, +} + +#[cfg(feature = "live-grid")] +impl fmt::Debug for LibremetaverseSessionBackend { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LibremetaverseSessionBackend") + .field("transport_connected", &self.network.native_connected()) + .field("login_url", &"[REDACTED ENDPOINT]") + .field("avatar", &"[REDACTED IDENTITY]") + .field("password", &self.password) + .finish_non_exhaustive() + } +} + #[cfg(feature = "live-grid")] impl LibremetaverseClientOwner { /// Builds the shared client composition root without starting login or I/O. @@ -107,6 +145,213 @@ impl LibremetaverseClientOwner { pub const fn client(&self) -> &libremetaverse::GridClient { &self.client } + + /// Creates the supervised live-session adapter without logging in. + pub fn session_backend( + &self, + connection: crate::config::GridConnection, + ) -> Result { + let avatar = connection.avatar_name.trim(); + let (first_name, last_name) = avatar + .split_once(char::is_whitespace) + .map_or((avatar, "Resident"), |(first, last)| (first, last.trim())); + if first_name.is_empty() || last_name.is_empty() { + return Err(BackendError::Configuration { + component: "grid avatar identity", + }); + } + Ok(LibremetaverseSessionBackend { + network: self.client.network(), + login_url: connection.login_url.expose_url().to_owned(), + first_name: first_name.to_owned(), + last_name: last_name.to_owned(), + password: connection.password, + }) + } +} + +#[cfg(feature = "live-grid")] +struct LibremetaverseSession { + generation: u64, + network: libremetaverse::NetworkManager, + signals: mpsc::Receiver, + subscriptions: Vec, +} + +#[cfg(feature = "live-grid")] +impl crate::session::GridSession for LibremetaverseSession { + fn generation(&self) -> u64 { + self.generation + } + + fn next_signal( + &mut self, + cancellation: CancellationToken, + ) -> crate::session::SessionFuture<'_, crate::session::SessionSignal> { + Box::pin(async move { + tokio::select! { + signal = self.signals.recv() => signal.unwrap_or_else(|| { + crate::session::SessionSignal::Disconnected( + crate::session::SessionFailure::new( + crate::session::SessionFailureKind::TransientTransport, + ), + ) + }), + () = cancellation.cancelled() => crate::session::SessionSignal::Disconnected( + crate::session::SessionFailure::new( + crate::session::SessionFailureKind::TransientTransport, + ), + ), + } + }) + } + + fn logout( + mut self: Box, + cancellation: CancellationToken, + ) -> crate::session::SessionFuture<'static, Result<(), crate::session::SessionFailure>> { + Box::pin(async move { + self.subscriptions.clear(); + self.network + .native_logout_async(Some(cancellation)) + .await + .map_err(|_| { + crate::session::SessionFailure::new( + crate::session::SessionFailureKind::TransientTransport, + ) + }) + }) + } +} + +#[cfg(feature = "live-grid")] +impl crate::session::GridSessionBackend for LibremetaverseSessionBackend { + fn login( + &self, + generation: u64, + cancellation: CancellationToken, + ) -> crate::session::SessionFuture< + '_, + Result, crate::session::SessionFailure>, + > { + Box::pin(async move { + let mut params = self + .network + .native_default_login_params( + self.first_name.clone(), + self.last_name.clone(), + self.password.expose_secret().to_owned(), + "MetaCrate".to_owned(), + env!("CARGO_PKG_VERSION").to_owned(), + ) + .map_err(|_| { + crate::session::SessionFailure::new( + crate::session::SessionFailureKind::InvalidConfiguration, + ) + })?; + params.uri.clone_from(&self.login_url); + params.start = "last".to_owned(); + let logged_in = self + .network + .native_login(params, Some(cancellation)) + .await + .map_err(|_| { + crate::session::SessionFailure::new( + crate::session::SessionFailureKind::TransientTransport, + ) + })?; + if !logged_in { + return Err(classify_native_login_failure( + &self.network.native_login_error_key(), + )); + } + + // Native login has already installed the current simulator and + // parsed inventory state. The existing GridClient managers retain + // world/inventory and movement-worker ownership; this generation + // adds only one disconnect and one readiness subscription. + let (sender, signals) = mpsc::channel(8); + let disconnected_sender = sender.clone(); + let disconnected = self + .network + .native_subscribe_disconnected(Arc::new(move |event| { + let kind = match event.reason() { + libremetaverse::NetworkManagerDisconnectType::NetworkTimeout => { + crate::session::SessionFailureKind::TransientTransport + } + libremetaverse::NetworkManagerDisconnectType::ServerInitiated => { + crate::session::SessionFailureKind::Kicked + } + libremetaverse::NetworkManagerDisconnectType::SimShutdown => { + crate::session::SessionFailureKind::Maintenance + } + libremetaverse::NetworkManagerDisconnectType::ClientInitiated => { + crate::session::SessionFailureKind::TransientTransport + } + }; + let _ = + disconnected_sender.try_send(crate::session::SessionSignal::Disconnected( + crate::session::SessionFailure::new(kind), + )); + })); + let ready_sender = sender.clone(); + let ready_once = Arc::new(AtomicBool::new(false)); + let callback_ready = Arc::clone(&ready_once); + let ready = self + .network + .native_subscribe_event_queue_running(Arc::new(move |_| { + if !callback_ready.swap(true, Ordering::AcqRel) { + let _ = ready_sender.try_send(crate::session::SessionSignal::Ready); + } + })); + if !self.network.native_connected() { + let _ = sender.try_send(crate::session::SessionSignal::Disconnected( + crate::session::SessionFailure::new( + crate::session::SessionFailureKind::TransientTransport, + ), + )); + } else if self + .network + .native_current_sim() + .and_then(|simulator| simulator.native_is_event_queue_running(None).ok()) + .unwrap_or(false) + { + if !ready_once.swap(true, Ordering::AcqRel) { + let _ = sender.try_send(crate::session::SessionSignal::Ready); + } + } + let session: Box = Box::new(LibremetaverseSession { + generation, + network: self.network.clone(), + signals, + subscriptions: vec![disconnected, ready], + }); + Ok(session) + }) + } + + fn flush_audit( + &self, + _cancellation: CancellationToken, + ) -> crate::session::SessionFuture<'_, Result<(), crate::session::SessionFailure>> { + Box::pin(async { Ok(()) }) + } +} + +#[cfg(feature = "live-grid")] +fn classify_native_login_failure(error_key: &str) -> crate::session::SessionFailure { + let normalized = error_key.trim().to_ascii_lowercase(); + let kind = if matches!( + normalized.as_str(), + "key" | "password" | "credential" | "account" | "username" | "user" + ) { + crate::session::SessionFailureKind::InvalidCredentials + } else if normalized == "canceled" { + crate::session::SessionFailureKind::TransientTransport + } else { + crate::session::SessionFailureKind::ServerFailure + }; + crate::session::SessionFailure::new(kind) } impl GridBackend for OfflineGridBackend { diff --git a/crates/metacrate-grid-agent/src/config.rs b/crates/metacrate-grid-agent/src/config.rs index 751b1b7..7c560fa 100644 --- a/crates/metacrate-grid-agent/src/config.rs +++ b/crates/metacrate-grid-agent/src/config.rs @@ -213,6 +213,7 @@ pub struct AgentConfig { pub limits: Limits, pub storage_path: PathBuf, pub behavior: BehaviorSettings, + pub reconnect: crate::session::ReconnectPolicy, } impl AgentConfig { @@ -239,6 +240,9 @@ impl AgentConfig { /// Returns the first invalid field without performing I/O. pub fn validate(&self) -> Result<(), ConfigError> { validate_limits(&self.limits)?; + self.reconnect + .validate() + .map_err(|_| ConfigError::InvalidReconnect)?; if self.mode != OperatingMode::OfflineFake && self.grid.is_none() { return Err(ConfigError::Missing { field: "grid", @@ -411,6 +415,7 @@ pub enum ConfigError { minimum: usize, maximum: usize, }, + InvalidReconnect, } impl fmt::Display for ConfigError { @@ -460,6 +465,7 @@ impl fmt::Display for ConfigError { formatter, "unsafe {field}={value}; expected {minimum}..={maximum}" ), + Self::InvalidReconnect => formatter.write_str("invalid reconnect policy bounds"), } } } @@ -479,6 +485,7 @@ struct FileConfig { limits: RawLimits, storage_path: Option, behavior: RawBehavior, + reconnect: RawReconnect, } #[derive(Clone, Default, Deserialize)] @@ -534,6 +541,16 @@ struct RawBehavior { heartbeat_seconds: Option, } +#[derive(Clone, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +struct RawReconnect { + initial_delay_milliseconds: Option, + maximum_delay_seconds: Option, + stable_reset_seconds: Option, + jitter_basis_points: Option, + offline_work_capacity: Option, +} + fn read_config(path: &Path) -> Result { let bytes = read_bounded_regular_file("configuration file", path, MAX_CONFIG_BYTES)?; serde_json::from_slice(&bytes).map_err(|error| ConfigError::InvalidSchema { @@ -666,31 +683,51 @@ fn resolve( }; validate_limits(&limits)?; + let timeouts = Timeouts { + startup: checked_duration( + "timeouts.startup_seconds", + raw.timeouts.startup.unwrap_or(30), + 1, + 300, + )?, + shutdown: checked_duration( + "timeouts.shutdown_seconds", + raw.timeouts.shutdown.unwrap_or(10), + 1, + 60, + )?, + request: checked_duration( + "timeouts.request_seconds", + raw.timeouts.request.unwrap_or(60), + 1, + 300, + )?, + }; + let reconnect_defaults = crate::session::ReconnectPolicy::default(); + let reconnect = crate::session::ReconnectPolicy { + initial_delay: Duration::from_millis( + raw.reconnect.initial_delay_milliseconds.unwrap_or(1_000), + ), + maximum_delay: Duration::from_secs(raw.reconnect.maximum_delay_seconds.unwrap_or(60)), + stable_reset_after: Duration::from_secs(raw.reconnect.stable_reset_seconds.unwrap_or(120)), + shutdown_deadline: timeouts.shutdown, + jitter_basis_points: raw + .reconnect + .jitter_basis_points + .unwrap_or(reconnect_defaults.jitter_basis_points), + instance_seed: reconnect_defaults.instance_seed, + offline_work_capacity: raw + .reconnect + .offline_work_capacity + .unwrap_or(reconnect_defaults.offline_work_capacity), + }; + let config = AgentConfig { mode, llm, grid, authorized_avatar_uuids, - timeouts: Timeouts { - startup: checked_duration( - "timeouts.startup_seconds", - raw.timeouts.startup.unwrap_or(30), - 1, - 300, - )?, - shutdown: checked_duration( - "timeouts.shutdown_seconds", - raw.timeouts.shutdown.unwrap_or(10), - 1, - 60, - )?, - request: checked_duration( - "timeouts.request_seconds", - raw.timeouts.request.unwrap_or(60), - 1, - 300, - )?, - }, + timeouts, limits, storage_path: raw .storage_path @@ -703,6 +740,7 @@ fn resolve( 300, )?, }, + reconnect, }; config.validate()?; Ok(config) @@ -1076,6 +1114,19 @@ mod tests { Err(ConfigError::UnsafeLimit { .. }) )); let _ = fs::remove_file(path); + + let reconnect = temporary_file( + "unsafe-reconnect.json", + r#"{"reconnect":{"initial_delay_milliseconds":0}}"#, + ); + assert!(matches!( + ConfigLoader::new() + .with_file(&reconnect) + .with_environment(offline_environment()) + .load(), + Err(ConfigError::InvalidReconnect) + )); + let _ = fs::remove_file(reconnect); } #[test] diff --git a/crates/metacrate-grid-agent/src/lib.rs b/crates/metacrate-grid-agent/src/lib.rs index 6d3a436..2fbc68d 100644 --- a/crates/metacrate-grid-agent/src/lib.rs +++ b/crates/metacrate-grid-agent/src/lib.rs @@ -9,18 +9,21 @@ pub mod config; pub mod llm; pub mod policy; pub mod service; +pub mod session; pub mod tool_loop; pub mod types; #[cfg(test)] mod policy_tests; +#[cfg(test)] +mod session_tests; -#[cfg(feature = "live-grid")] -pub use backend::LibremetaverseClientOwner; pub use backend::{ AuthorizedToolBackend, BackendError, BackendFuture, GridBackend, OfflineGridBackend, WorldMutator, }; +#[cfg(feature = "live-grid")] +pub use backend::{LibremetaverseClientOwner, LibremetaverseSessionBackend}; pub use config::{ AgentConfig, BehaviorSettings, ConfigError, ConfigLoader, EndpointUrl, Environment, GridConnection, Limits, LlmConnection, MapEnvironment, OperatingMode, SecretString, @@ -39,6 +42,12 @@ pub use policy::{ ResourceCost, ResourceEstimator, Risk, SchedulerGrantId, UntrustedData, UntrustedSource, }; pub use service::{AgentService, ServiceError, ServiceHandle, ServiceState}; +pub use session::{ + GridSession, GridSessionBackend, ReconnectPolicy, SessionControl, SessionFailure, + SessionFailureKind, SessionFuture, SessionObservation, SessionSignal, SessionState, + SessionStatus, SessionSupervisor, SessionSupervisorError, SessionSupervisorHandle, SessionWork, + WorkDisposition, WorkKind, +}; pub use tool_loop::{ HistorySummarizer, SessionGeneration, ToolExecution, ToolExecutor, ToolFuture, ToolLoop, ToolLoopError, ToolLoopLimits, ToolLoopOutcome, diff --git a/crates/metacrate-grid-agent/src/main.rs b/crates/metacrate-grid-agent/src/main.rs index b02251a..18731b1 100644 --- a/crates/metacrate-grid-agent/src/main.rs +++ b/crates/metacrate-grid-agent/src/main.rs @@ -4,6 +4,8 @@ use metacrate_grid_agent::{ use std::error::Error; use std::fmt; use std::path::PathBuf; +#[cfg(feature = "live-grid")] +use std::sync::Arc; const MAX_ARGUMENTS: usize = 8; @@ -84,9 +86,11 @@ async fn main() -> Result<(), Box> { return Ok(()); } if config.mode != OperatingMode::OfflineFake { + #[cfg(feature = "live-grid")] + return run_live(config, options.run_once).await; + #[cfg(not(feature = "live-grid"))] return Err(CliError( - "this architecture issue starts only the offline backend; live login is owned by a later milestone issue" - .into(), + "live grid mode requires rebuilding with --features live-grid".into(), ) .into()); } @@ -133,3 +137,101 @@ async fn main() -> Result<(), Box> { println!("grid agent stopped cleanly"); Ok(()) } + +#[cfg(feature = "live-grid")] +async fn run_live( + config: metacrate_grid_agent::AgentConfig, + run_once: bool, +) -> Result<(), Box> { + use metacrate_grid_agent::{ + GridSessionBackend, LibremetaverseClientOwner, SessionObservation, SessionState, + SessionSupervisor, + }; + + let connection = config.grid.clone().ok_or_else(|| { + CliError("validated live configuration did not contain a grid connection".into()) + })?; + let owner = LibremetaverseClientOwner::new()?; + let backend = owner.session_backend(connection)?; + let erased: Arc = Arc::new(backend); + let mut handle = SessionSupervisor::new( + erased, + config.reconnect, + config.limits.control_queue, + config.limits.observable_queue, + )? + .start(); + + if run_once { + let readiness = tokio::time::timeout(config.timeouts.startup, async { + loop { + match handle.next_observation().await { + Some(SessionObservation::Transition { status, .. }) if status.agent_ready => { + return Ok::<(), CliError>(()); + } + Some(SessionObservation::Transition { + status: + metacrate_grid_agent::SessionStatus { + state: SessionState::AuthenticationBlocked, + .. + }, + .. + }) => { + return Err(CliError( + "grid authentication/configuration requires operator action".into(), + )); + } + Some(_) => {} + None => { + return Err(CliError( + "session supervisor stopped before readiness".into(), + )); + } + } + } + }) + .await; + let readiness = match readiness { + Ok(result) => result, + Err(_) => Err(CliError("timed out waiting for full grid readiness".into())), + }; + if let Err(error) = readiness { + handle.shutdown().await?; + return Err(error.into()); + } + handle.shutdown().await?; + println!("grid agent completed one supervised login/logout cycle"); + return Ok(()); + } + + println!("grid agent session supervisor started; press Ctrl-C to stop"); + let mut signal_error = None; + loop { + tokio::select! { + signal = tokio::signal::ctrl_c() => { + if let Err(error) = signal { + signal_error = Some(error); + } + break; + } + event = handle.next_observation() => { + let Some(event) = event else { break; }; + if let SessionObservation::Transition { status, reason, retry_in } = event { + println!( + "grid session state={} generation={} transport_connected={} agent_ready={} reason={reason:?} retry_in={retry_in:?}", + status.state.as_str(), + status.generation, + status.transport_connected, + status.agent_ready, + ); + } + } + } + } + handle.shutdown().await?; + if let Some(error) = signal_error { + return Err(error.into()); + } + println!("grid agent stopped cleanly"); + Ok(()) +} diff --git a/crates/metacrate-grid-agent/src/session.rs b/crates/metacrate-grid-agent/src/session.rs new file mode 100644 index 0000000..705ab7d --- /dev/null +++ b/crates/metacrate-grid-agent/src/session.rs @@ -0,0 +1,1061 @@ +//! Supervised, generation-fenced grid-session lifecycle. + +#![allow(clippy::missing_errors_doc)] + +use crate::types::{BoundedText, MAX_IDENTIFIER_BYTES}; +use libremetaverse_types::compat::{CancellationToken, CancellationTokenSource}; +use std::collections::{BTreeSet, VecDeque}; +use std::error::Error; +use std::fmt; +use std::future::Future; +use std::pin::Pin; +use std::sync::atomic::{AtomicU8, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use tokio::sync::{mpsc, oneshot}; +use tokio::task::JoinHandle; +use tokio::time::Instant; + +const MAX_OFFLINE_WORK: usize = 1_024; +const MAX_SEEN_WORK: usize = 4_096; +const MAX_BACKOFF: Duration = Duration::from_hours(1); +const MAX_STABLE_RESET: Duration = Duration::from_hours(24); +const MAX_SHUTDOWN: Duration = Duration::from_mins(1); + +pub type SessionFuture<'a, T> = Pin + Send + 'a>>; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[repr(u8)] +pub enum SessionState { + Stopped = 0, + Connecting = 1, + Online = 2, + Degraded = 3, + Backoff = 4, + AuthenticationBlocked = 5, + Paused = 6, + ShuttingDown = 7, +} + +impl SessionState { + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Stopped => "stopped", + Self::Connecting => "connecting", + Self::Online => "online", + Self::Degraded => "degraded", + Self::Backoff => "backoff", + Self::AuthenticationBlocked => "authentication-blocked", + Self::Paused => "paused", + Self::ShuttingDown => "shutting-down", + } + } + + const fn from_u8(value: u8) -> Self { + match value { + 1 => Self::Connecting, + 2 => Self::Online, + 3 => Self::Degraded, + 4 => Self::Backoff, + 5 => Self::AuthenticationBlocked, + 6 => Self::Paused, + 7 => Self::ShuttingDown, + _ => Self::Stopped, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SessionReason { + Startup, + LoginSucceeded, + AgentReady, + ReadinessLost, + TransientTransport, + Maintenance, + Kicked, + SimulatorDisconnected, + ServerFailure, + InvalidCredentials, + InvalidConfiguration, + StableSessionReset, + OperatorPause, + OperatorResume, + OperatorForceReconnect, + OperatorLogout, + ShutdownRequested, + ShutdownComplete, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SessionFailureKind { + InvalidCredentials, + InvalidConfiguration, + TransientTransport, + Maintenance, + Kicked, + SimulatorDisconnected, + ServerFailure, +} + +/// Secret-free failure classification. Raw login/server text is deliberately +/// not representable at this boundary. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct SessionFailure { + kind: SessionFailureKind, + retry_after: Option, +} + +impl SessionFailure { + #[must_use] + pub const fn new(kind: SessionFailureKind) -> Self { + Self { + kind, + retry_after: None, + } + } + + #[must_use] + pub const fn with_retry_after(kind: SessionFailureKind, retry_after: Duration) -> Self { + Self { + kind, + retry_after: Some(retry_after), + } + } + + #[must_use] + pub const fn kind(self) -> SessionFailureKind { + self.kind + } + + #[must_use] + pub const fn retry_after(self) -> Option { + self.retry_after + } + + const fn retryable(self) -> bool { + !matches!( + self.kind, + SessionFailureKind::InvalidCredentials | SessionFailureKind::InvalidConfiguration + ) + } + + const fn reason(self) -> SessionReason { + match self.kind { + SessionFailureKind::InvalidCredentials => SessionReason::InvalidCredentials, + SessionFailureKind::InvalidConfiguration => SessionReason::InvalidConfiguration, + SessionFailureKind::TransientTransport => SessionReason::TransientTransport, + SessionFailureKind::Maintenance => SessionReason::Maintenance, + SessionFailureKind::Kicked => SessionReason::Kicked, + SessionFailureKind::SimulatorDisconnected => SessionReason::SimulatorDisconnected, + SessionFailureKind::ServerFailure => SessionReason::ServerFailure, + } + } +} + +impl fmt::Display for SessionFailure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(formatter, "grid session failure: {:?}", self.kind) + } +} + +impl Error for SessionFailure {} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SessionSignal { + Ready, + Degraded, + Disconnected(SessionFailure), +} + +/// One connected generation. Dropping its subscriptions and joining its +/// workers is the responsibility of `logout`, which the supervisor calls once. +pub trait GridSession: Send { + fn generation(&self) -> u64; + + fn next_signal(&mut self, cancellation: CancellationToken) -> SessionFuture<'_, SessionSignal>; + + fn logout( + self: Box, + cancellation: CancellationToken, + ) -> SessionFuture<'static, Result<(), SessionFailure>>; +} + +/// Injectable login owner. A production implementation must reuse the native +/// `NetworkManager`; deterministic tests supply a fake implementation. +pub trait GridSessionBackend: Send + Sync + 'static { + fn login( + &self, + generation: u64, + cancellation: CancellationToken, + ) -> SessionFuture<'_, Result, SessionFailure>>; + + fn flush_audit( + &self, + cancellation: CancellationToken, + ) -> SessionFuture<'_, Result<(), SessionFailure>>; +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ReconnectPolicy { + pub initial_delay: Duration, + pub maximum_delay: Duration, + pub stable_reset_after: Duration, + pub shutdown_deadline: Duration, + pub jitter_basis_points: u16, + pub instance_seed: u64, + pub offline_work_capacity: usize, +} + +impl Default for ReconnectPolicy { + fn default() -> Self { + Self { + initial_delay: Duration::from_secs(1), + maximum_delay: Duration::from_mins(1), + stable_reset_after: Duration::from_mins(2), + shutdown_deadline: Duration::from_secs(10), + jitter_basis_points: 2_000, + instance_seed: runtime_jitter_seed(), + offline_work_capacity: 128, + } + } +} + +fn runtime_jitter_seed() -> u64 { + static NEXT_SEED: AtomicU64 = AtomicU64::new(1); + let time = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |duration| { + u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX) + }); + mix64(time ^ u64::from(std::process::id()) ^ NEXT_SEED.fetch_add(1, Ordering::Relaxed)) +} + +impl ReconnectPolicy { + pub fn validate(self) -> Result { + if self.initial_delay.is_zero() + || self.initial_delay > self.maximum_delay + || self.maximum_delay > MAX_BACKOFF + || self.stable_reset_after.is_zero() + || self.stable_reset_after > MAX_STABLE_RESET + || self.shutdown_deadline.is_zero() + || self.shutdown_deadline > MAX_SHUTDOWN + || self.jitter_basis_points > 5_000 + || !(1..=MAX_OFFLINE_WORK).contains(&self.offline_work_capacity) + { + return Err(SessionSupervisorError::UnsafePolicy); + } + Ok(self) + } + + #[must_use] + pub fn retry_delay(self, failure_count: u32, hint: Option) -> Duration { + let shift = failure_count.saturating_sub(1).min(31); + let factor = 1_u128 << shift; + let base_ms = self.initial_delay.as_millis().saturating_mul(factor); + let cap_ms = self.maximum_delay.as_millis(); + let hinted_ms = hint.map_or(0, |value| value.as_millis().min(cap_ms)); + let unclamped = base_ms.max(hinted_ms).min(cap_ms); + let spread = unclamped.saturating_mul(u128::from(self.jitter_basis_points)) / 10_000; + let random = mix64(self.instance_seed ^ u64::from(failure_count)); + let width = spread.saturating_mul(2).saturating_add(1); + let offset = if width == 0 { + 0 + } else { + u128::from(random) % width + }; + let jittered = unclamped + .saturating_sub(spread) + .saturating_add(offset) + .max(hinted_ms); + Duration::from_millis(u64::try_from(jittered.min(cap_ms)).unwrap_or(u64::MAX)) + } +} + +fn mix64(mut value: u64) -> u64 { + value ^= value >> 30; + value = value.wrapping_mul(0xbf58_476d_1ce4_e5b9); + value ^= value >> 27; + value = value.wrapping_mul(0x94d0_49bb_1331_11eb); + value ^ (value >> 31) +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct SessionStatus { + pub state: SessionState, + pub generation: u64, + pub transport_connected: bool, + pub agent_ready: bool, + pub consecutive_failures: u32, +} + +impl Default for SessionStatus { + fn default() -> Self { + Self { + state: SessionState::Stopped, + generation: 0, + transport_connected: false, + agent_ready: false, + consecutive_failures: 0, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum WorkKind { + ReadOnly, + IdempotentMutation, + NonIdempotentMutation, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SessionWork { + pub id: BoundedText, + pub kind: WorkKind, +} + +impl SessionWork { + pub fn new(id: impl Into, kind: WorkKind) -> Result { + Ok(Self { + id: BoundedText::new("session.work_id", id) + .map_err(|_| SessionSupervisorError::InvalidWork)?, + kind, + }) + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum WorkDisposition { + Accepted { generation: u64 }, + Queued, + RejectedOfflineMutation, + RejectedQueueFull, + RejectedDuplicate, + RejectedShuttingDown, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum SessionObservation { + Transition { + status: SessionStatus, + reason: SessionReason, + retry_in: Option, + }, + Work { + id: BoundedText, + disposition: WorkDisposition, + }, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SessionControl { + Pause, + Resume, + ForceReconnect, + Logout, + Shutdown, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum SessionSupervisorError { + UnsafePolicy, + InvalidWork, + ControlClosed, + ObservationClosed, + TaskPanicked, + ShutdownTimedOut, +} + +impl fmt::Display for SessionSupervisorError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::UnsafePolicy => formatter.write_str("unsafe reconnect policy"), + Self::InvalidWork => formatter.write_str("invalid session work envelope"), + Self::ControlClosed => formatter.write_str("session control queue closed"), + Self::ObservationClosed => formatter.write_str("session observation queue closed"), + Self::TaskPanicked => formatter.write_str("session supervisor task panicked"), + Self::ShutdownTimedOut => formatter.write_str("session shutdown deadline exceeded"), + } + } +} + +impl Error for SessionSupervisorError {} + +enum Command { + Control(SessionControl), + Submit(SessionWork, oneshot::Sender), +} + +#[derive(Debug)] +struct GenerationFence { + generation: AtomicU64, + source: Mutex>, +} + +impl GenerationFence { + fn rotate(&self, generation: u64) -> CancellationToken { + let mut source = lock(&self.source); + if let Some(previous) = source.take() { + previous.cancel(); + } + let current = CancellationTokenSource::new(); + let token = current.token(); + *source = Some(current); + self.generation.store(generation, Ordering::Release); + token + } + + fn invalidate(&self) { + if let Some(source) = lock(&self.source).take() { + source.cancel(); + } + self.generation.store(0, Ordering::Release); + } + + fn token(&self, generation: u64) -> Option { + if self.generation.load(Ordering::Acquire) != generation { + return None; + } + lock(&self.source) + .as_ref() + .map(CancellationTokenSource::token) + } +} + +pub struct SessionSupervisor { + backend: Arc, + policy: ReconnectPolicy, + control_capacity: usize, + observation_capacity: usize, +} + +impl SessionSupervisor { + pub fn new( + backend: Arc, + policy: ReconnectPolicy, + control_capacity: usize, + observation_capacity: usize, + ) -> Result { + let policy = policy.validate()?; + if control_capacity == 0 + || control_capacity > 256 + || observation_capacity == 0 + || observation_capacity > 8_192 + { + return Err(SessionSupervisorError::UnsafePolicy); + } + Ok(Self { + backend, + policy, + control_capacity, + observation_capacity, + }) + } + + #[must_use] + pub fn start(self) -> SessionSupervisorHandle { + let (commands, receiver) = mpsc::channel(self.control_capacity); + let (observations, observation_receiver) = mpsc::channel(self.observation_capacity); + let cancellation = CancellationTokenSource::new(); + let state = Arc::new(AtomicU8::new(SessionState::Stopped as u8)); + let status = Arc::new(Mutex::new(SessionStatus::default())); + let fence = Arc::new(GenerationFence { + generation: AtomicU64::new(0), + source: Mutex::new(None), + }); + let runtime = Runtime { + backend: self.backend, + policy: self.policy, + commands: receiver, + observations, + cancellation: cancellation.clone(), + state: Arc::clone(&state), + status: Arc::clone(&status), + fence: Arc::clone(&fence), + queued: VecDeque::new(), + queued_ids: BTreeSet::new(), + seen: VecDeque::new(), + seen_ids: BTreeSet::new(), + next_generation: 1, + failures: 0, + }; + let task = tokio::spawn(runtime.run()); + SessionSupervisorHandle { + commands, + observations: observation_receiver, + cancellation, + state, + status, + fence, + task: Some(task), + shutdown_deadline: self.policy.shutdown_deadline, + } + } +} + +pub struct SessionSupervisorHandle { + commands: mpsc::Sender, + observations: mpsc::Receiver, + cancellation: CancellationTokenSource, + state: Arc, + status: Arc>, + fence: Arc, + task: Option>, + shutdown_deadline: Duration, +} + +impl SessionSupervisorHandle { + #[must_use] + pub fn state(&self) -> SessionState { + SessionState::from_u8(self.state.load(Ordering::Acquire)) + } + + #[must_use] + pub fn status(&self) -> SessionStatus { + *lock(&self.status) + } + + pub async fn control(&self, control: SessionControl) -> Result<(), SessionSupervisorError> { + self.commands + .send(Command::Control(control)) + .await + .map_err(|_| SessionSupervisorError::ControlClosed) + } + + pub async fn submit( + &self, + work: SessionWork, + ) -> Result { + let (sender, receiver) = oneshot::channel(); + self.commands + .send(Command::Submit(work, sender)) + .await + .map_err(|_| SessionSupervisorError::ControlClosed)?; + receiver + .await + .map_err(|_| SessionSupervisorError::ControlClosed) + } + + pub async fn next_observation(&mut self) -> Option { + self.observations.recv().await + } + + /// Returns the cancellation token shared by inference, scheduled work, and + /// tools for exactly one live generation. + #[must_use] + pub fn generation_token(&self, generation: u64) -> Option { + self.fence.token(generation) + } + + /// Late LLM/tool results are accepted only while their generation is fully ready. + #[must_use] + pub fn accepts_result(&self, generation: u64) -> bool { + let status = self.status(); + status.agent_ready && status.generation == generation + } + + pub async fn shutdown(&mut self) -> Result<(), SessionSupervisorError> { + self.state + .store(SessionState::ShuttingDown as u8, Ordering::Release); + self.cancellation.cancel(); + let Some(task) = self.task.as_mut() else { + return Ok(()); + }; + match tokio::time::timeout(self.shutdown_deadline, &mut *task).await { + Ok(Ok(())) => { + self.task = None; + Ok(()) + } + Ok(Err(_)) => { + self.task = None; + Err(SessionSupervisorError::TaskPanicked) + } + Err(_) => { + task.abort(); + let _ = (&mut *task).await; + self.task = None; + Err(SessionSupervisorError::ShutdownTimedOut) + } + } + } +} + +impl Drop for SessionSupervisorHandle { + fn drop(&mut self) { + self.cancellation.cancel(); + if let Some(task) = &self.task { + task.abort(); + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum DesiredState { + Running, + Paused, + LoggedOut, + AuthenticationBlocked, + Shutdown, +} + +enum LoginEvent { + Cancelled, + Command(Option), + Complete(Result, SessionFailure>), +} + +enum ActiveEvent { + Cancelled, + Command(Option), + Signal(SessionSignal), +} + +struct Runtime { + backend: Arc, + policy: ReconnectPolicy, + commands: mpsc::Receiver, + observations: mpsc::Sender, + cancellation: CancellationTokenSource, + state: Arc, + status: Arc>, + fence: Arc, + queued: VecDeque, + queued_ids: BTreeSet, + seen: VecDeque, + seen_ids: BTreeSet, + next_generation: u64, + failures: u32, +} + +impl Runtime { + #[allow(clippy::too_many_lines)] + async fn run(mut self) { + self.transition(SessionState::Stopped, SessionReason::Startup, None, 0); + let mut desired = DesiredState::Running; + let mut session: Option> = None; + let mut ready = false; + let mut stable_since = None; + + loop { + if self.cancellation.token().is_cancellation_requested() { + desired = DesiredState::Shutdown; + } + match desired { + DesiredState::Shutdown => break, + DesiredState::Paused + | DesiredState::LoggedOut + | DesiredState::AuthenticationBlocked => { + let command = tokio::select! { + () = self.cancellation.token().cancelled() => { + desired = DesiredState::Shutdown; + continue; + } + command = self.commands.recv() => command, + }; + desired = self.handle_idle_command(command, desired); + } + DesiredState::Running if session.is_none() => { + if self.next_generation == u64::MAX { + desired = DesiredState::Shutdown; + continue; + } + let generation = self.next_generation; + self.next_generation += 1; + let token = self.fence.rotate(generation); + self.transition( + SessionState::Connecting, + if self.failures == 0 { + SessionReason::OperatorResume + } else { + SessionReason::TransientTransport + }, + None, + generation, + ); + let login_event = { + let backend = Arc::clone(&self.backend); + let login = backend.login(generation, token); + tokio::pin!(login); + loop { + let event = tokio::select! { + () = self.cancellation.token().cancelled() => LoginEvent::Cancelled, + command = self.commands.recv() => LoginEvent::Command(command), + result = &mut login => LoginEvent::Complete(result), + }; + if let LoginEvent::Command(Some(Command::Submit(work, response))) = + event + { + let disposition = self.handle_work(work.clone(), false, 0); + let _ = response.send(disposition); + self.emit_work(work.id, disposition); + } else { + break event; + } + } + }; + let result = match login_event { + LoginEvent::Cancelled => { + desired = DesiredState::Shutdown; + self.fence.invalidate(); + continue; + } + LoginEvent::Command(command) => { + self.fence.invalidate(); + desired = self.handle_idle_command(command, DesiredState::Running); + continue; + } + LoginEvent::Complete(result) => result, + }; + match result { + Ok(connected) if connected.generation() == generation => { + session = Some(connected); + ready = false; + stable_since = None; + self.transition( + SessionState::Degraded, + SessionReason::LoginSucceeded, + None, + generation, + ); + } + Ok(connected) => { + self.fence.invalidate(); + self.close_session(connected).await; + self.failures = self.failures.saturating_add(1); + desired = self + .backoff(SessionFailure::new(SessionFailureKind::ServerFailure)) + .await; + } + Err(failure) if !failure.retryable() => { + self.fence.invalidate(); + self.transition( + SessionState::AuthenticationBlocked, + failure.reason(), + None, + generation, + ); + desired = DesiredState::AuthenticationBlocked; + } + Err(failure) => { + self.fence.invalidate(); + self.failures = self.failures.saturating_add(1); + desired = self.backoff(failure).await; + } + } + } + DesiredState::Running => { + let generation = self.status_snapshot().generation; + let Some(token) = self.fence.token(generation) else { + session = None; + continue; + }; + let active_event = { + let Some(active) = session.as_mut() else { + continue; + }; + let signal = active.next_signal(token); + tokio::pin!(signal); + tokio::select! { + () = self.cancellation.token().cancelled() => ActiveEvent::Cancelled, + command = self.commands.recv() => ActiveEvent::Command(command), + next = &mut signal => ActiveEvent::Signal(next), + } + }; + match active_event { + ActiveEvent::Cancelled => { + desired = DesiredState::Shutdown; + } + ActiveEvent::Command(command) => match command { + Some(Command::Submit(work, response)) => { + let disposition = self.handle_work(work.clone(), ready, generation); + let _ = response.send(disposition); + self.emit_work(work.id, disposition); + } + None => { + let (next, reconnect) = self.handle_active_control(None); + desired = next; + if reconnect { + self.failures = 0; + } + if desired != DesiredState::Running || reconnect { + self.fence.invalidate(); + } + } + Some(Command::Control(control)) => { + let (next, reconnect) = self.handle_active_control(Some(control)); + desired = next; + if reconnect { + self.failures = 0; + } + if desired != DesiredState::Running || reconnect { + self.fence.invalidate(); + } + } + }, + ActiveEvent::Signal(next) => match next { + SessionSignal::Ready => { + ready = true; + stable_since.get_or_insert_with(Instant::now); + self.transition( + SessionState::Online, + SessionReason::AgentReady, + None, + generation, + ); + self.release_queued(generation); + } + SessionSignal::Degraded => { + ready = false; + stable_since = None; + self.transition( + SessionState::Degraded, + SessionReason::ReadinessLost, + None, + generation, + ); + } + SessionSignal::Disconnected(failure) => { + ready = false; + self.fence.invalidate(); + if stable_since.is_some_and(|start| { + start.elapsed() >= self.policy.stable_reset_after + }) { + self.failures = 0; + self.transition( + SessionState::Degraded, + SessionReason::StableSessionReset, + None, + generation, + ); + } + self.failures = self.failures.saturating_add(1); + if let Some(disconnected) = session.take() { + self.close_session(disconnected).await; + } + desired = self.backoff(failure).await; + stable_since = None; + } + }, + } + if session.is_some() + && (desired != DesiredState::Running + || self.fence.token(generation).is_none()) + { + if let Some(active) = session.take() { + self.close_session(active).await; + } + ready = false; + stable_since = None; + } + } + } + } + + self.transition( + SessionState::ShuttingDown, + SessionReason::ShutdownRequested, + None, + self.status_snapshot().generation, + ); + self.fence.invalidate(); + if let Some(active) = session.take() { + self.close_session(active).await; + } + self.reject_queued(); + let cleanup = CancellationTokenSource::new(); + let _ = tokio::time::timeout( + self.cleanup_timeout(), + self.backend.flush_audit(cleanup.token()), + ) + .await; + cleanup.cancel(); + self.transition( + SessionState::Stopped, + SessionReason::ShutdownComplete, + None, + 0, + ); + } + + fn handle_idle_command( + &mut self, + command: Option, + current: DesiredState, + ) -> DesiredState { + match command { + None | Some(Command::Control(SessionControl::Shutdown)) => DesiredState::Shutdown, + Some(Command::Control(SessionControl::Pause)) => { + self.transition(SessionState::Paused, SessionReason::OperatorPause, None, 0); + DesiredState::Paused + } + Some(Command::Control(SessionControl::Resume | SessionControl::ForceReconnect)) => { + self.failures = 0; + DesiredState::Running + } + Some(Command::Control(SessionControl::Logout)) => { + self.transition( + SessionState::Stopped, + SessionReason::OperatorLogout, + None, + 0, + ); + DesiredState::LoggedOut + } + Some(Command::Submit(work, response)) => { + let disposition = self.handle_work(work.clone(), false, 0); + let _ = response.send(disposition); + self.emit_work(work.id, disposition); + current + } + } + } + + fn handle_active_control(&mut self, command: Option) -> (DesiredState, bool) { + match command { + None | Some(SessionControl::Shutdown) => (DesiredState::Shutdown, false), + Some(SessionControl::Pause) => { + self.transition(SessionState::Paused, SessionReason::OperatorPause, None, 0); + (DesiredState::Paused, false) + } + Some(SessionControl::Resume) => (DesiredState::Running, false), + Some(SessionControl::ForceReconnect) => { + self.transition( + SessionState::Connecting, + SessionReason::OperatorForceReconnect, + None, + 0, + ); + (DesiredState::Running, true) + } + Some(SessionControl::Logout) => { + self.transition( + SessionState::Stopped, + SessionReason::OperatorLogout, + None, + 0, + ); + (DesiredState::LoggedOut, false) + } + } + } + + async fn backoff(&mut self, failure: SessionFailure) -> DesiredState { + let delay = self + .policy + .retry_delay(self.failures, failure.retry_after()); + self.transition(SessionState::Backoff, failure.reason(), Some(delay), 0); + let sleep = tokio::time::sleep(delay); + tokio::pin!(sleep); + loop { + tokio::select! { + () = self.cancellation.token().cancelled() => return DesiredState::Shutdown, + () = &mut sleep => return DesiredState::Running, + command = self.commands.recv() => match command { + Some(Command::Submit(work, response)) => { + let disposition = self.handle_work(work.clone(), false, 0); + let _ = response.send(disposition); + self.emit_work(work.id, disposition); + } + other => return self.handle_idle_command(other, DesiredState::Running), + } + } + } + } + + async fn close_session(&self, session: Box) { + let cleanup = CancellationTokenSource::new(); + let _ = tokio::time::timeout(self.cleanup_timeout(), session.logout(cleanup.token())).await; + cleanup.cancel(); + } + + fn cleanup_timeout(&self) -> Duration { + (self.policy.shutdown_deadline / 3).max(Duration::from_millis(1)) + } + + fn handle_work(&mut self, work: SessionWork, ready: bool, generation: u64) -> WorkDisposition { + if self.cancellation.token().is_cancellation_requested() { + return WorkDisposition::RejectedShuttingDown; + } + if self.seen_ids.contains(work.id.as_str()) || self.queued_ids.contains(work.id.as_str()) { + return WorkDisposition::RejectedDuplicate; + } + if ready { + self.remember_work(work.id.as_str().to_owned()); + return WorkDisposition::Accepted { generation }; + } + if work.kind != WorkKind::ReadOnly { + return WorkDisposition::RejectedOfflineMutation; + } + if self.queued.len() == self.policy.offline_work_capacity { + return WorkDisposition::RejectedQueueFull; + } + self.queued_ids.insert(work.id.as_str().to_owned()); + self.queued.push_back(work); + WorkDisposition::Queued + } + + fn release_queued(&mut self, generation: u64) { + while let Some(work) = self.queued.pop_front() { + self.queued_ids.remove(work.id.as_str()); + self.remember_work(work.id.as_str().to_owned()); + self.emit_work(work.id, WorkDisposition::Accepted { generation }); + } + } + + fn remember_work(&mut self, id: String) { + if self.seen.len() == MAX_SEEN_WORK + && let Some(expired) = self.seen.pop_front() + { + self.seen_ids.remove(&expired); + } + self.seen_ids.insert(id.clone()); + self.seen.push_back(id); + } + + fn reject_queued(&mut self) { + while let Some(work) = self.queued.pop_front() { + self.queued_ids.remove(work.id.as_str()); + self.emit_work(work.id, WorkDisposition::RejectedShuttingDown); + } + } + + fn transition( + &self, + state: SessionState, + reason: SessionReason, + retry_in: Option, + generation: u64, + ) { + self.state.store(state as u8, Ordering::Release); + let status = SessionStatus { + state, + generation, + transport_connected: matches!(state, SessionState::Online | SessionState::Degraded), + agent_ready: state == SessionState::Online, + consecutive_failures: self.failures, + }; + *lock(&self.status) = status; + let _ = self.observations.try_send(SessionObservation::Transition { + status, + reason, + retry_in, + }); + } + + fn emit_work(&self, id: BoundedText, disposition: WorkDisposition) { + let _ = self + .observations + .try_send(SessionObservation::Work { id, disposition }); + } + + fn status_snapshot(&self) -> SessionStatus { + *lock(&self.status) + } +} + +fn lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} diff --git a/crates/metacrate-grid-agent/src/session_tests.rs b/crates/metacrate-grid-agent/src/session_tests.rs new file mode 100644 index 0000000..f7cdb70 --- /dev/null +++ b/crates/metacrate-grid-agent/src/session_tests.rs @@ -0,0 +1,526 @@ +use crate::session::*; +use libremetaverse_types::compat::CancellationToken; +use std::collections::VecDeque; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +#[derive(Clone)] +enum LoginPlan { + Failure { + after: Duration, + failure: SessionFailure, + }, + Success { + after: Duration, + signals: VecDeque<(Duration, SessionSignal)>, + }, +} + +#[derive(Default)] +struct FakeStats { + logins: AtomicUsize, + logouts: AtomicUsize, + flushes: AtomicUsize, + active_sessions: AtomicUsize, + active_callbacks: AtomicUsize, + active_workers: AtomicUsize, + active_sockets: AtomicUsize, + max_sessions: AtomicUsize, + max_callbacks: AtomicUsize, + max_workers: AtomicUsize, + max_sockets: AtomicUsize, +} + +impl FakeStats { + fn acquire(&self) { + update_max(&self.active_sessions, &self.max_sessions, 1); + update_max(&self.active_callbacks, &self.max_callbacks, 4); + update_max(&self.active_workers, &self.max_workers, 2); + update_max(&self.active_sockets, &self.max_sockets, 1); + } + + fn release(&self) { + self.active_sessions.fetch_sub(1, Ordering::AcqRel); + self.active_callbacks.fetch_sub(4, Ordering::AcqRel); + self.active_workers.fetch_sub(2, Ordering::AcqRel); + self.active_sockets.fetch_sub(1, Ordering::AcqRel); + } +} + +fn update_max(active: &AtomicUsize, maximum: &AtomicUsize, amount: usize) { + let next = active.fetch_add(amount, Ordering::AcqRel) + amount; + maximum.fetch_max(next, Ordering::AcqRel); +} + +struct FakeBackend { + plans: Mutex>, + stats: Arc, +} + +impl FakeBackend { + fn new(plans: impl IntoIterator) -> (Arc, Arc) { + let stats = Arc::new(FakeStats::default()); + ( + Arc::new(Self { + plans: Mutex::new(plans.into_iter().collect()), + stats: Arc::clone(&stats), + }), + stats, + ) + } +} + +impl GridSessionBackend for FakeBackend { + fn login( + &self, + generation: u64, + cancellation: CancellationToken, + ) -> SessionFuture<'_, Result, SessionFailure>> { + self.stats.logins.fetch_add(1, Ordering::AcqRel); + let plan = self + .plans + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .pop_front() + .unwrap_or(LoginPlan::Success { + after: Duration::ZERO, + signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]), + }); + let stats = Arc::clone(&self.stats); + Box::pin(async move { + let after = match &plan { + LoginPlan::Failure { after, .. } | LoginPlan::Success { after, .. } => *after, + }; + tokio::select! { + () = tokio::time::sleep(after) => {} + () = cancellation.cancelled() => { + return Err(SessionFailure::new(SessionFailureKind::TransientTransport)); + } + } + match plan { + LoginPlan::Failure { failure, .. } => Err(failure), + LoginPlan::Success { signals, .. } => { + stats.acquire(); + let session: Box = Box::new(FakeSession { + generation, + signals, + stats, + released: false, + }); + Ok(session) + } + } + }) + } + + fn flush_audit( + &self, + _cancellation: CancellationToken, + ) -> SessionFuture<'_, Result<(), SessionFailure>> { + self.stats.flushes.fetch_add(1, Ordering::AcqRel); + Box::pin(async { Ok(()) }) + } +} + +struct FakeSession { + generation: u64, + signals: VecDeque<(Duration, SessionSignal)>, + stats: Arc, + released: bool, +} + +impl FakeSession { + fn release(&mut self) { + if !self.released { + self.released = true; + self.stats.release(); + } + } +} + +impl GridSession for FakeSession { + fn generation(&self) -> u64 { + self.generation + } + + fn next_signal(&mut self, cancellation: CancellationToken) -> SessionFuture<'_, SessionSignal> { + let next = self.signals.pop_front(); + Box::pin(async move { + if let Some((after, signal)) = next { + tokio::select! { + () = tokio::time::sleep(after) => signal, + () = cancellation.cancelled() => SessionSignal::Disconnected( + SessionFailure::new(SessionFailureKind::TransientTransport) + ), + } + } else { + cancellation.cancelled().await; + SessionSignal::Disconnected(SessionFailure::new( + SessionFailureKind::TransientTransport, + )) + } + }) + } + + fn logout( + mut self: Box, + _cancellation: CancellationToken, + ) -> SessionFuture<'static, Result<(), SessionFailure>> { + self.stats.logouts.fetch_add(1, Ordering::AcqRel); + self.release(); + Box::pin(async { Ok(()) }) + } +} + +impl Drop for FakeSession { + fn drop(&mut self) { + self.release(); + } +} + +fn ready_then(after: Duration, failure: SessionFailureKind) -> LoginPlan { + LoginPlan::Success { + after: Duration::ZERO, + signals: VecDeque::from([ + (Duration::ZERO, SessionSignal::Ready), + ( + after, + SessionSignal::Disconnected(SessionFailure::new(failure)), + ), + ]), + } +} + +fn test_policy() -> ReconnectPolicy { + ReconnectPolicy { + initial_delay: Duration::from_secs(1), + maximum_delay: Duration::from_secs(8), + stable_reset_after: Duration::from_secs(10), + shutdown_deadline: Duration::from_secs(2), + jitter_basis_points: 0, + instance_seed: 7, + offline_work_capacity: 2, + } +} + +#[test] +fn retry_hints_are_minimums_and_all_delays_remain_capped() { + let mut policy = test_policy(); + policy.jitter_basis_points = 5_000; + for seed in 0..128 { + policy.instance_seed = seed; + let hinted = Duration::from_secs(6); + let delay = policy.retry_delay(1, Some(hinted)); + assert!(delay >= hinted); + assert!(delay <= policy.maximum_delay); + assert!(policy.retry_delay(64, None) <= policy.maximum_delay); + } +} + +fn start(plans: impl IntoIterator) -> (SessionSupervisorHandle, Arc) { + let (backend, stats) = FakeBackend::new(plans); + let erased: Arc = backend; + let handle = SessionSupervisor::new(erased, test_policy(), 32, 256) + .expect("supervisor") + .start(); + (handle, stats) +} + +async fn settle() { + for _ in 0..20 { + tokio::task::yield_now().await; + } +} + +async fn wait_state(handle: &SessionSupervisorHandle, expected: SessionState) { + for _ in 0..100 { + if handle.state() == expected { + return; + } + tokio::task::yield_now().await; + } + panic!("expected {expected:?}, got {:?}", handle.status()); +} + +#[tokio::test(start_paused = true)] +async fn paused_time_classifies_failures_backoff_flapping_and_stable_reset() { + let plans = [ + LoginPlan::Failure { + after: Duration::ZERO, + failure: SessionFailure::new(SessionFailureKind::TransientTransport), + }, + LoginPlan::Failure { + after: Duration::ZERO, + failure: SessionFailure::with_retry_after( + SessionFailureKind::Maintenance, + Duration::from_secs(5), + ), + }, + ready_then(Duration::from_secs(1), SessionFailureKind::Kicked), + ready_then( + Duration::from_secs(11), + SessionFailureKind::SimulatorDisconnected, + ), + LoginPlan::Success { + after: Duration::ZERO, + signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]), + }, + ]; + let (mut handle, stats) = start(plans); + wait_state(&handle, SessionState::Backoff).await; + assert_eq!(handle.status().consecutive_failures, 1); + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + wait_state(&handle, SessionState::Backoff).await; + assert_eq!(handle.status().consecutive_failures, 2); + tokio::time::advance(Duration::from_secs(5)).await; + settle().await; + wait_state(&handle, SessionState::Online).await; + let first_online = handle.status().generation; + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + wait_state(&handle, SessionState::Backoff).await; + assert!(!handle.accepts_result(first_online)); + tokio::time::advance(Duration::from_secs(4)).await; + settle().await; + wait_state(&handle, SessionState::Online).await; + tokio::time::advance(Duration::from_secs(11)).await; + settle().await; + wait_state(&handle, SessionState::Backoff).await; + assert_eq!(handle.status().consecutive_failures, 1); + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + wait_state(&handle, SessionState::Online).await; + handle.shutdown().await.expect("shutdown"); + assert_eq!(stats.flushes.load(Ordering::Acquire), 1); +} + +#[tokio::test(start_paused = true)] +async fn invalid_credentials_and_configuration_block_until_operator_reconnect() { + for failure in [ + SessionFailureKind::InvalidCredentials, + SessionFailureKind::InvalidConfiguration, + ] { + let plans = [ + LoginPlan::Failure { + after: Duration::ZERO, + failure: SessionFailure::new(failure), + }, + LoginPlan::Success { + after: Duration::ZERO, + signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]), + }, + ]; + let (mut handle, stats) = start(plans); + wait_state(&handle, SessionState::AuthenticationBlocked).await; + tokio::time::advance(Duration::from_hours(1)).await; + settle().await; + assert_eq!(stats.logins.load(Ordering::Acquire), 1); + handle + .control(SessionControl::ForceReconnect) + .await + .expect("force reconnect"); + settle().await; + wait_state(&handle, SessionState::Online).await; + handle.shutdown().await.expect("shutdown"); + } +} + +#[tokio::test(start_paused = true)] +async fn manual_pause_logout_reconnect_and_offline_work_are_predictable() { + let (mut handle, stats) = start([LoginPlan::Success { + after: Duration::ZERO, + signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]), + }]); + wait_state(&handle, SessionState::Online).await; + let generation = handle.status().generation; + let token = handle + .generation_token(generation) + .expect("generation token"); + handle.control(SessionControl::Pause).await.expect("pause"); + settle().await; + wait_state(&handle, SessionState::Paused).await; + assert!(token.is_cancellation_requested()); + assert!(!handle.accepts_result(generation)); + assert_eq!( + handle + .submit(SessionWork::new("mutate", WorkKind::IdempotentMutation).expect("work")) + .await + .expect("decision"), + WorkDisposition::RejectedOfflineMutation + ); + assert_eq!( + handle + .submit(SessionWork::new("read", WorkKind::ReadOnly).expect("work")) + .await + .expect("decision"), + WorkDisposition::Queued + ); + handle + .control(SessionControl::Resume) + .await + .expect("resume"); + settle().await; + wait_state(&handle, SessionState::Online).await; + assert_ne!(handle.status().generation, generation); + assert_eq!( + handle + .submit(SessionWork::new("read", WorkKind::ReadOnly).expect("work")) + .await + .expect("dedupe decision"), + WorkDisposition::RejectedDuplicate + ); + let before_force = handle.status().generation; + handle + .control(SessionControl::ForceReconnect) + .await + .expect("active reconnect"); + settle().await; + wait_state(&handle, SessionState::Online).await; + assert_ne!(handle.status().generation, before_force); + handle + .control(SessionControl::Logout) + .await + .expect("logout"); + settle().await; + wait_state(&handle, SessionState::Stopped).await; + handle + .control(SessionControl::ForceReconnect) + .await + .expect("reconnect"); + settle().await; + wait_state(&handle, SessionState::Online).await; + handle.shutdown().await.expect("shutdown"); + assert_eq!(stats.active_sessions.load(Ordering::Acquire), 0); +} + +#[tokio::test(start_paused = true)] +async fn transport_connection_is_distinct_from_full_agent_readiness() { + let (mut handle, _) = start([LoginPlan::Success { + after: Duration::ZERO, + signals: VecDeque::from([(Duration::from_secs(10), SessionSignal::Ready)]), + }]); + wait_state(&handle, SessionState::Degraded).await; + let degraded = handle.status(); + assert!(degraded.transport_connected); + assert!(!degraded.agent_ready); + assert!(!handle.accepts_result(degraded.generation)); + tokio::time::advance(Duration::from_secs(10)).await; + settle().await; + wait_state(&handle, SessionState::Online).await; + assert!(handle.status().agent_ready); + handle.shutdown().await.expect("shutdown"); +} + +#[tokio::test(start_paused = true)] +async fn shutdown_covers_degraded_authentication_blocked_stopped_and_online_states() { + let cases = [ + ( + LoginPlan::Success { + after: Duration::ZERO, + signals: VecDeque::from([(Duration::from_hours(1), SessionSignal::Ready)]), + }, + SessionState::Degraded, + ), + ( + LoginPlan::Failure { + after: Duration::ZERO, + failure: SessionFailure::new(SessionFailureKind::InvalidCredentials), + }, + SessionState::AuthenticationBlocked, + ), + ( + LoginPlan::Success { + after: Duration::ZERO, + signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]), + }, + SessionState::Online, + ), + ]; + for (plan, expected) in cases { + let (mut handle, stats) = start([plan]); + wait_state(&handle, expected).await; + if expected == SessionState::Online { + handle + .control(SessionControl::Logout) + .await + .expect("logout"); + settle().await; + wait_state(&handle, SessionState::Stopped).await; + } + handle.shutdown().await.expect("shutdown"); + assert_eq!(stats.active_sessions.load(Ordering::Acquire), 0); + assert_eq!(stats.flushes.load(Ordering::Acquire), 1); + } +} + +#[tokio::test(start_paused = true)] +async fn fake_grid_reconnects_keep_exactly_one_resource_set_and_no_stale_execution() { + let plans = [ + ready_then(Duration::from_secs(1), SessionFailureKind::Maintenance), + ready_then(Duration::from_secs(1), SessionFailureKind::Kicked), + LoginPlan::Success { + after: Duration::ZERO, + signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]), + }, + ]; + let (mut handle, stats) = start(plans); + wait_state(&handle, SessionState::Online).await; + let stale = handle.status().generation; + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + wait_state(&handle, SessionState::Online).await; + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + tokio::time::advance(Duration::from_secs(2)).await; + settle().await; + wait_state(&handle, SessionState::Online).await; + assert!(!handle.accepts_result(stale)); + assert!(handle.accepts_result(handle.status().generation)); + assert_eq!(stats.max_sessions.load(Ordering::Acquire), 1); + assert_eq!(stats.max_callbacks.load(Ordering::Acquire), 4); + assert_eq!(stats.max_workers.load(Ordering::Acquire), 2); + assert_eq!(stats.max_sockets.load(Ordering::Acquire), 1); + handle.shutdown().await.expect("shutdown"); + assert_eq!(stats.active_sessions.load(Ordering::Acquire), 0); + assert_eq!(stats.active_callbacks.load(Ordering::Acquire), 0); + assert_eq!(stats.active_workers.load(Ordering::Acquire), 0); + assert_eq!(stats.active_sockets.load(Ordering::Acquire), 0); + assert_eq!(stats.logouts.load(Ordering::Acquire), 3); + assert_eq!(stats.flushes.load(Ordering::Acquire), 1); +} + +#[tokio::test(start_paused = true)] +async fn shutdown_is_cancellation_driven_during_connect_backoff_paused_and_online() { + let scenarios = [ + LoginPlan::Success { + after: Duration::from_hours(1), + signals: VecDeque::new(), + }, + LoginPlan::Failure { + after: Duration::ZERO, + failure: SessionFailure::new(SessionFailureKind::TransientTransport), + }, + LoginPlan::Success { + after: Duration::ZERO, + signals: VecDeque::from([(Duration::ZERO, SessionSignal::Ready)]), + }, + ]; + for (index, plan) in scenarios.into_iter().enumerate() { + let (mut handle, stats) = start([plan]); + settle().await; + if index == 2 { + wait_state(&handle, SessionState::Online).await; + handle.control(SessionControl::Pause).await.expect("pause"); + settle().await; + wait_state(&handle, SessionState::Paused).await; + } + handle.shutdown().await.expect("bounded shutdown"); + assert_eq!(handle.state(), SessionState::Stopped); + assert_eq!(stats.active_sessions.load(Ordering::Acquire), 0); + assert_eq!(stats.flushes.load(Ordering::Acquire), 1); + } +} diff --git a/crates/metacrate-grid-agent/tests/dependency_policy.rs b/crates/metacrate-grid-agent/tests/dependency_policy.rs index 0595dfd..0e13be1 100644 --- a/crates/metacrate-grid-agent/tests/dependency_policy.rs +++ b/crates/metacrate-grid-agent/tests/dependency_policy.rs @@ -46,7 +46,7 @@ fn runtime_source_has_no_subprocess_or_native_abi_escape_hatch() { let mut files = Vec::with_capacity(8); collect_rust_files(&source, &mut files); assert!( - files.len() <= 10, + files.len() <= 12, "source-file count needs a reviewed bound update" ); for path in files { @@ -86,6 +86,25 @@ fn world_backend_requires_the_opaque_policy_authorization() { assert!(!backend.contains("decision: PolicyDecision")); } +#[test] +fn live_session_adapter_reuses_only_the_native_network_manager_lifecycle() { + let root = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + let backend = fs::read_to_string(root.join("src/backend.rs")).expect("backend source"); + for required in [ + ".native_default_login_params(", + ".native_login(", + ".native_subscribe_disconnected(", + ".native_subscribe_event_queue_running(", + ".native_logout_async(", + ] { + assert!( + backend.contains(required), + "missing native lifecycle {required}" + ); + } + assert!(!backend.contains("reqwest")); +} + fn collect_rust_files(directory: &Path, output: &mut Vec) { for entry in fs::read_dir(directory).expect("read source directory") { let path = entry.expect("source entry").path(); diff --git a/crates/metacrate-grid-agent/tests/session_supervisor.rs b/crates/metacrate-grid-agent/tests/session_supervisor.rs new file mode 100644 index 0000000..614b4c6 --- /dev/null +++ b/crates/metacrate-grid-agent/tests/session_supervisor.rs @@ -0,0 +1,227 @@ +use libremetaverse_types::compat::CancellationToken; +use metacrate_grid_agent::{ + GridSession, GridSessionBackend, ReconnectPolicy, SessionFailure, SessionFailureKind, + SessionFuture, SessionSignal, SessionState, SessionSupervisor, SessionSupervisorHandle, + SessionWork, WorkDisposition, WorkKind, +}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + +#[derive(Default)] +struct ResourceStats { + attempts: AtomicUsize, + logouts: AtomicUsize, + flushes: AtomicUsize, + sessions: AtomicUsize, + callbacks: AtomicUsize, + workers: AtomicUsize, + sockets: AtomicUsize, + max_sessions: AtomicUsize, + max_callbacks: AtomicUsize, + max_workers: AtomicUsize, + max_sockets: AtomicUsize, +} + +impl ResourceStats { + fn acquire(&self) { + acquire(&self.sessions, &self.max_sessions, 1); + acquire(&self.callbacks, &self.max_callbacks, 3); + acquire(&self.workers, &self.max_workers, 2); + acquire(&self.sockets, &self.max_sockets, 1); + } + + fn release(&self) { + self.sessions.fetch_sub(1, Ordering::AcqRel); + self.callbacks.fetch_sub(3, Ordering::AcqRel); + self.workers.fetch_sub(2, Ordering::AcqRel); + self.sockets.fetch_sub(1, Ordering::AcqRel); + } +} + +fn acquire(active: &AtomicUsize, maximum: &AtomicUsize, amount: usize) { + let next = active.fetch_add(amount, Ordering::AcqRel) + amount; + maximum.fetch_max(next, Ordering::AcqRel); +} + +struct ReconnectingFakeGrid { + stats: Arc, +} + +impl GridSessionBackend for ReconnectingFakeGrid { + fn login( + &self, + generation: u64, + _cancellation: CancellationToken, + ) -> SessionFuture<'_, Result, SessionFailure>> { + let attempt = self.stats.attempts.fetch_add(1, Ordering::AcqRel) + 1; + self.stats.acquire(); + let stats = Arc::clone(&self.stats); + Box::pin(async move { + let session: Box = Box::new(FakeConnectedSession { + generation, + attempt, + step: 0, + stats, + released: false, + }); + Ok(session) + }) + } + + fn flush_audit( + &self, + _cancellation: CancellationToken, + ) -> SessionFuture<'_, Result<(), SessionFailure>> { + self.stats.flushes.fetch_add(1, Ordering::AcqRel); + Box::pin(async { Ok(()) }) + } +} + +struct FakeConnectedSession { + generation: u64, + attempt: usize, + step: u8, + stats: Arc, + released: bool, +} + +impl FakeConnectedSession { + fn release(&mut self) { + if !self.released { + self.released = true; + self.stats.release(); + } + } +} + +impl GridSession for FakeConnectedSession { + fn generation(&self) -> u64 { + self.generation + } + + fn next_signal(&mut self, cancellation: CancellationToken) -> SessionFuture<'_, SessionSignal> { + let attempt = self.attempt; + let step = self.step; + self.step = self.step.saturating_add(1); + Box::pin(async move { + if step == 0 { + return SessionSignal::Ready; + } + if attempt <= 2 { + tokio::select! { + () = tokio::time::sleep(Duration::from_secs(1)) => { + let kind = if attempt == 1 { + SessionFailureKind::Maintenance + } else { + SessionFailureKind::Kicked + }; + SessionSignal::Disconnected(SessionFailure::new(kind)) + } + () = cancellation.cancelled() => SessionSignal::Disconnected( + SessionFailure::new(SessionFailureKind::TransientTransport), + ), + } + } else { + cancellation.cancelled().await; + SessionSignal::Disconnected(SessionFailure::new( + SessionFailureKind::TransientTransport, + )) + } + }) + } + + fn logout( + mut self: Box, + _cancellation: CancellationToken, + ) -> SessionFuture<'static, Result<(), SessionFailure>> { + self.stats.logouts.fetch_add(1, Ordering::AcqRel); + self.release(); + Box::pin(async { Ok(()) }) + } +} + +impl Drop for FakeConnectedSession { + fn drop(&mut self) { + self.release(); + } +} + +async fn settle() { + for _ in 0..30 { + tokio::task::yield_now().await; + } +} + +async fn wait_online(handle: &SessionSupervisorHandle) { + for _ in 0..100 { + if handle.state() == SessionState::Online { + return; + } + tokio::task::yield_now().await; + } + panic!("fake grid did not become ready: {:?}", handle.status()); +} + +#[tokio::test(start_paused = true)] +async fn repeated_reconnects_have_one_resource_set_no_replay_and_no_shutdown_leaks() { + let stats = Arc::new(ResourceStats::default()); + let backend: Arc = Arc::new(ReconnectingFakeGrid { + stats: Arc::clone(&stats), + }); + let policy = ReconnectPolicy { + initial_delay: Duration::from_secs(1), + maximum_delay: Duration::from_secs(4), + stable_reset_after: Duration::from_secs(30), + shutdown_deadline: Duration::from_secs(3), + jitter_basis_points: 0, + instance_seed: 1, + offline_work_capacity: 8, + }; + let mut handle = SessionSupervisor::new(backend, policy, 32, 256) + .expect("supervisor") + .start(); + + wait_online(&handle).await; + let stale_generation = handle.status().generation; + assert_eq!( + handle + .submit(SessionWork::new("response-1", WorkKind::ReadOnly).expect("work")) + .await + .expect("accepted"), + WorkDisposition::Accepted { + generation: stale_generation + } + ); + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + wait_online(&handle).await; + tokio::time::advance(Duration::from_secs(1)).await; + settle().await; + tokio::time::advance(Duration::from_secs(2)).await; + settle().await; + wait_online(&handle).await; + + assert!(!handle.accepts_result(stale_generation)); + assert_eq!( + handle + .submit(SessionWork::new("response-1", WorkKind::ReadOnly).expect("work")) + .await + .expect("dedupe"), + WorkDisposition::RejectedDuplicate + ); + assert_eq!(stats.max_sessions.load(Ordering::Acquire), 1); + assert_eq!(stats.max_callbacks.load(Ordering::Acquire), 3); + assert_eq!(stats.max_workers.load(Ordering::Acquire), 2); + assert_eq!(stats.max_sockets.load(Ordering::Acquire), 1); + + handle.shutdown().await.expect("shutdown"); + assert_eq!(stats.sessions.load(Ordering::Acquire), 0); + assert_eq!(stats.callbacks.load(Ordering::Acquire), 0); + assert_eq!(stats.workers.load(Ordering::Acquire), 0); + assert_eq!(stats.sockets.load(Ordering::Acquire), 0); + assert_eq!(stats.logouts.load(Ordering::Acquire), 3); + assert_eq!(stats.flushes.load(Ordering::Acquire), 1); +} diff --git a/docs/grid-agent-architecture.md b/docs/grid-agent-architecture.md index dd547b9..f381f82 100644 --- a/docs/grid-agent-architecture.md +++ b/docs/grid-agent-architecture.md @@ -10,18 +10,21 @@ platform-specific core path. ## Ownership and bounds -`AgentService::start` is the sole task-creation point. It validates the complete -configuration before allocating channels, calling a backend, or permitting -network access. `ServiceHandle` then exclusively owns cancellation, both join -handles, the control sender, and observable receiver. +`AgentService::start` owns offline task creation, while +`SessionSupervisor::start` owns the live lifecycle task. Configuration is +validated before either allocates channels, calls a backend, or permits network +access. Their handles exclusively own cancellation, join handles, controls, and +observable receivers. | Resource | Owner | Hard/configured bound | Backpressure/termination | | --- | --- | --- | --- | | Grid-event queue | coordinator receives; backend sends | 8,192 / `grid_event_queue` | async send or cancellation | | Control queue | coordinator receives; handle sends | 256 / `control_queue` | async send; closed after stop | | Observable queue | handle receives; coordinator/backend send | 8,192 / `observable_queue` | async send or cancellation | -| Backend task | `ServiceHandle.tasks[0]` | exactly one | shared cancellation token, joined first | +| Backend task | `ServiceHandle.tasks[0]` | exactly one in offline mode | shared cancellation token, joined first | | Coordinator task | `ServiceHandle.tasks[1]` | exactly one | shared cancellation token, joined second | +| Live session supervisor | `SessionSupervisorHandle` | one owner; one active generation/session | generation cancellation, exact-once logout, bounded join | +| Offline session work | live supervisor | 1,024 hard / `reconnect.offline_work_capacity` | read-only queue; mutations rejected while not ready | | Body / message | typed boundary owners | 8 MiB / 64 KiB hard ceilings, with lower configured limits | rejected before enqueue | | Conversation / tool calls | request owner | 256 messages / 64 calls, with lower configured limits | rejected before request | | LLM request slots | shared `LlmClient` semaphore | 256 hard / configured concurrent requests | async acquire or cancellation | @@ -36,7 +39,7 @@ type. Dynamic configuration can lower these absolute ceilings but cannot raise them. Backend error text is not forwarded; observations publish a fixed bounded diagnostic. -## State machine and shutdown +## State machines and shutdown The explicit service states are `starting -> running <-> paused -> stopping -> stopped`, with `failed` reserved for task failure. The offline backend publishes @@ -54,6 +57,13 @@ feature-enabled live adapter drops its `LibremetaverseClientOwner` last, invoking the existing client ownership shutdown. +Live modes use `stopped`, `connecting`, `degraded` (transport connected but not +fully ready), `online`, `backoff`, `authentication-blocked`, `paused`, and +`shutting-down`. Every attempt rotates a generation cancellation token. +Disconnect, pause, logout, force reconnect, or shutdown invalidates it before +cleanup, fencing old events and late LLM/tool results. See +[`grid-agent-session.md`](grid-agent-session.md). + ## Trust boundaries - JSON configuration and environment text are untrusted. Unknown fields, @@ -83,7 +93,7 @@ shutdown. metacrate-grid-agent binary (portable Ctrl-C + config path) | v -AgentService -> bounded Tokio channels/tasks -> injected GridBackend +offline AgentService -> bounded Tokio channels/tasks -> injected GridBackend | | v v typed config/events/policy boundaries live-grid feature boundary @@ -104,3 +114,5 @@ The precise LLM compatibility and cancellation contract is documented in [`grid-agent-llm.md`](grid-agent-llm.md). The origin/capability matrix and opaque mutation boundary are documented in [`grid-agent-policy.md`](grid-agent-policy.md). +The live lifecycle and generation contract is documented in +[`grid-agent-session.md`](grid-agent-session.md). diff --git a/docs/grid-agent-session.md b/docs/grid-agent-session.md new file mode 100644 index 0000000..ad2190c --- /dev/null +++ b/docs/grid-agent-session.md @@ -0,0 +1,75 @@ +# Grid-session supervision + +Live integrated and split modes run through `SessionSupervisor`. Its injected +`GridSessionBackend` makes the lifecycle deterministic under a fake grid while +the `live-grid` adapter reuses `libremetaverse::NetworkManager` for native +login, event-queue readiness, disconnect callbacks, and logout. + +The native login does not yield a session until the login response has populated +the existing inventory skeleton and a current simulator/UDP circuit exists. +The adapter then waits for that simulator's event queue before publishing full +readiness. Existing `GridClient` inventory/world managers and its single +movement-update owner remain the composition owners; the adapter never creates +duplicates. Only its generation-scoped readiness/disconnect subscriptions are +recreated, and their RAII guards are dropped before logout. + +## States and reasons + +| State | Transport connected | Agent ready | Exit condition | +| --- | --- | --- | --- | +| `stopped` | no | no | startup, resume, or force reconnect | +| `connecting` | no | no | login result, operator control, or cancellation | +| `degraded` | yes | no | readiness, disconnect, or operator control | +| `online` | yes | yes | readiness loss, disconnect, or operator control | +| `backoff` | no | no | cancellation-driven timer or operator control | +| `authentication-blocked` | no | no | resume/force reconnect or shutdown | +| `paused` | no | no | resume/force reconnect, logout, or shutdown | +| `shutting-down` | no | no | session logout, audit flush, and joined owner task | + +Every transition carries a stable `SessionReason`; raw login responses, server +messages, credentials, capability URLs, and session tokens cannot enter the +failure or observation types. `SessionStatus` reports transport connectivity +and complete readiness separately. + +Invalid credentials and invalid local login configuration stop automatic +retry. Transport failures, maintenance, kicks, simulator disconnects, and +server failures retry with exponential backoff. `reconnect.maximum_delay_seconds` +is a hard cap. A server retry hint is a minimum up to that cap. Per-instance +jitter is bounded by `jitter_basis_points`, preventing synchronized reconnects, +and a connection that remains up for `stable_reset_seconds` resets the failure +streak. + +Backoff defaults are one second initially, 60 seconds maximum, 20 percent +jitter, and a 120-second stable reset window. All waits use Tokio timers inside +`select!` with cancellation/control; there are no blocking sleeps. + +## Generation and work safety + +Each attempt receives a monotonically changing generation and cancellation +token. Session-scoped subscriptions, readiness ownership, and workers belong to +the returned `GridSession` and are closed once by its consuming `logout` method. +Late inference/tool results are accepted only when `accepts_result` sees their +exact generation in fully ready state. + +While not ready, bounded read-only work may queue. Mutation work—including +nominally idempotent mutation—is rejected, because the supervisor cannot prove +that a previous policy authorization remains valid. Work IDs are deduplicated +across reconnects with a bounded recent-ID set, so reconnect never silently +duplicates an outbound response. Queued reads are released once into the new +generation. + +## Shutdown + +Shutdown first rejects new work and invalidates the generation token, which +cancels inference, tools, and scheduled consumers sharing it. It then consumes +the active session for one logout attempt, waits within +`timeouts.shutdown_seconds`, calls the backend's bounded audit-flush hook, and +joins the supervisor owner task. A late task is aborted and awaited by the +handle, so no task or socket is detached. + +Focused deterministic verification: + +```sh +cargo test --locked -p metacrate-grid-agent --lib session_tests +cargo clippy --locked -p metacrate-grid-agent --all-targets -- -D warnings +```