From 962d17257d657639fa0299030dd32432d8c4c491 Mon Sep 17 00:00:00 2001 From: Chili Palmer Date: Tue, 18 Aug 2026 04:29:12 +0000 Subject: [PATCH] feat(grid-agent): add portable control plane (#126) --- Cargo.lock | 2 + crates/metacrate-grid-agent/Cargo.toml | 2 + crates/metacrate-grid-agent/README.md | 8 +- crates/metacrate-grid-agent/src/behavior.rs | 176 +- .../src/behavior_tests.rs | 42 + crates/metacrate-grid-agent/src/config.rs | 243 +++ .../metacrate-grid-agent/src/control_plane.rs | 1780 +++++++++++++++++ .../src/control_plane_tests.rs | 746 +++++++ .../src/control_runtime.rs | 522 +++++ .../src/control_runtime_tests.rs | 167 ++ crates/metacrate-grid-agent/src/lib.rs | 30 +- crates/metacrate-grid-agent/src/main.rs | 119 +- crates/metacrate-grid-agent/src/policy.rs | 101 +- .../metacrate-grid-agent/src/policy_tests.rs | 19 + .../tests/dependency_policy.rs | 6 +- docs/grid-agent-control-plane.md | 91 + 16 files changed, 4023 insertions(+), 31 deletions(-) create mode 100644 crates/metacrate-grid-agent/src/control_plane.rs create mode 100644 crates/metacrate-grid-agent/src/control_plane_tests.rs create mode 100644 crates/metacrate-grid-agent/src/control_runtime.rs create mode 100644 crates/metacrate-grid-agent/src/control_runtime_tests.rs create mode 100644 docs/grid-agent-control-plane.md diff --git a/Cargo.lock b/Cargo.lock index 7c555bb..06001b6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2201,10 +2201,12 @@ dependencies = [ "libremetaverse", "libremetaverse-types", "reqwest", + "rustls", "serde", "serde_json", "sha2 0.11.0", "tokio", + "tokio-rustls", "url", ] diff --git a/crates/metacrate-grid-agent/Cargo.toml b/crates/metacrate-grid-agent/Cargo.toml index 8e7f50b..d2e0494 100644 --- a/crates/metacrate-grid-agent/Cargo.toml +++ b/crates/metacrate-grid-agent/Cargo.toml @@ -12,10 +12,12 @@ publish = false libremetaverse = { version = "0.0.1", path = "../libremetaverse", default-features = false, optional = true } libremetaverse-types = { version = "0.0.1", path = "../libremetaverse-types" } reqwest = { version = "0.13.4", default-features = false, features = ["rustls"] } +rustls = { version = "0.23.43", default-features = false, features = ["aws_lc_rs", "std", "tls12"] } serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.11" tokio = { version = "1.53.1", features = ["macros", "rt", "sync", "time"] } +tokio-rustls = { version = "0.26.4", default-features = false, features = ["aws_lc_rs", "tls12"] } url = "2.5.8" [dev-dependencies] diff --git a/crates/metacrate-grid-agent/README.md b/crates/metacrate-grid-agent/README.md index a426bdc..75e1819 100644 --- a/crates/metacrate-grid-agent/README.md +++ b/crates/metacrate-grid-agent/README.md @@ -22,7 +22,10 @@ optional JSON file, its referenced secret files, then environment values (an environment-referenced secret file is below a direct environment secret). Supported secret environment variables are `METACRATE_AGENT_LLM_API_KEY[_FILE]` and -`METACRATE_AGENT_GRID_PASSWORD[_FILE]`. Secret files must be bounded regular, +`METACRATE_AGENT_GRID_PASSWORD[_FILE]`. Split control uses the separate +`METACRATE_AGENT_CONTROL_OPERATOR_TOKEN[_FILE]` and optional +`METACRATE_AGENT_CONTROL_OBSERVER_TOKEN[_FILE]`, plus +`METACRATE_AGENT_CONTROL_LISTEN`. Secret files must be bounded regular, non-symlink UTF-8 files containing one line. Operators must restrict their OS ACLs to the service identity; the core uses only portable `std::fs` checks and does not assume Unix permission bits. @@ -51,3 +54,6 @@ shutdown deadline are specified in Per-avatar/channel expiry, compaction, redaction, optional atomic persistence, and metadata-only operator controls are specified in [`../../docs/grid-agent-conversation.md`](../../docs/grid-agent-conversation.md). +The versioned integrated/TCP protocol, roles, framing, bounds, TLS remote-mode +requirements, events, and management methods are specified in +[`../../docs/grid-agent-control-plane.md`](../../docs/grid-agent-control-plane.md). diff --git a/crates/metacrate-grid-agent/src/behavior.rs b/crates/metacrate-grid-agent/src/behavior.rs index 7c15697..f427447 100644 --- a/crates/metacrate-grid-agent/src/behavior.rs +++ b/crates/metacrate-grid-agent/src/behavior.rs @@ -19,8 +19,8 @@ use serde_json::{Map, Value, json}; use std::collections::{BTreeMap, BTreeSet}; use std::error::Error; use std::fmt; -use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tokio::sync::{mpsc, oneshot, watch}; use tokio::task::JoinHandle; @@ -36,6 +36,7 @@ pub const CURRENT_POSE_TOOL: &str = "behavior_current_pose"; const WALK_POLL: Duration = Duration::from_millis(250); const TURN_INTERVAL: Duration = Duration::from_millis(40); +const MAX_ACTION_ID_BYTES: usize = 96; const MAX_TURN_STEP_DEGREES: f64 = 30.0; const ARRIVAL_METERS: f64 = 0.5; const STUCK_PROGRESS_METERS: f64 = 0.05; @@ -196,6 +197,7 @@ pub enum BehaviorObservation { trigger: BehaviorTrigger, }, Action { + action_id: String, generation: Option, region_id: Option, action: String, @@ -243,15 +245,23 @@ impl BehaviorAction { } struct ActionRequest { + action_id: String, action: BehaviorAction, trigger: BehaviorTrigger, policy: BehaviorPolicyResult, reply: oneshot::Sender>, } +#[derive(Default)] +struct ActionRegistry { + active: BTreeSet, + cancelled: BTreeSet, +} + enum Command { Action(ActionRequest), Attention { + action_id: String, delivery_id: String, avatar_id: UUID, reply: oneshot::Sender>, @@ -266,6 +276,10 @@ pub struct BehaviorIngress { ready: watch::Sender>, paused: watch::Sender, emergency: watch::Sender, + action_cancel: watch::Sender, + actions: Arc>, + next_action: Arc, + action_capacity: usize, } impl fmt::Debug for BehaviorIngress { @@ -313,6 +327,28 @@ impl BehaviorIngress { let _ = self.emergency.send(false); } + /// Cancels one exact queued or executing behavior action ID. + pub fn cancel_action(&self, action_id: &str) -> Result { + if !valid_action_id(action_id) { + return Err(BehaviorError::InvalidArguments); + } + let cancelled = { + let mut actions = lock_actions(&self.actions); + if actions.active.contains(action_id) { + actions.cancelled.insert(action_id.to_owned()); + true + } else { + false + } + }; + if cancelled { + self.action_cancel.send_modify(|generation| { + *generation = generation.saturating_add(1); + }); + } + Ok(cancelled) + } + /// Marks a separately authorized scheduler task as roaming. The controller /// itself never invents a roaming route or exposes follow/wander behavior. pub fn set_roaming(&self, roaming: bool) -> Result<(), BehaviorError> { @@ -323,20 +359,30 @@ impl BehaviorIngress { async fn request( &self, + action_id: String, action: BehaviorAction, trigger: BehaviorTrigger, policy: BehaviorPolicyResult, ) -> Result { + if !self.register_action(&action_id) { + return Err(BehaviorError::InvalidArguments); + } let (reply, receive) = oneshot::channel(); - self.commands + if self + .commands .send(Command::Action(ActionRequest { + action_id: action_id.clone(), action, trigger, policy, reply, })) .await - .map_err(|_| BehaviorError::QueueClosed)?; + .is_err() + { + self.unregister_action(&action_id); + return Err(BehaviorError::QueueClosed); + } receive.await.map_err(|_| BehaviorError::QueueClosed)? } @@ -345,17 +391,44 @@ impl BehaviorIngress { delivery_id: String, avatar_id: UUID, ) -> Result<(), BehaviorError> { + let action_id = format!( + "attention-{}", + self.next_action.fetch_add(1, Ordering::Relaxed) + ); + if !self.register_action(&action_id) { + return Err(BehaviorError::InvalidArguments); + } let (reply, receive) = oneshot::channel(); - self.commands + if self + .commands .send(Command::Attention { + action_id: action_id.clone(), delivery_id, avatar_id, reply, }) .await - .map_err(|_| BehaviorError::QueueClosed)?; + .is_err() + { + self.unregister_action(&action_id); + return Err(BehaviorError::QueueClosed); + } receive.await.map_err(|_| BehaviorError::QueueClosed)? } + + fn register_action(&self, action_id: &str) -> bool { + if !valid_action_id(action_id) { + return false; + } + let mut actions = lock_actions(&self.actions); + actions.active.len() < self.action_capacity && actions.active.insert(action_id.to_owned()) + } + + fn unregister_action(&self, action_id: &str) { + let mut actions = lock_actions(&self.actions); + actions.active.remove(action_id); + actions.cancelled.remove(action_id); + } } impl ResponsePacer for BehaviorIngress { @@ -485,12 +558,18 @@ impl BehaviorController { let (ready, ready_rx) = watch::channel(None); let (paused, paused_rx) = watch::channel(false); let (emergency, emergency_rx) = watch::channel(false); + let (action_cancel, action_cancel_rx) = watch::channel(0_u64); + let actions = Arc::new(Mutex::new(ActionRegistry::default())); let (observations, observation_rx) = mpsc::channel(self.observation_capacity); let ingress = BehaviorIngress { commands, ready, paused, emergency, + action_cancel, + actions: actions.clone(), + next_action: Arc::new(AtomicU64::new(1)), + action_capacity: self.queue_capacity, }; let shutdown_timeout = self.shutdown_timeout; let task = tokio::spawn(run_actor( @@ -499,6 +578,8 @@ impl BehaviorController { ready_rx, paused_rx, emergency_rx, + action_cancel_rx, + actions, observations, )); BehaviorHandle { @@ -547,13 +628,15 @@ impl BehaviorRandom for SystemBehaviorRandom { } } -#[allow(clippy::too_many_lines)] // Lifecycle, preemption, idle, and commands share one ordered owner. +#[allow(clippy::too_many_arguments, clippy::too_many_lines)] // One ordered lifecycle owner. async fn run_actor( controller: BehaviorController, mut commands: mpsc::Receiver, mut ready: watch::Receiver>, mut paused: watch::Receiver, mut emergency: watch::Receiver, + action_cancel: watch::Receiver, + actions: Arc>, observations: mpsc::Sender, ) { let mut mode = BehaviorMode::Offline; @@ -616,8 +699,16 @@ async fn run_actor( () = &mut idle, if controller.settings.idle_look_enabled => { if mode == BehaviorMode::Available { let (reply, _) = oneshot::channel(); - let request = ActionRequest { action: BehaviorAction::LookAround, trigger: BehaviorTrigger::IdleTimer, policy: BehaviorPolicyResult::InternalIdle, reply }; - execute_request(&controller, request, &mut mode, &mut last_action, &ready, &paused, &emergency, &observations).await; + let action_id = format!("idle-{}", unix_millis_now()); + let registered = { + let mut registry = lock_actions(&actions); + registry.active.len() < controller.queue_capacity + && registry.active.insert(action_id.clone()) + }; + if registered { + let request = ActionRequest { action_id, action: BehaviorAction::LookAround, trigger: BehaviorTrigger::IdleTimer, policy: BehaviorPolicyResult::InternalIdle, reply }; + execute_request(&controller, request, &mut mode, &mut last_action, &ready, &paused, &emergency, &action_cancel, &actions, &observations).await; + } } idle.as_mut().reset(tokio::time::Instant::now() + controller.settings.idle_interval); } @@ -642,9 +733,10 @@ async fn run_actor( }; transition(&observations, &mut mode, next, BehaviorTrigger::IdleTimer).await; } - Command::Action(request) => execute_request(&controller, request, &mut mode, &mut last_action, &ready, &paused, &emergency, &observations).await, - Command::Attention { delivery_id, avatar_id, reply } => { + Command::Action(request) => execute_request(&controller, request, &mut mode, &mut last_action, &ready, &paused, &emergency, &action_cancel, &actions, &observations).await, + Command::Attention { action_id, delivery_id, avatar_id, reply } => { let request = ActionRequest { + action_id, action: BehaviorAction::FaceAvatar(avatar_id), trigger: BehaviorTrigger::PublicResponse { delivery_id, avatar_id }, policy: BehaviorPolicyResult::BuiltInAttention, @@ -652,7 +744,7 @@ async fn run_actor( }; let delay = random_duration(&controller, controller.settings.response_delay_min, controller.settings.response_delay_max); tokio::time::sleep(delay).await; - execute_request(&controller, request, &mut mode, &mut last_action, &ready, &paused, &emergency, &observations).await; + execute_request(&controller, request, &mut mode, &mut last_action, &ready, &paused, &emergency, &action_cancel, &actions, &observations).await; if mode == BehaviorMode::Engaged { tokio::time::sleep(controller.settings.attention_dwell).await; if !*paused.borrow() && ready.borrow().is_some() { transition(&observations, &mut mode, BehaviorMode::Available, BehaviorTrigger::IdleTimer).await; } @@ -702,12 +794,17 @@ async fn execute_request( ready: &watch::Receiver>, paused: &watch::Receiver, emergency: &watch::Receiver, + action_cancel: &watch::Receiver, + actions: &Arc>, observations: &mpsc::Sender, ) { let started = Instant::now(); + let action_id = request.action_id.clone(); let action_name = request.action.name().to_owned(); let state = *ready.borrow(); - let mut result = if *emergency.borrow() { + let mut result = if take_action_cancellation(actions, &action_id) { + Err(BehaviorError::Cancelled) + } else if *emergency.borrow() { Err(BehaviorError::EmergencyStopped) } else if *paused.borrow() { Err(BehaviorError::Paused) @@ -737,11 +834,15 @@ async fn execute_request( let mut action_ready = ready.clone(); let mut action_paused = paused.clone(); let mut action_emergency = emergency.clone(); + let action_cancel = action_cancel.clone(); + let action_registry = actions.clone(); + let cancelled_id = action_id.clone(); result = tokio::select! { value = tokio::time::timeout(controller.settings.action_timeout, perform_action(controller, state, &request.action)) => value.unwrap_or(Err(BehaviorError::TimedOut)), _ = action_ready.changed() => Err(BehaviorError::Cancelled), _ = action_paused.changed() => Err(BehaviorError::Paused), _ = action_emergency.changed() => Err(BehaviorError::EmergencyStopped), + () = wait_for_action_cancellation(action_cancel, action_registry, cancelled_id) => Err(BehaviorError::Cancelled), }; if result.is_err() && matches!(request.action, BehaviorAction::WalkShort { .. }) { let _ = controller @@ -811,6 +912,7 @@ async fn execute_request( }; let _ = observations .send(BehaviorObservation::Action { + action_id: action_id.clone(), generation: state.map(|value| value.generation), region_id: state.map(|value| value.region_id), action: action_name, @@ -821,6 +923,22 @@ async fn execute_request( }) .await; let _ = request.reply.send(result); + unregister_action(actions, &action_id); +} + +async fn wait_for_action_cancellation( + mut signal: watch::Receiver, + actions: Arc>, + action_id: String, +) { + loop { + if take_action_cancellation(&actions, &action_id) { + return; + } + if signal.changed().await.is_err() { + std::future::pending::<()>().await; + } + } } async fn perform_action( @@ -1086,6 +1204,38 @@ fn random_duration( minimum + Duration::from_millis(u64::try_from(offset).unwrap_or(u64::MAX)) } +fn valid_action_id(value: &str) -> bool { + !value.is_empty() + && value.len() <= MAX_ACTION_ID_BYTES + && value.chars().all(|character| { + character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '.' | ':') + }) +} + +fn lock_actions(value: &Mutex) -> MutexGuard<'_, ActionRegistry> { + value + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +fn take_action_cancellation(actions: &Mutex, action_id: &str) -> bool { + lock_actions(actions).cancelled.remove(action_id) +} + +fn unregister_action(actions: &Mutex, action_id: &str) { + let mut actions = lock_actions(actions); + actions.active.remove(action_id); + actions.cancelled.remove(action_id); +} + +fn unix_millis_now() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |duration| { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) + }) +} + #[derive(Clone)] pub struct BehaviorBackend { ingress: BehaviorIngress, @@ -1136,7 +1286,7 @@ impl AuthorizedToolBackend for BehaviorBackend { let result = match parsed { Ok(behavior_action) => tokio::select! { () = cancellation.cancelled() => Err(BehaviorError::Cancelled), - value = self.ingress.request(behavior_action, BehaviorTrigger::AuthorizedTool { authorization_id, tool: action.call().name.as_str().to_owned() }, BehaviorPolicyResult::Authorized(authorization_id)) => value, + value = self.ingress.request(call_id.as_str().to_owned(), behavior_action, BehaviorTrigger::AuthorizedTool { authorization_id, tool: action.call().name.as_str().to_owned() }, BehaviorPolicyResult::Authorized(authorization_id)) => value, }, Err(error) => Err(error), }; diff --git a/crates/metacrate-grid-agent/src/behavior_tests.rs b/crates/metacrate-grid-agent/src/behavior_tests.rs index adc784a..4c29508 100644 --- a/crates/metacrate-grid-agent/src/behavior_tests.rs +++ b/crates/metacrate-grid-agent/src/behavior_tests.rs @@ -453,6 +453,48 @@ async fn pause_preempts_stuck_walk_and_blocks_all_motion() { handle.shutdown().await.expect("shutdown"); } +#[tokio::test(start_paused = true)] +async fn exact_action_id_cancels_only_the_matching_pending_action() { + let region = uuid(10); + let sink = Arc::new(FakeSink::new(1, region)); + sink.set_stuck(true); + let mut config = settings(); + config.settle_delay = Duration::ZERO; + let handle = BehaviorController::new(config.clone(), sink, 16, 32, Duration::from_secs(1)) + .expect("controller") + .start(); + let ingress = handle.ingress(); + ingress.connected(1, region).expect("ready"); + tokio::task::yield_now().await; + let backend = BehaviorBackend::new(ingress.clone()); + let task = tokio::spawn({ + let backend = backend.clone(); + let config = config.clone(); + async move { + run_tool( + &backend, + &config, + WALK_SHORT_TOOL, + json!({"heading_degrees":90.0,"distance_meters":2.0}), + ) + .await + } + }); + tokio::time::advance(Duration::from_millis(300)).await; + assert!(!ingress.cancel_action("unrelated-action").expect("cancel")); + assert!(ingress.cancel_action("behavior-call").expect("cancel")); + tokio::task::yield_now().await; + assert!(matches!( + task.await.expect("task"), + ToolCallOutcome::Rejected { .. } + )); + assert!(matches!( + run_tool(&backend, &config, SIT_TOOL, json!({})).await, + ToolCallOutcome::Completed { .. } + )); + handle.shutdown().await.expect("shutdown"); +} + #[tokio::test(start_paused = true)] async fn offline_and_emergency_stop_emit_zero_motion() { let region = uuid(10); diff --git a/crates/metacrate-grid-agent/src/config.rs b/crates/metacrate-grid-agent/src/config.rs index 9d8d55b..383ed6a 100644 --- a/crates/metacrate-grid-agent/src/config.rs +++ b/crates/metacrate-grid-agent/src/config.rs @@ -7,12 +7,15 @@ use std::collections::{BTreeMap, BTreeSet}; use std::error::Error; use std::fmt; use std::fs; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::time::Duration; use url::Url; const MAX_CONFIG_BYTES: u64 = 64 * 1024; const MAX_SECRET_BYTES: u64 = 16 * 1024; +const MAX_TLS_DER_BYTES: u64 = 1024 * 1024; const MAX_AUTHORIZED_AVATARS: usize = 1_024; const MAX_QUEUE_CAPACITY: usize = 8_192; const MAX_BACKGROUND_TASKS: usize = 2; @@ -27,6 +30,11 @@ const ENV_GRID_AVATAR_NAME: &str = "METACRATE_AGENT_GRID_AVATAR_NAME"; const ENV_GRID_PASSWORD: &str = "METACRATE_AGENT_GRID_PASSWORD"; const ENV_GRID_PASSWORD_FILE: &str = "METACRATE_AGENT_GRID_PASSWORD_FILE"; const ENV_AUTHORIZED_AVATARS: &str = "METACRATE_AGENT_AUTHORIZED_AVATAR_UUIDS"; +const ENV_CONTROL_LISTEN: &str = "METACRATE_AGENT_CONTROL_LISTEN"; +const ENV_CONTROL_OPERATOR_TOKEN: &str = "METACRATE_AGENT_CONTROL_OPERATOR_TOKEN"; +const ENV_CONTROL_OPERATOR_TOKEN_FILE: &str = "METACRATE_AGENT_CONTROL_OPERATOR_TOKEN_FILE"; +const ENV_CONTROL_OBSERVER_TOKEN: &str = "METACRATE_AGENT_CONTROL_OBSERVER_TOKEN"; +const ENV_CONTROL_OBSERVER_TOKEN_FILE: &str = "METACRATE_AGENT_CONTROL_OBSERVER_TOKEN_FILE"; /// Wrapper that never reveals its contents through `Debug` or `Display` and /// deliberately does not implement serialization. @@ -247,6 +255,66 @@ pub struct ConversationSettings { pub limits: crate::conversation::ConversationLimits, } +/// Portable control-plane settings. Plain TCP is loopback-only; a remote +/// listener is admitted only with an explicit Rustls certificate and key. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ControlSettings { + pub listen: SocketAddr, + pub operator_token: Option, + pub observer_token: Option, + pub limits: crate::control_plane::ControlLimits, + pub remote_tls: Option, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct RemoteTlsSettings { + pub certificate_der: PathBuf, + pub private_key_der: PathBuf, +} + +impl RemoteTlsSettings { + /// Loads bounded, regular DER files into a portable Rustls server config. + /// + /// # Errors + /// + /// Returns a configuration error when either file is unavailable, unsafe, + /// oversized, malformed, or does not form a valid certificate/key pair. + pub fn server_config(&self) -> Result, ConfigError> { + let certificate = read_bounded_regular_file( + "control.remote_tls_certificate_der", + &self.certificate_der, + MAX_TLS_DER_BYTES, + )?; + let private_key = read_bounded_regular_file( + "control.remote_tls_private_key_der", + &self.private_key_der, + MAX_TLS_DER_BYTES, + )?; + let private_key = rustls::pki_types::PrivateKeyDer::try_from(private_key) + .map_err(|_| ConfigError::InvalidControl)?; + let config = rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert( + vec![rustls::pki_types::CertificateDer::from(certificate)], + private_key, + ) + .map_err(|_| ConfigError::InvalidControl)?; + Ok(Arc::new(config)) + } +} + +impl Default for ControlSettings { + fn default() -> Self { + Self { + listen: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 7943), + operator_token: None, + observer_token: None, + limits: crate::control_plane::ControlLimits::default(), + remote_tls: None, + } + } +} + /// Fully resolved configuration. It cannot be constructed without validation. #[derive(Clone, Debug, Eq, PartialEq)] pub struct AgentConfig { @@ -261,6 +329,7 @@ pub struct AgentConfig { pub reconnect: crate::session::ReconnectPolicy, pub conversation: ConversationSettings, pub interaction: crate::interaction::InteractionSettings, + pub control: ControlSettings, } impl AgentConfig { @@ -300,6 +369,45 @@ impl AgentConfig { if !self.behavior.is_valid() { return Err(ConfigError::InvalidBehavior); } + if (!self.control.listen.ip().is_loopback() && self.control.remote_tls.is_none()) + || !self.control.limits.is_valid() + { + return Err(ConfigError::InvalidControl); + } + if self.mode == OperatingMode::SplitService && self.control.operator_token.is_none() { + return Err(ConfigError::Missing { + field: "control.operator_token", + required_for: "split service mode", + }); + } + if let Some(operator) = &self.control.operator_token + && (self + .control + .observer_token + .as_ref() + .is_some_and(|observer| observer == operator) + || operator == &self.llm.api_key + || self + .grid + .as_ref() + .is_some_and(|grid| operator == &grid.password)) + { + return Err(ConfigError::InvalidControl); + } + if self + .control + .observer_token + .as_ref() + .is_some_and(|observer| { + observer == &self.llm.api_key + || self + .grid + .as_ref() + .is_some_and(|grid| observer == &grid.password) + }) + { + return Err(ConfigError::InvalidControl); + } if self.mode != OperatingMode::OfflineFake && self.grid.is_none() { return Err(ConfigError::Missing { field: "grid", @@ -476,6 +584,7 @@ pub enum ConfigError { InvalidConversationMemory, InvalidInteraction, InvalidBehavior, + InvalidControl, } impl fmt::Display for ConfigError { @@ -531,6 +640,9 @@ impl fmt::Display for ConfigError { } Self::InvalidInteraction => formatter.write_str("invalid interaction bounds"), Self::InvalidBehavior => formatter.write_str("invalid embodied behavior bounds"), + Self::InvalidControl => formatter.write_str( + "invalid control plane: use distinct non-credential tokens and loopback TCP or an explicit TLS remote listener", + ), } } } @@ -553,6 +665,7 @@ struct FileConfig { reconnect: RawReconnect, conversation: RawConversation, interaction: RawInteraction, + control: RawControl, } #[derive(Clone, Default, Deserialize)] @@ -575,6 +688,18 @@ struct RawGrid { struct RawSecretFiles { llm_api_key: Option, grid_password: Option, + control_operator_token: Option, + control_observer_token: Option, +} + +#[derive(Clone, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +struct RawControl { + listen: Option, + operator_token: Option, + observer_token: Option, + remote_tls_certificate_der: Option, + remote_tls_private_key_der: Option, } #[derive(Clone, Default, Deserialize)] @@ -932,6 +1057,57 @@ fn resolve( .max_debounce_fragments .unwrap_or(interaction_defaults.max_debounce_fragments), }; + let control_defaults = ControlSettings::default(); + let listen_text = environment + .get(ENV_CONTROL_LISTEN) + .or(raw.control.listen) + .unwrap_or_else(|| control_defaults.listen.to_string()); + let listen = listen_text + .parse::() + .map_err(|_| ConfigError::InvalidControl)?; + let operator_token = optional_secret_from_layers( + environment.get(ENV_CONTROL_OPERATOR_TOKEN), + environment + .get(ENV_CONTROL_OPERATOR_TOKEN_FILE) + .map(PathBuf::from), + raw.control.operator_token, + raw.secret_files.control_operator_token, + base, + "control.operator_token", + )?; + let observer_token = optional_secret_from_layers( + environment.get(ENV_CONTROL_OBSERVER_TOKEN), + environment + .get(ENV_CONTROL_OBSERVER_TOKEN_FILE) + .map(PathBuf::from), + raw.control.observer_token, + raw.secret_files.control_observer_token, + base, + "control.observer_token", + )?; + let remote_tls = match ( + raw.control.remote_tls_certificate_der, + raw.control.remote_tls_private_key_der, + ) { + (None, None) => None, + (Some(certificate_der), Some(private_key_der)) => Some(RemoteTlsSettings { + certificate_der: resolve_path(base, &certificate_der), + private_key_der: resolve_path(base, &private_key_der), + }), + _ => return Err(ConfigError::InvalidControl), + }; + let control = ControlSettings { + listen, + operator_token, + observer_token, + limits: crate::control_plane::ControlLimits { + command_queue: limits.control_queue, + request_timeout: timeouts.request.min(Duration::from_mins(2)), + write_timeout: timeouts.shutdown.min(Duration::from_secs(30)), + ..crate::control_plane::ControlLimits::default() + }, + remote_tls, + }; let config = AgentConfig { mode, @@ -996,6 +1172,7 @@ fn resolve( reconnect, conversation, interaction, + control, }; config.validate()?; Ok(config) @@ -1064,6 +1241,28 @@ fn secret_from_layers( }) } +fn optional_secret_from_layers( + environment_value: Option, + environment_file: Option, + file_value: Option, + file_path: Option, + base: &Path, + field: &'static str, +) -> Result, ConfigError> { + if let Some(value) = environment_value { + return SecretString::new(field, value).map(Some); + } + if let Some(path) = environment_file { + return read_secret(field, &resolve_path(base, &path)).map(Some); + } + if let Some(path) = file_path { + return read_secret(field, &resolve_path(base, &path)).map(Some); + } + file_value + .map(|value| SecretString::new(field, value)) + .transpose() +} + fn resolve_path(base: &Path, path: &Path) -> PathBuf { if path.is_absolute() { path.to_owned() @@ -1437,6 +1636,50 @@ mod tests { )); } + #[test] + fn split_control_requires_distinct_tokens_and_tls_for_remote_bind() { + let split = MapEnvironment::from_pairs([ + (ENV_SPLIT, "true"), + (ENV_LLM_ENDPOINT, "https://llm.invalid/chat"), + (ENV_LLM_API_KEY, "llm-key"), + (ENV_GRID_LOGIN_URL, "https://grid.invalid/login"), + (ENV_GRID_AVATAR_NAME, "Control Agent"), + (ENV_GRID_PASSWORD, "grid-password"), + (ENV_CONTROL_OPERATOR_TOKEN, "operator-capability"), + (ENV_CONTROL_OBSERVER_TOKEN, "observer-capability"), + ]); + let config = ConfigLoader::new() + .with_environment(split.clone()) + .load() + .expect("bounded loopback split control"); + assert_eq!(config.mode, OperatingMode::SplitService); + let diagnostic = format!("{config:?}"); + for secret in [ + "operator-capability", + "observer-capability", + "llm-key", + "grid-password", + ] { + assert!(!diagnostic.contains(secret)); + } + + let mut reused = split.clone(); + reused.insert(ENV_CONTROL_OPERATOR_TOKEN, "llm-key"); + assert_eq!( + ConfigLoader::new().with_environment(reused).load(), + Err(ConfigError::InvalidControl) + ); + + let mut remote_plaintext = split; + remote_plaintext.insert(ENV_CONTROL_LISTEN, "0.0.0.0:7943"); + assert_eq!( + ConfigLoader::new() + .with_environment(remote_plaintext) + .load(), + Err(ConfigError::InvalidControl) + ); + } + #[test] fn secret_file_is_bounded_regular_and_trailing_newline_is_removed() { let secret = temporary_file("api-key", "secret-from-file\n"); diff --git a/crates/metacrate-grid-agent/src/control_plane.rs b/crates/metacrate-grid-agent/src/control_plane.rs new file mode 100644 index 0000000..e6fbff5 --- /dev/null +++ b/crates/metacrate-grid-agent/src/control_plane.rs @@ -0,0 +1,1780 @@ +//! Versioned, authenticated, transport-neutral operator control plane. + +#![allow(clippy::missing_errors_doc)] + +use crate::backend::BackendFuture; +use crate::config::SecretString; +use libremetaverse_types::compat::{CancellationToken, CancellationTokenSource}; +use serde::{Deserialize, Serialize}; +use std::collections::{BTreeMap, BTreeSet, VecDeque}; +use std::error::Error; +use std::fmt; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard, Weak}; +use std::time::{Duration, Instant}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::{mpsc, oneshot}; +use tokio::task::{JoinHandle, JoinSet}; + +pub const CONTROL_PROTOCOL_VERSION: u16 = 1; +const MAX_REQUEST_ID_BYTES: usize = 96; +const MAX_TOKEN_BYTES: usize = 16 * 1024; +const MAX_OPERATOR_MESSAGE_BYTES: usize = 16 * 1024; +const MAX_PAGE_SIZE: u16 = 100; +const MAX_EVENT_BYTES: usize = 4 * 1024; + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ControlRole { + Observer, + Operator, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ControlErrorCode { + AuthenticationFailed, + VersionMismatch, + PermissionDenied, + InvalidRequest, + Replay, + NotFound, + Conflict, + Cancelled, + TimedOut, + Busy, + Backpressure, + FrameTooLarge, + IdleTimeout, + TransportClosed, + Internal, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct ControlErrorBody { + pub code: ControlErrorCode, + pub message: String, + pub retryable: bool, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ControlError { + pub code: ControlErrorCode, + pub message: &'static str, + pub retryable: bool, +} + +impl ControlError { + const fn new(code: ControlErrorCode, message: &'static str, retryable: bool) -> Self { + Self { + code, + message, + retryable, + } + } + + fn body(&self) -> ControlErrorBody { + ControlErrorBody { + code: self.code, + message: self.message.to_owned(), + retryable: self.retryable, + } + } +} + +impl fmt::Display for ControlError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(self.message) + } +} + +impl Error for ControlError {} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ControlLimits { + pub max_frame_bytes: usize, + pub max_connections: usize, + pub max_in_flight_per_connection: usize, + pub command_queue: usize, + pub event_queue: usize, + pub event_history: usize, + pub max_events_per_second: usize, + pub max_subscriptions: usize, + pub replay_history: usize, + pub request_timeout: Duration, + pub idle_timeout: Duration, + pub write_timeout: Duration, +} + +impl Default for ControlLimits { + fn default() -> Self { + Self { + max_frame_bytes: 64 * 1024, + max_connections: 16, + max_in_flight_per_connection: 16, + command_queue: 64, + event_queue: 128, + event_history: 512, + max_events_per_second: 1_024, + max_subscriptions: 32, + replay_history: 1_024, + request_timeout: Duration::from_secs(15), + idle_timeout: Duration::from_mins(2), + write_timeout: Duration::from_secs(5), + } + } +} + +impl ControlLimits { + #[must_use] + pub fn is_valid(self) -> bool { + (1_024..=1024 * 1024).contains(&self.max_frame_bytes) + && (1..=128).contains(&self.max_connections) + && (1..=128).contains(&self.max_in_flight_per_connection) + && (1..=8_192).contains(&self.command_queue) + && (1..=8_192).contains(&self.event_queue) + && (1..=8_192).contains(&self.event_history) + && (1..=65_536).contains(&self.max_events_per_second) + && (1..=128).contains(&self.max_subscriptions) + && (1..=16_384).contains(&self.replay_history) + && !self.request_timeout.is_zero() + && self.request_timeout <= Duration::from_mins(2) + && self.idle_timeout >= Duration::from_secs(5) + && self.idle_timeout <= Duration::from_hours(1) + && !self.write_timeout.is_zero() + && self.write_timeout <= Duration::from_secs(30) + } +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PageRequest { + pub cursor: Option, + pub limit: u16, +} + +impl Default for PageRequest { + fn default() -> Self { + Self { + cursor: None, + limit: 50, + } + } +} + +impl PageRequest { + fn valid(&self) -> bool { + (1..=MAX_PAGE_SIZE).contains(&self.limit) + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct Page { + pub items: Vec, + pub next_cursor: Option, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct HealthView { + pub service_state: String, + pub ready: bool, + pub uptime_seconds: u64, + pub protocol_version: u16, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct RuntimeView { + pub grid_state: String, + pub generation: u64, + pub transport_connected: bool, + pub agent_ready: bool, + pub region_id: Option, + pub region_name: Option, + pub position: Option<[f64; 3]>, + pub behavior_mode: String, + pub control_queue_used: usize, + pub control_queue_capacity: usize, + pub budget_tool_calls_used: u64, + pub budget_movement_millimeters_used: u64, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct SessionMetadataView { + pub session_id: String, + pub avatar_id: String, + pub channel: String, + pub created_unix_millis: u64, + pub last_active_unix_millis: u64, + pub turns: usize, + pub bytes: usize, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct ScheduledJobView { + pub job_id: String, + pub kind: String, + pub enabled: bool, + pub next_run_unix_millis: Option, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct PendingApprovalView { + pub approval_id: u64, + pub tool: String, + pub principal: String, + pub expires_unix_seconds: u64, + pub movement_millimeters: u64, + pub inventory_operations: u64, + pub build_prims: u64, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct AuditEventView { + pub sequence: u64, + pub unix_millis: u64, + pub principal: String, + pub operation: String, + pub outcome: String, + pub authorization_id: Option, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", content = "data", rename_all = "snake_case")] +pub enum ControlPayload { + Health(HealthView), + Runtime(RuntimeView), + Sessions(Page), + ScheduledJobs(Page), + PendingApprovals(Page), + AuditEvents(Page), + Accepted { operation_id: String }, + Completed, + Subscribed { current_sequence: u64 }, + Cancelled { request_id: String }, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ConversationChannelView { + PublicChat, + DirectIm, +} + +#[derive(Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "method", content = "parameters", rename_all = "snake_case")] +pub enum ControlRequest { + Health, + Runtime, + ListSessions { + page: PageRequest, + }, + ListScheduledJobs { + page: PageRequest, + }, + ListPendingApprovals { + page: PageRequest, + }, + ListAuditEvents { + page: PageRequest, + }, + SubscribeEvents { + after_sequence: Option, + }, + CancelRequest { + target_request_id: String, + }, + PauseAutonomy, + ResumeAutonomy, + CancelAction { + action_id: String, + }, + DecideApproval { + approval_id: u64, + approve: bool, + }, + ForceReconnect, + ExpireConversation { + avatar_id: String, + channel: ConversationChannelView, + }, + SetRoamingJob { + job_id: String, + enabled: bool, + }, + InjectOperatorMessage { + message: String, + }, + GracefulShutdown, +} + +impl fmt::Debug for ControlRequest { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + if let Self::InjectOperatorMessage { message } = self { + return formatter + .debug_struct("InjectOperatorMessage") + .field("message_bytes", &message.len()) + .finish(); + } + formatter.write_str(request_name(self)) + } +} + +impl ControlRequest { + fn mutating(&self) -> bool { + !matches!( + self, + Self::Health + | Self::Runtime + | Self::ListSessions { .. } + | Self::ListScheduledJobs { .. } + | Self::ListPendingApprovals { .. } + | Self::ListAuditEvents { .. } + | Self::SubscribeEvents { .. } + ) + } + + fn valid(&self) -> bool { + match self { + Self::ListSessions { page } + | Self::ListScheduledJobs { page } + | Self::ListPendingApprovals { page } + | Self::ListAuditEvents { page } => page.valid(), + Self::CancelRequest { target_request_id } => valid_identifier(target_request_id), + Self::CancelAction { action_id } => valid_identifier(action_id), + Self::DecideApproval { approval_id, .. } => *approval_id != 0, + Self::ExpireConversation { avatar_id, .. } => valid_uuid_text(avatar_id), + Self::SetRoamingJob { job_id, .. } => valid_identifier(job_id), + Self::InjectOperatorMessage { message } => { + !message.trim().is_empty() + && message.len() <= MAX_OPERATOR_MESSAGE_BYTES + && !message.contains('\0') + } + _ => true, + } + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ControlRequestEnvelope { + pub version: u16, + pub request_id: String, + pub request: ControlRequest, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ControlResponseEnvelope { + pub version: u16, + pub request_id: String, + pub result: Result, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", content = "data", rename_all = "snake_case")] +pub enum ControlEventKind { + Mutation { + principal: String, + operation: String, + outcome: String, + }, + StateChanged { + component: String, + state: String, + }, + Gap { + first_available: u64, + last_missed: u64, + }, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct ControlEvent { + pub version: u16, + pub sequence: u64, + pub unix_millis: u64, + pub event: ControlEventKind, +} + +pub type ControlFuture<'a> = BackendFuture<'a, Result>; + +/// Authenticated request identity supplied to the management implementation. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ControlContext { + pub role: ControlRole, + pub principal: String, +} + +/// Management implementation injected beneath both transports. Protocol types +/// deliberately do not expose internal runtime structs or secret-bearing data. +pub trait ControlTarget: Send + Sync + 'static { + fn execute( + &self, + context: ControlContext, + request: ControlRequest, + cancellation: CancellationToken, + ) -> ControlFuture<'_>; +} + +struct TokenSet { + observer: Option, + operator: SecretString, +} + +struct EventHubState { + next_sequence: u64, + history: VecDeque, + subscribers: BTreeMap>, + next_subscriber: u64, + rate_window: Instant, + events_in_window: usize, +} + +struct EventHub { + state: Mutex, + history_capacity: usize, + subscriber_capacity: usize, + max_events_per_second: usize, + max_subscriptions: usize, +} + +impl EventHub { + fn new( + history_capacity: usize, + subscriber_capacity: usize, + max_events_per_second: usize, + max_subscriptions: usize, + ) -> Self { + Self { + state: Mutex::new(EventHubState { + next_sequence: 1, + history: VecDeque::with_capacity(history_capacity), + subscribers: BTreeMap::new(), + next_subscriber: 1, + rate_window: Instant::now(), + events_in_window: 0, + }), + history_capacity, + subscriber_capacity, + max_events_per_second, + max_subscriptions, + } + } + + fn publish(&self, event: ControlEventKind) -> bool { + if serde_json::to_vec(&event).map_or(true, |body| body.len() > MAX_EVENT_BYTES) { + return false; + } + let mut state = lock(&self.state); + if state.rate_window.elapsed() >= Duration::from_secs(1) { + state.rate_window = Instant::now(); + state.events_in_window = 0; + } + if state.events_in_window >= self.max_events_per_second { + return false; + } + state.events_in_window = state.events_in_window.saturating_add(1); + let sequence = state.next_sequence; + state.next_sequence = state.next_sequence.saturating_add(1); + let event = ControlEvent { + version: CONTROL_PROTOCOL_VERSION, + sequence, + unix_millis: unix_millis(), + event, + }; + if state.history.len() == self.history_capacity { + state.history.pop_front(); + } + state.history.push_back(event.clone()); + state + .subscribers + .retain(|_, subscriber| subscriber.try_send(event.clone()).is_ok()); + true + } + + fn subscribe( + self: &Arc, + after: Option, + ) -> Result<(u64, ControlSubscription), ControlError> { + let mut state = lock(&self.state); + if state.subscribers.len() >= self.max_subscriptions { + return Err(ControlError::new( + ControlErrorCode::Busy, + "event subscription limit reached", + true, + )); + } + let current = state.next_sequence.saturating_sub(1); + let after = after.unwrap_or(current); + let retained_first = state + .history + .front() + .map_or(state.next_sequence, |item| item.sequence); + let replay_first = state + .next_sequence + .saturating_sub(u64::try_from(self.subscriber_capacity).unwrap_or(u64::MAX)); + let first = retained_first.max(replay_first); + let mut initial = VecDeque::new(); + if after.saturating_add(1) < first { + initial.push_back(ControlEvent { + version: CONTROL_PROTOCOL_VERSION, + sequence: first.saturating_sub(1), + unix_millis: unix_millis(), + event: ControlEventKind::Gap { + first_available: first, + last_missed: first.saturating_sub(1), + }, + }); + } + initial.extend( + state + .history + .iter() + .filter(|item| item.sequence >= first && item.sequence > after) + .cloned(), + ); + let (sender, receiver) = mpsc::channel(self.subscriber_capacity); + let id = state.next_subscriber; + state.next_subscriber = state.next_subscriber.saturating_add(1); + state.subscribers.insert(id, sender); + Ok(( + current, + ControlSubscription { + id, + hub: Arc::downgrade(self), + initial, + receiver, + }, + )) + } +} + +pub struct ControlSubscription { + id: u64, + hub: Weak, + initial: VecDeque, + receiver: mpsc::Receiver, +} + +impl Drop for ControlSubscription { + fn drop(&mut self) { + if let Some(hub) = self.hub.upgrade() { + lock(&hub.state).subscribers.remove(&self.id); + } + } +} + +impl ControlSubscription { + pub async fn recv(&mut self) -> Option { + if let Some(event) = self.initial.pop_front() { + Some(event) + } else { + self.receiver.recv().await + } + } +} + +pub struct ControlPlane { + target: Arc, + tokens: TokenSet, + limits: ControlLimits, + events: Arc, + next_connection: AtomicU64, + active_connections: AtomicUsize, +} + +impl fmt::Debug for ControlPlane { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("ControlPlane") + .field("limits", &self.limits) + .field("tokens", &"[REDACTED]") + .field( + "active_connections", + &self.active_connections.load(Ordering::Acquire), + ) + .finish_non_exhaustive() + } +} + +impl ControlPlane { + /// Creates an authenticated in-process operator connection with a fresh + /// capability that is never exposed through configuration or diagnostics. + pub fn integrated( + target: Arc, + limits: ControlLimits, + ) -> Result<(Arc, InProcessControlClient), ControlError> { + let random = libremetaverse_types::UUID::secure_random().map_err(|_| { + ControlError::new( + ControlErrorCode::Internal, + "cannot generate integrated control capability", + false, + ) + })?; + let value = format!("integrated-{random}"); + let token = SecretString::new("control.integrated_token", value.clone()).map_err(|_| { + ControlError::new( + ControlErrorCode::Internal, + "cannot construct integrated control capability", + false, + ) + })?; + let plane = Self::new(target, token, None, limits)?; + let client = plane.connect(&value)?; + Ok((plane, client)) + } + + pub fn new( + target: Arc, + operator_token: SecretString, + observer_token: Option, + limits: ControlLimits, + ) -> Result, ControlError> { + if !limits.is_valid() + || observer_token.as_ref().is_some_and(|token| { + constant_time_eq(token.expose_secret(), operator_token.expose_secret()) + }) + { + return Err(ControlError::new( + ControlErrorCode::InvalidRequest, + "unsafe control-plane configuration", + false, + )); + } + Ok(Arc::new(Self { + target, + tokens: TokenSet { + observer: observer_token, + operator: operator_token, + }, + limits, + events: Arc::new(EventHub::new( + limits.event_history, + limits.event_queue, + limits.max_events_per_second, + limits.max_subscriptions, + )), + next_connection: AtomicU64::new(1), + active_connections: AtomicUsize::new(0), + })) + } + + pub fn connect(self: &Arc, token: &str) -> Result { + let core = self.open(token)?; + Ok(InProcessControlClient { core }) + } + + fn open(self: &Arc, token: &str) -> Result, ControlError> { + let role = self.authenticate(token)?; + let prior = self.active_connections.fetch_add(1, Ordering::AcqRel); + if prior >= self.limits.max_connections { + self.active_connections.fetch_sub(1, Ordering::AcqRel); + return Err(ControlError::new( + ControlErrorCode::Busy, + "connection limit reached", + true, + )); + } + let id = self.next_connection.fetch_add(1, Ordering::Relaxed); + Ok(Arc::new(ConnectionCore { + id, + role, + plane: Arc::clone(self), + replay: Mutex::new((VecDeque::new(), BTreeSet::new())), + active: Mutex::new(BTreeMap::new()), + in_flight: AtomicUsize::new(0), + })) + } + + fn authenticate(&self, token: &str) -> Result { + if token.is_empty() || token.len() > MAX_TOKEN_BYTES || token.contains(['\0', '\r', '\n']) { + return Err(auth_error()); + } + if constant_time_eq(token, self.tokens.operator.expose_secret()) { + return Ok(ControlRole::Operator); + } + if self + .tokens + .observer + .as_ref() + .is_some_and(|expected| constant_time_eq(token, expected.expose_secret())) + { + return Ok(ControlRole::Observer); + } + Err(auth_error()) + } + + #[must_use] + pub fn active_connections(&self) -> usize { + self.active_connections.load(Ordering::Acquire) + } + + #[must_use] + pub fn retained_event_count(&self) -> usize { + lock(&self.events.state).history.len() + } + + pub fn publish(&self, event: ControlEventKind) -> bool { + self.events.publish(event) + } +} + +struct ConnectionCore { + id: u64, + role: ControlRole, + plane: Arc, + replay: Mutex<(VecDeque, BTreeSet)>, + active: Mutex>, + in_flight: AtomicUsize, +} + +impl Drop for ConnectionCore { + fn drop(&mut self) { + for source in lock(&self.active).values() { + source.cancel(); + } + self.plane.active_connections.fetch_sub(1, Ordering::AcqRel); + } +} + +impl ConnectionCore { + async fn request( + self: &Arc, + envelope: ControlRequestEnvelope, + ) -> (ControlResponseEnvelope, Option) { + let request_id = envelope.request_id.clone(); + let mutation = envelope + .request + .mutating() + .then(|| request_name(&envelope.request).to_owned()); + let result = self.execute_envelope(envelope).await; + if let Some(operation) = mutation { + self.plane.events.publish(ControlEventKind::Mutation { + principal: format!("control:{}:{}", role_name(self.role), self.id), + operation, + outcome: match &result { + Ok((ControlPayload::Accepted { .. }, _)) => "accepted", + Ok(_) => "completed", + Err(_) => "rejected", + } + .to_owned(), + }); + } + match result { + Ok((payload, subscription)) => (response_ok(request_id, payload), subscription), + Err(error) => (response_error(request_id, &error), None), + } + } + + async fn execute_envelope( + self: &Arc, + envelope: ControlRequestEnvelope, + ) -> Result<(ControlPayload, Option), ControlError> { + if envelope.version != CONTROL_PROTOCOL_VERSION { + return Err(ControlError::new( + ControlErrorCode::VersionMismatch, + "unsupported control protocol version", + false, + )); + } + if !valid_identifier(&envelope.request_id) || !envelope.request.valid() { + return Err(ControlError::new( + ControlErrorCode::InvalidRequest, + "invalid bounded control request", + false, + )); + } + self.remember(&envelope.request_id)?; + if envelope.request.mutating() && self.role != ControlRole::Operator { + return Err(ControlError::new( + ControlErrorCode::PermissionDenied, + "operator permission required", + false, + )); + } + if let ControlRequest::CancelRequest { target_request_id } = &envelope.request { + let cancelled = lock(&self.active) + .get(target_request_id) + .is_some_and(|source| { + source.cancel(); + true + }); + if !cancelled { + return Err(ControlError::new( + ControlErrorCode::NotFound, + "active request not found", + false, + )); + } + return Ok(( + ControlPayload::Cancelled { + request_id: target_request_id.clone(), + }, + None, + )); + } + if let ControlRequest::SubscribeEvents { after_sequence } = envelope.request { + let (current, subscription) = self.plane.events.subscribe(after_sequence)?; + return Ok(( + ControlPayload::Subscribed { + current_sequence: current, + }, + Some(subscription), + )); + } + let prior = self.in_flight.fetch_add(1, Ordering::AcqRel); + if prior >= self.plane.limits.max_in_flight_per_connection { + self.in_flight.fetch_sub(1, Ordering::AcqRel); + return Err(ControlError::new( + ControlErrorCode::Busy, + "request concurrency limit reached", + true, + )); + } + let source = CancellationTokenSource::new(); + lock(&self.active).insert(envelope.request_id.clone(), source.clone()); + let context = ControlContext { + role: self.role, + principal: format!("control:{}:{}", role_name(self.role), self.id), + }; + let timed = tokio::time::timeout( + self.plane.limits.request_timeout, + self.plane + .target + .execute(context, envelope.request, source.token()), + ) + .await; + if timed.is_err() { + source.cancel(); + } + lock(&self.active).remove(&envelope.request_id); + self.in_flight.fetch_sub(1, Ordering::AcqRel); + let result = timed.map_err(|_| { + ControlError::new( + ControlErrorCode::TimedOut, + "control request timed out", + true, + ) + })?; + let payload = result?; + if !payload_valid(&payload, self.plane.limits.max_frame_bytes) { + return Err(ControlError::new( + ControlErrorCode::FrameTooLarge, + "control response exceeds configured bound", + false, + )); + } + Ok((payload, None)) + } + + fn remember(&self, request_id: &str) -> Result<(), ControlError> { + let mut replay = lock(&self.replay); + if replay.1.contains(request_id) { + return Err(ControlError::new( + ControlErrorCode::Replay, + "request ID replayed", + false, + )); + } + if replay.0.len() == self.plane.limits.replay_history + && let Some(expired) = replay.0.pop_front() + { + replay.1.remove(&expired); + } + replay.0.push_back(request_id.to_owned()); + replay.1.insert(request_id.to_owned()); + Ok(()) + } +} + +#[derive(Clone)] +pub struct InProcessControlClient { + core: Arc, +} + +impl InProcessControlClient { + pub async fn request(&self, envelope: ControlRequestEnvelope) -> ControlResponseEnvelope { + self.core.request(envelope).await.0 + } + + pub async fn subscribe( + &self, + envelope: ControlRequestEnvelope, + ) -> Result<(ControlResponseEnvelope, ControlSubscription), ControlError> { + let (response, subscription) = self.core.request(envelope).await; + if let Some(subscription) = subscription { + return Ok((response, subscription)); + } + if let Err(error) = &response.result { + return Err(ControlError { + code: error.code, + message: "event subscription rejected", + retryable: error.retryable, + }); + } + Err(ControlError::new( + ControlErrorCode::InvalidRequest, + "request did not create a subscription", + false, + )) + } + + #[must_use] + pub fn role(&self) -> ControlRole { + self.core.role + } +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct ClientHello { + version: u16, + token: String, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +struct ServerHello { + version: u16, + role: Option, + error: Option, +} + +#[derive(Serialize, Deserialize)] +#[serde(tag = "frame", content = "body", rename_all = "snake_case")] +enum ClientFrame { + Hello(ClientHello), + Request(ControlRequestEnvelope), +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(tag = "frame", content = "body", rename_all = "snake_case")] +enum ServerFrame { + Hello(ServerHello), + Response(ControlResponseEnvelope), + Event(ControlEvent), +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct TcpControlConfig { + pub listen: SocketAddr, + pub limits: ControlLimits, +} + +impl Default for TcpControlConfig { + fn default() -> Self { + Self { + listen: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 7943), + limits: ControlLimits::default(), + } + } +} + +pub struct TcpControlServer { + address: SocketAddr, + cancellation: CancellationTokenSource, + task: Option>, + shutdown_timeout: Duration, +} + +impl Drop for TcpControlServer { + fn drop(&mut self) { + self.cancellation.cancel(); + if let Some(task) = &self.task { + task.abort(); + } + } +} + +impl TcpControlServer { + pub async fn bind( + plane: Arc, + config: TcpControlConfig, + ) -> Result { + if !config.listen.ip().is_loopback() + || !config.limits.is_valid() + || config.limits != plane.limits + { + return Err(ControlError::new( + ControlErrorCode::InvalidRequest, + "plain TCP control transport is loopback-only", + false, + )); + } + Self::bind_listener(plane, config, None).await + } + + /// Binds an explicitly configured TLS listener. Unlike [`Self::bind`], + /// this may use a non-loopback address because plaintext never leaves the + /// TLS session and token authentication still applies inside it. + pub async fn bind_tls( + plane: Arc, + config: TcpControlConfig, + server_config: Arc, + ) -> Result { + if !config.limits.is_valid() || config.limits != plane.limits { + return Err(ControlError::new( + ControlErrorCode::InvalidRequest, + "invalid TLS control transport configuration", + false, + )); + } + Self::bind_listener( + plane, + config, + Some(tokio_rustls::TlsAcceptor::from(server_config)), + ) + .await + } + + async fn bind_listener( + plane: Arc, + config: TcpControlConfig, + tls: Option, + ) -> Result { + let listener = TcpListener::bind(config.listen).await.map_err(|_| { + ControlError::new( + ControlErrorCode::TransportClosed, + "cannot bind control transport", + true, + ) + })?; + let address = listener.local_addr().map_err(|_| { + ControlError::new( + ControlErrorCode::TransportClosed, + "cannot read control transport address", + true, + ) + })?; + let cancellation = CancellationTokenSource::new(); + let task_cancellation = cancellation.token(); + let shutdown_timeout = config.limits.write_timeout; + let task = tokio::spawn(async move { + let mut connections = JoinSet::new(); + loop { + let accepted = tokio::select! { + () = task_cancellation.cancelled() => break, + Some(_) = connections.join_next(), if !connections.is_empty() => continue, + accepted = listener.accept() => accepted, + }; + let Ok((stream, peer)) = accepted else { + continue; + }; + if (tls.is_none() && !peer.ip().is_loopback()) + || connections.len() >= plane.limits.max_connections + { + continue; + } + let connection_plane = Arc::clone(&plane); + let connection_cancel = task_cancellation.clone(); + let connection_tls = tls.clone(); + connections.spawn(async move { + if let Some(acceptor) = connection_tls { + if let Ok(Ok(stream)) = tokio::time::timeout( + connection_plane.limits.idle_timeout, + acceptor.accept(stream), + ) + .await + { + let _ = serve_control_connection( + connection_plane, + stream, + connection_cancel, + ) + .await; + } + } else { + let _ = + serve_control_connection(connection_plane, stream, connection_cancel) + .await; + } + }); + } + connections.abort_all(); + while connections.join_next().await.is_some() {} + }); + Ok(Self { + address, + cancellation, + task: Some(task), + shutdown_timeout, + }) + } + + #[must_use] + pub const fn local_addr(&self) -> SocketAddr { + self.address + } + + pub async fn shutdown(mut self) -> Result<(), ControlError> { + self.cancellation.cancel(); + if let Some(task) = self.task.take() { + tokio::time::timeout(self.shutdown_timeout, task) + .await + .map_err(|_| { + ControlError::new( + ControlErrorCode::TimedOut, + "control server shutdown timed out", + true, + ) + })? + .map_err(|_| { + ControlError::new( + ControlErrorCode::Internal, + "control server task failed", + false, + ) + })?; + } + Ok(()) + } +} + +#[allow(clippy::too_many_lines)] // Ordered handshake, reader, writer, and task ownership. +async fn serve_control_connection( + plane: Arc, + mut stream: S, + cancellation: CancellationToken, +) -> Result<(), ControlError> +where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + let hello: ClientFrame = read_frame( + &mut stream, + plane.limits.max_frame_bytes, + plane.limits.idle_timeout, + ) + .await?; + let ClientFrame::Hello(hello) = hello else { + return Err(ControlError::new( + ControlErrorCode::AuthenticationFailed, + "authentication required", + false, + )); + }; + if hello.version != CONTROL_PROTOCOL_VERSION { + let error = ControlError::new( + ControlErrorCode::VersionMismatch, + "unsupported control protocol version", + false, + ); + let _ = write_frame( + &mut stream, + &ServerFrame::Hello(ServerHello { + version: CONTROL_PROTOCOL_VERSION, + role: None, + error: Some(error.body()), + }), + plane.limits.max_frame_bytes, + plane.limits.write_timeout, + ) + .await; + return Err(error); + } + let core = match plane.open(&hello.token) { + Ok(core) => core, + Err(error) => { + let _ = write_frame( + &mut stream, + &ServerFrame::Hello(ServerHello { + version: CONTROL_PROTOCOL_VERSION, + role: None, + error: Some(error.body()), + }), + plane.limits.max_frame_bytes, + plane.limits.write_timeout, + ) + .await; + return Err(error); + } + }; + write_frame( + &mut stream, + &ServerFrame::Hello(ServerHello { + version: CONTROL_PROTOCOL_VERSION, + role: Some(core.role), + error: None, + }), + plane.limits.max_frame_bytes, + plane.limits.write_timeout, + ) + .await?; + let (mut reader, mut writer) = tokio::io::split(stream); + let (outbound, mut outbound_rx) = mpsc::channel::(plane.limits.command_queue); + let connection_cancel = CancellationTokenSource::new(); + let writer_cancel = connection_cancel.token(); + let writer_source = connection_cancel.clone(); + let write_timeout = plane.limits.write_timeout; + let max_frame = plane.limits.max_frame_bytes; + let writer_task = tokio::spawn(async move { + loop { + let frame = tokio::select! { + () = writer_cancel.cancelled() => break, + frame = outbound_rx.recv() => frame, + }; + let Some(frame) = frame else { + break; + }; + if write_frame(&mut writer, &frame, max_frame, write_timeout) + .await + .is_err() + { + writer_source.cancel(); + break; + } + } + }); + let mut tasks = JoinSet::new(); + loop { + while tasks.try_join_next().is_some() {} + let frame = tokio::select! { + () = cancellation.cancelled() => break, + () = connection_cancel.token().cancelled() => break, + frame = read_frame::<_, ClientFrame>(&mut reader, plane.limits.max_frame_bytes, plane.limits.idle_timeout) => frame, + }; + let Ok(ClientFrame::Request(envelope)) = frame else { + break; + }; + if tasks.len() >= plane.limits.max_in_flight_per_connection { + let response = response_error( + envelope.request_id, + &ControlError::new( + ControlErrorCode::Busy, + "connection task limit reached", + true, + ), + ); + if outbound.try_send(ServerFrame::Response(response)).is_err() { + break; + } + continue; + } + let request_core = Arc::clone(&core); + let request_outbound = outbound.clone(); + let request_cancel = connection_cancel.clone(); + tasks.spawn(async move { + let (response, subscription) = request_core.request(envelope).await; + if request_outbound + .try_send(ServerFrame::Response(response)) + .is_err() + { + request_cancel.cancel(); + return; + } + if let Some(mut subscription) = subscription { + while let Some(event) = subscription.recv().await { + if request_outbound + .try_send(ServerFrame::Event(event)) + .is_err() + { + request_cancel.cancel(); + break; + } + } + } + }); + } + connection_cancel.cancel(); + tasks.abort_all(); + while tasks.join_next().await.is_some() {} + drop(outbound); + let _ = tokio::time::timeout(plane.limits.write_timeout, writer_task).await; + Ok(()) +} + +struct TcpClientShared { + writer: tokio::sync::Mutex>, + pending: Mutex>>, + events: tokio::sync::Mutex>, + limits: ControlLimits, + closed: CancellationTokenSource, +} + +#[derive(Clone)] +pub struct TcpControlClient { + shared: Arc, + role: ControlRole, +} + +impl TcpControlClient { + pub async fn connect( + address: SocketAddr, + token: &str, + limits: ControlLimits, + ) -> Result { + if !address.ip().is_loopback() || !limits.is_valid() { + return Err(ControlError::new( + ControlErrorCode::InvalidRequest, + "TCP control client requires bounded loopback configuration", + false, + )); + } + let stream = tokio::time::timeout(limits.request_timeout, TcpStream::connect(address)) + .await + .map_err(|_| { + ControlError::new( + ControlErrorCode::TimedOut, + "control connection timed out", + true, + ) + })? + .map_err(|_| { + ControlError::new( + ControlErrorCode::TransportClosed, + "cannot connect to control server", + true, + ) + })?; + Self::connect_stream(stream, token, limits).await + } + + /// Connects to an explicitly configured TLS control endpoint. Certificate + /// roots and server-name policy are supplied by the embedding client. + pub async fn connect_tls( + address: SocketAddr, + server_name: &str, + token: &str, + limits: ControlLimits, + client_config: Arc, + ) -> Result { + if !limits.is_valid() { + return Err(ControlError::new( + ControlErrorCode::InvalidRequest, + "TLS control client requires bounded configuration", + false, + )); + } + let server_name = + rustls::pki_types::ServerName::try_from(server_name.to_owned()).map_err(|_| { + ControlError::new( + ControlErrorCode::InvalidRequest, + "invalid TLS control server name", + false, + ) + })?; + let stream = tokio::time::timeout(limits.request_timeout, TcpStream::connect(address)) + .await + .map_err(|_| { + ControlError::new( + ControlErrorCode::TimedOut, + "TLS control connection timed out", + true, + ) + })? + .map_err(|_| { + ControlError::new( + ControlErrorCode::TransportClosed, + "cannot connect to TLS control server", + true, + ) + })?; + let stream = tokio::time::timeout( + limits.idle_timeout, + tokio_rustls::TlsConnector::from(client_config).connect(server_name, stream), + ) + .await + .map_err(|_| { + ControlError::new( + ControlErrorCode::TimedOut, + "TLS control handshake timed out", + true, + ) + })? + .map_err(|_| { + ControlError::new( + ControlErrorCode::AuthenticationFailed, + "TLS control server authentication failed", + false, + ) + })?; + Self::connect_stream(stream, token, limits).await + } + + async fn connect_stream( + mut stream: S, + token: &str, + limits: ControlLimits, + ) -> Result + where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, + { + write_frame( + &mut stream, + &ClientFrame::Hello(ClientHello { + version: CONTROL_PROTOCOL_VERSION, + token: token.to_owned(), + }), + limits.max_frame_bytes, + limits.write_timeout, + ) + .await?; + let hello: ServerFrame = + read_frame(&mut stream, limits.max_frame_bytes, limits.idle_timeout).await?; + let ServerFrame::Hello(hello) = hello else { + return Err(ControlError::new( + ControlErrorCode::AuthenticationFailed, + "invalid server handshake", + false, + )); + }; + if let Some(error) = hello.error { + return Err(ControlError { + code: error.code, + message: "control handshake rejected", + retryable: error.retryable, + }); + } + let role = hello.role.ok_or_else(auth_error)?; + let (reader, writer) = tokio::io::split(stream); + let (event_tx, event_rx) = mpsc::channel(limits.event_queue); + let shared = Arc::new(TcpClientShared { + writer: tokio::sync::Mutex::new(Box::new(writer)), + pending: Mutex::new(BTreeMap::new()), + events: tokio::sync::Mutex::new(event_rx), + limits, + closed: CancellationTokenSource::new(), + }); + let reader_shared = Arc::clone(&shared); + tokio::spawn(async move { + tcp_client_reader(reader_shared, Box::new(reader), event_tx).await; + }); + Ok(Self { shared, role }) + } + + pub async fn request( + &self, + envelope: ControlRequestEnvelope, + ) -> Result { + if !valid_identifier(&envelope.request_id) { + return Err(ControlError::new( + ControlErrorCode::InvalidRequest, + "invalid request ID", + false, + )); + } + let (sender, receiver) = oneshot::channel(); + if lock(&self.shared.pending) + .insert(envelope.request_id.clone(), sender) + .is_some() + { + return Err(ControlError::new( + ControlErrorCode::Replay, + "request ID already pending", + false, + )); + } + let write = { + let mut writer = self.shared.writer.lock().await; + write_frame( + &mut **writer, + &ClientFrame::Request(envelope.clone()), + self.shared.limits.max_frame_bytes, + self.shared.limits.write_timeout, + ) + .await + }; + if let Err(error) = write { + lock(&self.shared.pending).remove(&envelope.request_id); + return Err(error); + } + let result = tokio::time::timeout( + self.shared.limits.request_timeout + self.shared.limits.write_timeout, + receiver, + ) + .await + .map_err(|_| { + ControlError::new( + ControlErrorCode::TimedOut, + "control response timed out", + true, + ) + })? + .map_err(|_| { + ControlError::new( + ControlErrorCode::TransportClosed, + "control response channel closed", + true, + ) + }); + if result.is_err() { + lock(&self.shared.pending).remove(&envelope.request_id); + } + result + } + + pub async fn next_event(&self) -> Option { + self.shared.events.lock().await.recv().await + } + + #[must_use] + pub fn role(&self) -> ControlRole { + self.role + } +} + +impl Drop for TcpControlClient { + fn drop(&mut self) { + if Arc::strong_count(&self.shared) == 2 { + self.shared.closed.cancel(); + } + } +} + +async fn tcp_client_reader( + shared: Arc, + mut reader: Box, + events: mpsc::Sender, +) { + loop { + let frame = tokio::select! { + () = shared.closed.token().cancelled() => break, + frame = read_frame::<_, ServerFrame>(&mut reader, shared.limits.max_frame_bytes, shared.limits.idle_timeout) => frame, + }; + match frame { + Ok(ServerFrame::Response(response)) => { + if let Some(sender) = lock(&shared.pending).remove(&response.request_id) { + let _ = sender.send(response); + } + } + Ok(ServerFrame::Event(event)) => { + if events.try_send(event).is_err() { + break; + } + } + _ => break, + } + } + shared.closed.cancel(); + lock(&shared.pending).clear(); +} + +trait ControlRead: AsyncRead + Unpin + Send {} +impl ControlRead for T {} + +trait ControlWrite: AsyncWrite + Unpin + Send {} +impl ControlWrite for T {} + +async fn read_frame(reader: &mut R, maximum: usize, idle: Duration) -> Result +where + R: AsyncRead + Unpin, + T: for<'de> Deserialize<'de>, +{ + let mut header = [0_u8; 4]; + tokio::time::timeout(idle, reader.read_exact(&mut header)) + .await + .map_err(|_| { + ControlError::new( + ControlErrorCode::IdleTimeout, + "control connection idle timeout", + true, + ) + })? + .map_err(|_| { + ControlError::new( + ControlErrorCode::TransportClosed, + "control frame header closed", + true, + ) + })?; + let length = usize::try_from(u32::from_be_bytes(header)).map_err(|_| { + ControlError::new( + ControlErrorCode::FrameTooLarge, + "control frame length is invalid", + false, + ) + })?; + if length == 0 || length > maximum { + return Err(ControlError::new( + ControlErrorCode::FrameTooLarge, + "control frame exceeds configured bound", + false, + )); + } + let mut body = vec![0_u8; length]; + tokio::time::timeout(idle, reader.read_exact(&mut body)) + .await + .map_err(|_| { + ControlError::new( + ControlErrorCode::IdleTimeout, + "partial control frame timed out", + true, + ) + })? + .map_err(|_| { + ControlError::new( + ControlErrorCode::TransportClosed, + "partial control frame closed", + true, + ) + })?; + serde_json::from_slice(&body).map_err(|_| { + ControlError::new( + ControlErrorCode::InvalidRequest, + "control frame JSON is invalid", + false, + ) + }) +} + +async fn write_frame( + writer: &mut W, + value: &T, + maximum: usize, + timeout: Duration, +) -> Result<(), ControlError> +where + W: AsyncWrite + Unpin + ?Sized, + T: Serialize, +{ + let body = serde_json::to_vec(value).map_err(|_| { + ControlError::new( + ControlErrorCode::Internal, + "control frame serialization failed", + false, + ) + })?; + if body.is_empty() || body.len() > maximum || body.len() > u32::MAX as usize { + return Err(ControlError::new( + ControlErrorCode::FrameTooLarge, + "control frame exceeds configured bound", + false, + )); + } + let length = u32::try_from(body.len()).map_err(|_| { + ControlError::new( + ControlErrorCode::FrameTooLarge, + "control frame length overflow", + false, + ) + })?; + tokio::time::timeout(timeout, async { + writer.write_all(&length.to_be_bytes()).await?; + writer.write_all(&body).await?; + writer.flush().await + }) + .await + .map_err(|_| { + ControlError::new( + ControlErrorCode::TimedOut, + "control frame write timed out", + true, + ) + })? + .map_err(|_| { + ControlError::new( + ControlErrorCode::TransportClosed, + "control frame write failed", + true, + ) + }) +} + +fn response_ok(request_id: String, payload: ControlPayload) -> ControlResponseEnvelope { + ControlResponseEnvelope { + version: CONTROL_PROTOCOL_VERSION, + request_id, + result: Ok(payload), + } +} + +fn response_error(request_id: String, error: &ControlError) -> ControlResponseEnvelope { + ControlResponseEnvelope { + version: CONTROL_PROTOCOL_VERSION, + request_id, + result: Err(error.body()), + } +} + +fn auth_error() -> ControlError { + ControlError::new( + ControlErrorCode::AuthenticationFailed, + "control authentication failed", + false, + ) +} + +fn valid_identifier(value: &str) -> bool { + !value.is_empty() + && value.len() <= MAX_REQUEST_ID_BYTES + && value.chars().all(|character| { + character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '.' | ':') + }) +} + +fn valid_uuid_text(value: &str) -> bool { + libremetaverse_types::UUID::parse(value.to_owned()).is_ok() +} + +fn payload_valid(payload: &ControlPayload, maximum_frame_bytes: usize) -> bool { + let page_is_bounded = match payload { + ControlPayload::Sessions(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE), + ControlPayload::ScheduledJobs(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE), + ControlPayload::PendingApprovals(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE), + ControlPayload::AuditEvents(page) => page.items.len() <= usize::from(MAX_PAGE_SIZE), + _ => true, + }; + page_is_bounded + && serde_json::to_vec(payload).is_ok_and(|body| { + // Reserve space for the versioned response envelope and frame tag. + body.len().saturating_add(512) <= maximum_frame_bytes + }) +} + +fn constant_time_eq(left: &str, right: &str) -> bool { + let left = left.as_bytes(); + let right = right.as_bytes(); + let mut difference = left.len() ^ right.len(); + let maximum = left.len().max(right.len()); + for index in 0..maximum { + difference |= usize::from( + left.get(index).copied().unwrap_or(0) ^ right.get(index).copied().unwrap_or(0), + ); + } + difference == 0 +} + +fn role_name(role: ControlRole) -> &'static str { + match role { + ControlRole::Observer => "observer", + ControlRole::Operator => "operator", + } +} + +fn request_name(request: &ControlRequest) -> &'static str { + match request { + ControlRequest::Health => "health", + ControlRequest::Runtime => "runtime", + ControlRequest::ListSessions { .. } => "list_sessions", + ControlRequest::ListScheduledJobs { .. } => "list_scheduled_jobs", + ControlRequest::ListPendingApprovals { .. } => "list_pending_approvals", + ControlRequest::ListAuditEvents { .. } => "list_audit_events", + ControlRequest::SubscribeEvents { .. } => "subscribe_events", + ControlRequest::CancelRequest { .. } => "cancel_request", + ControlRequest::PauseAutonomy => "pause_autonomy", + ControlRequest::ResumeAutonomy => "resume_autonomy", + ControlRequest::CancelAction { .. } => "cancel_action", + ControlRequest::DecideApproval { approve: true, .. } => "approve_proposal", + ControlRequest::DecideApproval { approve: false, .. } => "deny_proposal", + ControlRequest::ForceReconnect => "force_reconnect", + ControlRequest::ExpireConversation { .. } => "expire_conversation", + ControlRequest::SetRoamingJob { enabled: true, .. } => "enable_roaming_job", + ControlRequest::SetRoamingJob { enabled: false, .. } => "disable_roaming_job", + ControlRequest::InjectOperatorMessage { .. } => "inject_operator_message", + ControlRequest::GracefulShutdown => "graceful_shutdown", + } +} + +fn unix_millis() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |duration| { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) + }) +} + +fn lock(value: &Mutex) -> MutexGuard<'_, T> { + value + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} diff --git a/crates/metacrate-grid-agent/src/control_plane_tests.rs b/crates/metacrate-grid-agent/src/control_plane_tests.rs new file mode 100644 index 0000000..786f587 --- /dev/null +++ b/crates/metacrate-grid-agent/src/control_plane_tests.rs @@ -0,0 +1,746 @@ +use crate::config::SecretString; +use crate::control_plane::*; +use libremetaverse_types::compat::CancellationToken; +use serde_json::json; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; + +#[derive(Default)] +struct FakeTarget { + mutations: Mutex>, + shutdown: AtomicBool, + active: AtomicUsize, + maximum_active: AtomicUsize, +} + +impl FakeTarget { + fn mutation_count(&self) -> usize { + self.mutations.lock().expect("mutations").len() + } +} + +impl ControlTarget for FakeTarget { + #[allow(clippy::too_many_lines)] // Exhaustive fake covers every response family. + fn execute( + &self, + context: ControlContext, + request: ControlRequest, + cancellation: CancellationToken, + ) -> ControlFuture<'_> { + Box::pin(async move { + let active = self.active.fetch_add(1, Ordering::AcqRel) + 1; + self.maximum_active.fetch_max(active, Ordering::AcqRel); + let result = match request { + ControlRequest::Health => Ok(ControlPayload::Health(HealthView { + service_state: "running".into(), + ready: true, + uptime_seconds: 42, + protocol_version: CONTROL_PROTOCOL_VERSION, + })), + ControlRequest::Runtime => Ok(ControlPayload::Runtime(RuntimeView { + grid_state: "online".into(), + generation: 7, + transport_connected: true, + agent_ready: true, + region_id: Some("00000000-0000-4000-8000-000000000001".into()), + region_name: Some("Safe Region".into()), + position: Some([128.0, 128.0, 24.0]), + behavior_mode: "available".into(), + control_queue_used: 1, + control_queue_capacity: 64, + budget_tool_calls_used: 3, + budget_movement_millimeters_used: 2_000, + })), + ControlRequest::ListSessions { page } => Ok(ControlPayload::Sessions(page_values( + &page, + (0..7) + .map(|index| SessionMetadataView { + session_id: format!("session-{index}"), + avatar_id: format!("00000000-0000-4000-8000-{index:012x}"), + channel: "direct_im".into(), + created_unix_millis: 10, + last_active_unix_millis: 20, + turns: 2, + bytes: 32, + }) + .collect(), + ))), + ControlRequest::ListScheduledJobs { page } => { + Ok(ControlPayload::ScheduledJobs(page_values( + &page, + vec![ScheduledJobView { + job_id: "roam-1".into(), + kind: "landmark_roam".into(), + enabled: true, + next_run_unix_millis: Some(50), + }], + ))) + } + ControlRequest::ListPendingApprovals { page } => { + Ok(ControlPayload::PendingApprovals(page_values( + &page, + vec![PendingApprovalView { + approval_id: 1, + tool: "build".into(), + principal: "operator:local".into(), + expires_unix_seconds: 99, + movement_millimeters: 0, + inventory_operations: 0, + build_prims: 1, + }], + ))) + } + ControlRequest::ListAuditEvents { page } => { + Ok(ControlPayload::AuditEvents(page_values( + &page, + vec![AuditEventView { + sequence: 1, + unix_millis: 1, + principal: "operator:local".into(), + operation: "pause_autonomy".into(), + outcome: "completed".into(), + authorization_id: None, + }], + ))) + } + ControlRequest::InjectOperatorMessage { message } if message == "wait" => { + cancellation.cancelled().await; + Err(ControlError { + code: ControlErrorCode::Cancelled, + message: "operation cancelled", + retryable: false, + }) + } + ControlRequest::GracefulShutdown => { + self.shutdown.store(true, Ordering::Release); + self.mutations + .lock() + .expect("mutations") + .push("graceful_shutdown".into()); + Ok(ControlPayload::Completed) + } + other => { + assert_eq!(context.role, ControlRole::Operator); + self.mutations + .lock() + .expect("mutations") + .push(format!("{other:?}")); + Ok(ControlPayload::Completed) + } + }; + self.active.fetch_sub(1, Ordering::AcqRel); + result + }) + } +} + +fn page_values(page: &PageRequest, values: Vec) -> Page { + let start = usize::try_from(page.cursor.unwrap_or(0)) + .expect("cursor") + .min(values.len()); + let end = start + .saturating_add(usize::from(page.limit)) + .min(values.len()); + let next_cursor = (end < values.len()).then(|| u64::try_from(end).expect("cursor")); + Page { + items: values.into_iter().skip(start).take(end - start).collect(), + next_cursor, + } +} + +fn token(value: &str) -> SecretString { + SecretString::new("control.test_token", value).expect("token") +} + +fn limits() -> ControlLimits { + ControlLimits { + request_timeout: Duration::from_secs(5), + idle_timeout: Duration::from_secs(5), + write_timeout: Duration::from_secs(1), + ..ControlLimits::default() + } +} + +fn plane(target: Arc, configured: ControlLimits) -> Arc { + ControlPlane::new( + target, + token("operator-secret"), + Some(token("observer-secret")), + configured, + ) + .expect("plane") +} + +fn request(id: &str, request: ControlRequest) -> ControlRequestEnvelope { + ControlRequestEnvelope { + version: CONTROL_PROTOCOL_VERSION, + request_id: id.into(), + request, + } +} + +enum Client { + InProcess(InProcessControlClient), + Tcp(TcpControlClient), +} + +impl Client { + async fn request(&self, envelope: ControlRequestEnvelope) -> ControlResponseEnvelope { + match self { + Self::InProcess(client) => client.request(envelope).await, + Self::Tcp(client) => client.request(envelope).await.expect("TCP response"), + } + } +} + +enum Transport { + InProcess, + Tcp, +} + +struct Fixture { + client: Client, + plane: Arc, + target: Arc, + server: Option, +} + +async fn fixture(transport: Transport, role: ControlRole, configured: ControlLimits) -> Fixture { + let target = Arc::new(FakeTarget::default()); + let plane = plane(target.clone(), configured); + let value = if role == ControlRole::Operator { + "operator-secret" + } else { + "observer-secret" + }; + match transport { + Transport::InProcess => Fixture { + client: Client::InProcess(plane.connect(value).expect("connect")), + plane, + target, + server: None, + }, + Transport::Tcp => { + let server = TcpControlServer::bind( + plane.clone(), + TcpControlConfig { + listen: "127.0.0.1:0".parse().expect("address"), + limits: configured, + }, + ) + .await + .expect("bind"); + let client = TcpControlClient::connect(server.local_addr(), value, configured) + .await + .expect("connect"); + Fixture { + client: Client::Tcp(client), + plane, + target, + server: Some(server), + } + } + } +} + +async fn close(fixture: Fixture) { + drop(fixture.client); + if let Some(server) = fixture.server { + server.shutdown().await.expect("shutdown"); + } +} + +async fn conformance(transport: Transport) { + let fixture = fixture(transport, ControlRole::Operator, limits()).await; + let health = fixture + .client + .request(request("health-1", ControlRequest::Health)) + .await; + assert_eq!(health.version, CONTROL_PROTOCOL_VERSION); + assert!(matches!( + health.result, + Ok(ControlPayload::Health(HealthView { ready: true, .. })) + )); + let page = fixture + .client + .request(request( + "sessions-1", + ControlRequest::ListSessions { + page: PageRequest { + cursor: Some(2), + limit: 3, + }, + }, + )) + .await; + let Ok(ControlPayload::Sessions(page)) = page.result else { + panic!("session page") + }; + assert_eq!(page.items.len(), 3); + assert_eq!(page.items[0].session_id, "session-2"); + assert_eq!(page.next_cursor, Some(5)); + assert!( + fixture + .client + .request(request("pause-1", ControlRequest::PauseAutonomy)) + .await + .result + .is_ok() + ); + let replay = fixture + .client + .request(request("pause-1", ControlRequest::ResumeAutonomy)) + .await; + assert_eq!( + replay.result.expect_err("replay").code, + ControlErrorCode::Replay + ); + assert_eq!(fixture.target.mutation_count(), 1); + let shutdown = fixture + .client + .request(request("shutdown-1", ControlRequest::GracefulShutdown)) + .await; + assert!(shutdown.result.is_ok()); + assert!(fixture.target.shutdown.load(Ordering::Acquire)); + close(fixture).await; +} + +#[tokio::test] +async fn same_conformance_suite_runs_in_process() { + conformance(Transport::InProcess).await; +} + +#[tokio::test] +async fn same_conformance_suite_runs_over_loopback_tcp() { + conformance(Transport::Tcp).await; +} + +#[tokio::test] +async fn observer_is_read_only_and_tokens_are_redacted() { + let fixture = fixture(Transport::InProcess, ControlRole::Observer, limits()).await; + assert!(matches!( + fixture + .client + .request(request("read", ControlRequest::Runtime)) + .await + .result, + Ok(ControlPayload::Runtime(_)) + )); + let denied = fixture + .client + .request(request("mutate", ControlRequest::PauseAutonomy)) + .await; + assert_eq!( + denied.result.expect_err("denied").code, + ControlErrorCode::PermissionDenied + ); + assert_eq!(fixture.target.mutation_count(), 0); + assert!(!format!("{:?}", fixture.plane).contains("operator-secret")); + let Err(error) = fixture.plane.connect("wrong") else { + panic!("bad token was accepted") + }; + assert_eq!(error.code, ControlErrorCode::AuthenticationFailed); + close(fixture).await; +} + +#[tokio::test] +async fn version_bounds_and_malformed_requests_are_typed() { + let fixture = fixture(Transport::InProcess, ControlRole::Operator, limits()).await; + let mut wrong = request("version", ControlRequest::Health); + wrong.version = 99; + assert_eq!( + fixture + .client + .request(wrong) + .await + .result + .expect_err("version") + .code, + ControlErrorCode::VersionMismatch + ); + let invalid = fixture + .client + .request(request( + "page", + ControlRequest::ListSessions { + page: PageRequest { + cursor: None, + limit: 101, + }, + }, + )) + .await; + assert_eq!( + invalid.result.expect_err("page").code, + ControlErrorCode::InvalidRequest + ); + let secret_payload = fixture + .client + .request(request( + "audit", + ControlRequest::ListAuditEvents { + page: PageRequest::default(), + }, + )) + .await; + let rendered = serde_json::to_string(&secret_payload).expect("JSON"); + for secret in ["operator-secret", "observer-secret", "Bearer ", "CAPS/"] { + assert!(!rendered.contains(secret)); + } + close(fixture).await; +} + +#[tokio::test] +async fn cancellation_is_per_request_and_concurrent_operators_stay_responsive() { + for transport in [Transport::InProcess, Transport::Tcp] { + let fixture = fixture(transport, ControlRole::Operator, limits()).await; + let slow = match &fixture.client { + Client::InProcess(value) => Client::InProcess(value.clone()), + Client::Tcp(value) => Client::Tcp(value.clone()), + }; + let task = tokio::spawn(async move { + slow.request(request( + "slow", + ControlRequest::InjectOperatorMessage { + message: "wait".into(), + }, + )) + .await + }); + tokio::time::timeout(Duration::from_secs(1), async { + while fixture.target.active.load(Ordering::Acquire) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("slow request became active"); + let concurrent_operator = match &fixture.server { + Some(server) => Client::Tcp( + TcpControlClient::connect(server.local_addr(), "operator-secret", limits()) + .await + .expect("second TCP operator"), + ), + None => Client::InProcess( + fixture + .plane + .connect("operator-secret") + .expect("second in-process operator"), + ), + }; + let health = concurrent_operator + .request(request("health-during", ControlRequest::Health)) + .await; + assert!(health.result.is_ok()); + let cancelled = fixture + .client + .request(request( + "cancel", + ControlRequest::CancelRequest { + target_request_id: "slow".into(), + }, + )) + .await; + assert!(matches!( + cancelled.result, + Ok(ControlPayload::Cancelled { .. }) + )); + let slow = task.await.expect("slow task"); + assert_eq!( + slow.result.expect_err("cancelled").code, + ControlErrorCode::Cancelled + ); + assert!(fixture.target.maximum_active.load(Ordering::Acquire) >= 2); + drop(concurrent_operator); + close(fixture).await; + } +} + +#[tokio::test] +async fn event_stream_resubscribes_with_explicit_gap_and_slow_consumers_are_cut_off() { + let mut configured = limits(); + configured.event_queue = 2; + configured.event_history = 3; + let fixture = fixture(Transport::InProcess, ControlRole::Operator, configured).await; + let mut in_subscription = None; + if let Client::InProcess(client) = &fixture.client { + let (_, subscription) = client + .subscribe(request( + "subscribe", + ControlRequest::SubscribeEvents { + after_sequence: Some(0), + }, + )) + .await + .expect("subscribe"); + in_subscription = Some(subscription); + } + for index in 0..10 { + fixture.plane.publish(ControlEventKind::StateChanged { + component: "grid".into(), + state: format!("state-{index}"), + }); + } + assert_eq!(fixture.plane.retained_event_count(), 3); + let first = in_subscription + .as_mut() + .expect("subscription") + .recv() + .await + .expect("buffered"); + assert_eq!(first.sequence, 1); + drop(in_subscription); + let client = fixture.plane.connect("operator-secret").expect("reconnect"); + let (_, mut resumed) = client + .subscribe(request( + "resubscribe", + ControlRequest::SubscribeEvents { + after_sequence: Some(1), + }, + )) + .await + .expect("resubscribe"); + let gap = resumed.recv().await.expect("gap"); + assert!(matches!( + gap.event, + ControlEventKind::Gap { + first_available: 9, + last_missed: 8 + } + )); + close(fixture).await; +} + +#[tokio::test] +async fn zero_client_event_flood_remains_bounded_and_headless() { + let target = Arc::new(FakeTarget::default()); + let mut configured = limits(); + configured.event_history = 4; + let plane = plane(target, configured); + assert_eq!(plane.active_connections(), 0); + for index in 0..10_000 { + plane.publish(ControlEventKind::StateChanged { + component: "load".into(), + state: index.to_string(), + }); + } + assert_eq!(plane.retained_event_count(), 4); + assert_eq!(plane.active_connections(), 0); +} + +#[tokio::test] +async fn subscriptions_have_a_hard_global_bound_and_release_on_drop() { + let mut configured = limits(); + configured.max_subscriptions = 2; + let target = Arc::new(FakeTarget::default()); + let plane = plane(target, configured); + let client = plane.connect("operator-secret").expect("client"); + let (_, first) = client + .subscribe(request( + "subscribe-1", + ControlRequest::SubscribeEvents { + after_sequence: None, + }, + )) + .await + .expect("first subscription"); + let (_, second) = client + .subscribe(request( + "subscribe-2", + ControlRequest::SubscribeEvents { + after_sequence: None, + }, + )) + .await + .expect("second subscription"); + let error = client + .subscribe(request( + "subscribe-3", + ControlRequest::SubscribeEvents { + after_sequence: None, + }, + )) + .await + .err() + .expect("subscription bound"); + assert_eq!(error.code, ControlErrorCode::Busy); + drop(first); + client + .subscribe(request( + "subscribe-4", + ControlRequest::SubscribeEvents { + after_sequence: None, + }, + )) + .await + .expect("released subscription"); + drop(second); +} + +#[tokio::test] +#[allow(clippy::too_many_lines)] // One raw-transport matrix shares one listener. +async fn tcp_rejects_bad_tokens_oversized_and_partial_frames() { + let configured = limits(); + let target = Arc::new(FakeTarget::default()); + let plane = plane(target, configured); + let server = TcpControlServer::bind( + plane.clone(), + TcpControlConfig { + listen: "127.0.0.1:0".parse().expect("address"), + limits: configured, + }, + ) + .await + .expect("server"); + let Err(error) = TcpControlClient::connect(server.local_addr(), "wrong", configured).await + else { + panic!("bad token was accepted") + }; + assert_eq!(error.code, ControlErrorCode::AuthenticationFailed); + + let mut wrong_version = TcpStream::connect(server.local_addr()) + .await + .expect("version connect"); + let hello = serde_json::to_vec(&json!({ + "frame":"hello", + "body":{"version":99,"token":"operator-secret"} + })) + .expect("hello JSON"); + wrong_version + .write_all( + &u32::try_from(hello.len()) + .expect("frame length") + .to_be_bytes(), + ) + .await + .expect("version header"); + wrong_version.write_all(&hello).await.expect("version body"); + let mut response_header = [0_u8; 4]; + wrong_version + .read_exact(&mut response_header) + .await + .expect("version response header"); + let mut response = + vec![0_u8; usize::try_from(u32::from_be_bytes(response_header)).expect("response length")]; + wrong_version + .read_exact(&mut response) + .await + .expect("version response"); + let response: serde_json::Value = serde_json::from_slice(&response).expect("version JSON"); + assert_eq!(response["body"]["error"]["code"], json!("version_mismatch")); + + let subscriber = TcpControlClient::connect(server.local_addr(), "operator-secret", configured) + .await + .expect("subscriber"); + let subscription_response = subscriber + .request(request( + "events", + ControlRequest::SubscribeEvents { + after_sequence: None, + }, + )) + .await + .expect("subscription response"); + assert!(matches!( + subscription_response.result, + Ok(ControlPayload::Subscribed { .. }) + )); + plane.publish(ControlEventKind::StateChanged { + component: "session".into(), + state: "online".into(), + }); + let event = tokio::time::timeout(Duration::from_secs(1), subscriber.next_event()) + .await + .expect("event deadline") + .expect("event"); + assert!(matches!(event.event, ControlEventKind::StateChanged { .. })); + let last_sequence = event.sequence; + drop(subscriber); + plane.publish(ControlEventKind::StateChanged { + component: "session".into(), + state: "reconnecting".into(), + }); + let reconnected = TcpControlClient::connect(server.local_addr(), "operator-secret", configured) + .await + .expect("reconnected subscriber"); + reconnected + .request(request( + "events-resumed", + ControlRequest::SubscribeEvents { + after_sequence: Some(last_sequence), + }, + )) + .await + .expect("resubscription response"); + let replayed = tokio::time::timeout(Duration::from_secs(1), reconnected.next_event()) + .await + .expect("replay deadline") + .expect("replayed event"); + assert_eq!(replayed.sequence, last_sequence.saturating_add(1)); + let mut oversized = TcpStream::connect(server.local_addr()) + .await + .expect("raw connect"); + oversized + .write_all(&(u32::try_from(configured.max_frame_bytes).expect("bound") + 1).to_be_bytes()) + .await + .expect("header"); + let mut byte = [0_u8; 1]; + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), oversized.read(&mut byte)) + .await + .expect("closed") + .expect("read"), + 0 + ); + let mut partial = TcpStream::connect(server.local_addr()) + .await + .expect("raw connect"); + partial + .write_all(&100_u32.to_be_bytes()) + .await + .expect("header"); + partial.write_all(b"{}").await.expect("partial"); + partial.shutdown().await.expect("close"); + drop(reconnected); + server.shutdown().await.expect("shutdown"); +} + +#[tokio::test] +async fn plaintext_server_rejects_non_loopback_before_binding() { + let configured = limits(); + let target = Arc::new(FakeTarget::default()); + let plane = plane(target, configured); + let Err(error) = TcpControlServer::bind( + plane, + TcpControlConfig { + listen: "0.0.0.0:7943".parse().expect("address"), + limits: configured, + }, + ) + .await + else { + panic!("plaintext remote listener was accepted") + }; + assert_eq!(error.code, ControlErrorCode::InvalidRequest); +} + +#[test] +fn plain_tcp_is_loopback_only_and_protocol_json_is_stable() { + let configured = limits(); + assert!( + !TcpControlConfig { + listen: "0.0.0.0:7943".parse().expect("address"), + limits: configured + } + .listen + .ip() + .is_loopback() + ); + let encoded = serde_json::to_value(request("health", ControlRequest::Health)).expect("JSON"); + assert_eq!( + encoded, + json!({"version":1,"request_id":"health","request":{"method":"health"}}) + ); +} diff --git a/crates/metacrate-grid-agent/src/control_runtime.rs b/crates/metacrate-grid-agent/src/control_runtime.rs new file mode 100644 index 0000000..8ab99a1 --- /dev/null +++ b/crates/metacrate-grid-agent/src/control_runtime.rs @@ -0,0 +1,522 @@ +//! Production management target backed by the agent's bounded runtime stores. + +#![allow(clippy::missing_errors_doc)] + +use crate::behavior::{BehaviorIngress, BehaviorMode}; +use crate::control_plane::{ + AuditEventView, CONTROL_PROTOCOL_VERSION, ControlContext, ControlError, ControlErrorCode, + ControlFuture, ControlPayload, ControlRequest, ControlTarget, ConversationChannelView, + HealthView, Page, PageRequest, PendingApprovalView, RuntimeView, ScheduledJobView, + SessionMetadataView, +}; +use crate::conversation::{ConversationChannel, ConversationKey, ConversationStore}; +use crate::policy::{ + ApprovalId, AuthenticatedPrincipal, MemoryPolicyAudit, PolicyFinalOutcome, PolicyGateway, + PolicyReasonCode, +}; +use crate::session::{SessionState, SessionStatus}; +use libremetaverse_types::compat::CancellationToken; +use std::collections::BTreeMap; +use std::fmt; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; +use tokio::sync::mpsc; + +const ROAMING_JOB_ID: &str = "default-roaming"; + +/// Commands whose ownership remains with the service lifecycle loop. +pub enum RuntimeControlCommand { + Pause, + Resume, + ForceReconnect, + OperatorMessage { message: String }, + GracefulShutdown, +} + +impl fmt::Debug for RuntimeControlCommand { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::OperatorMessage { message } => formatter + .debug_struct("OperatorMessage") + .field("message_bytes", &message.len()) + .finish(), + Self::Pause => formatter.write_str("Pause"), + Self::Resume => formatter.write_str("Resume"), + Self::ForceReconnect => formatter.write_str("ForceReconnect"), + Self::GracefulShutdown => formatter.write_str("GracefulShutdown"), + } + } +} + +#[derive(Clone, Debug)] +struct RuntimeState { + session: SessionStatus, + behavior: BehaviorMode, + service_state: &'static str, + region_id: Option, + region_name: Option, + position: Option<[f64; 3]>, +} + +impl Default for RuntimeState { + fn default() -> Self { + Self { + session: SessionStatus::default(), + behavior: BehaviorMode::Offline, + service_state: "starting", + region_id: None, + region_name: None, + position: None, + } + } +} + +/// Real control target shared by integrated and split transports. +pub struct AgentControlTarget { + started: Instant, + state: Mutex, + conversations: Arc, + policy: Arc, + audit: Arc, + behavior: BehaviorIngress, + commands: mpsc::Sender, + command_capacity: usize, + jobs: Mutex>, + next_operation: AtomicU64, +} + +impl fmt::Debug for AgentControlTarget { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("AgentControlTarget") + .field("state", &lock(&self.state)) + .field("command_capacity", &self.command_capacity) + .finish_non_exhaustive() + } +} + +impl AgentControlTarget { + pub fn new( + conversations: Arc, + policy: Arc, + audit: Arc, + behavior: BehaviorIngress, + command_capacity: usize, + ) -> Result<(Arc, mpsc::Receiver), ControlError> { + if command_capacity == 0 || command_capacity > 8_192 { + return Err(control_error( + ControlErrorCode::InvalidRequest, + "invalid runtime control queue capacity", + false, + )); + } + let (commands, receiver) = mpsc::channel(command_capacity); + let mut jobs = BTreeMap::new(); + jobs.insert(ROAMING_JOB_ID.to_owned(), false); + Ok(( + Arc::new(Self { + started: Instant::now(), + state: Mutex::new(RuntimeState::default()), + conversations, + policy, + audit, + behavior, + commands, + command_capacity, + jobs: Mutex::new(jobs), + next_operation: AtomicU64::new(1), + }), + receiver, + )) + } + + pub fn update_session(&self, session: SessionStatus) { + let mut state = lock(&self.state); + state.session = session; + state.service_state = if session.state == SessionState::ShuttingDown { + "stopping" + } else { + "running" + }; + } + + pub fn update_behavior(&self, behavior: BehaviorMode) { + lock(&self.state).behavior = behavior; + } + + pub fn update_region( + &self, + region_id: Option, + region_name: Option, + position: Option<[f64; 3]>, + ) { + let mut state = lock(&self.state); + state.region_id = region_id.filter(|value| value.len() <= 64); + state.region_name = region_name.filter(|value| value.len() <= 256); + state.position = + position.filter(|value| value.iter().all(|component| component.is_finite())); + } + + pub fn mark_stopping(&self) { + lock(&self.state).service_state = "stopping"; + } + + fn enqueue( + &self, + operation: &'static str, + command: RuntimeControlCommand, + ) -> Result { + self.commands + .try_send(command) + .map_err(|error| match error { + mpsc::error::TrySendError::Full(_) => control_error( + ControlErrorCode::Backpressure, + "runtime control queue is full", + true, + ), + mpsc::error::TrySendError::Closed(_) => control_error( + ControlErrorCode::TransportClosed, + "runtime control queue is closed", + true, + ), + })?; + Ok(ControlPayload::Accepted { + operation_id: format!( + "{operation}-{}", + self.next_operation.fetch_add(1, Ordering::Relaxed) + ), + }) + } + + #[allow(clippy::too_many_lines)] // Exhaustive protocol-to-runtime mapping stays auditable. + fn execute_now( + &self, + context: &ControlContext, + request: ControlRequest, + ) -> Result { + match request { + ControlRequest::Health => { + let state = lock(&self.state); + Ok(ControlPayload::Health(HealthView { + service_state: state.service_state.to_owned(), + ready: state.session.agent_ready, + uptime_seconds: self.started.elapsed().as_secs(), + protocol_version: CONTROL_PROTOCOL_VERSION, + })) + } + ControlRequest::Runtime => { + let state = lock(&self.state).clone(); + let usage = self.policy.global_budget_usage(); + Ok(ControlPayload::Runtime(RuntimeView { + grid_state: state.session.state.as_str().to_owned(), + generation: state.session.generation, + transport_connected: state.session.transport_connected, + agent_ready: state.session.agent_ready, + region_id: state.region_id, + region_name: state.region_name, + position: state.position, + behavior_mode: behavior_name(state.behavior).to_owned(), + control_queue_used: self + .command_capacity + .saturating_sub(self.commands.capacity()), + control_queue_capacity: self.command_capacity, + budget_tool_calls_used: usage.tool_calls, + budget_movement_millimeters_used: usage.movement_millimeters, + })) + } + ControlRequest::ListSessions { page } => { + let values = self + .conversations + .list_metadata() + .into_iter() + .map(|metadata| SessionMetadataView { + session_id: metadata.session_id.as_str().to_owned(), + avatar_id: metadata.avatar_id.to_string(), + channel: match metadata.channel { + ConversationChannel::PublicChat => "public_chat", + ConversationChannel::DirectIm => "direct_im", + } + .to_owned(), + created_unix_millis: metadata.created_unix_millis, + last_active_unix_millis: metadata.last_active_unix_millis, + turns: metadata.turns, + bytes: metadata.bytes, + }) + .collect(); + Ok(ControlPayload::Sessions(page_values(&page, values))) + } + ControlRequest::ListScheduledJobs { page } => { + let values = lock(&self.jobs) + .iter() + .map(|(job_id, enabled)| ScheduledJobView { + job_id: job_id.clone(), + kind: "bounded_roaming".to_owned(), + enabled: *enabled, + next_run_unix_millis: None, + }) + .collect(); + Ok(ControlPayload::ScheduledJobs(page_values(&page, values))) + } + ControlRequest::ListPendingApprovals { page } => { + let values = self + .policy + .pending_approvals(unix_seconds()) + .into_iter() + .map(|approval| PendingApprovalView { + approval_id: approval.id.get(), + tool: approval.tool, + principal: approval.principal, + expires_unix_seconds: approval.expires_at, + movement_millimeters: approval.cost.movement_millimeters, + inventory_operations: approval.cost.inventory_operations, + build_prims: approval.cost.build_prims, + }) + .collect(); + Ok(ControlPayload::PendingApprovals(page_values(&page, values))) + } + ControlRequest::ListAuditEvents { page } => { + let values = self + .audit + .snapshot() + .into_iter() + .enumerate() + .map(|(index, record)| AuditEventView { + sequence: u64::try_from(index).unwrap_or(u64::MAX).saturating_add(1), + unix_millis: record.recorded_unix_millis, + principal: record.principal.as_str().to_owned(), + operation: record.tool.as_str().to_owned(), + outcome: policy_outcome_name(record.final_outcome).to_owned(), + authorization_id: record.authorization_id, + }) + .collect(); + Ok(ControlPayload::AuditEvents(page_values(&page, values))) + } + ControlRequest::PauseAutonomy => { + let response = self.enqueue("pause", RuntimeControlCommand::Pause)?; + self.behavior.pause(); + Ok(response) + } + ControlRequest::ResumeAutonomy => { + let response = self.enqueue("resume", RuntimeControlCommand::Resume)?; + self.behavior.resume(); + Ok(response) + } + ControlRequest::CancelAction { action_id } => { + let cancelled = self.behavior.cancel_action(&action_id).map_err(|_| { + control_error(ControlErrorCode::InvalidRequest, "invalid action ID", false) + })?; + if !cancelled { + return Err(control_error( + ControlErrorCode::NotFound, + "active behavior action not found", + false, + )); + } + Ok(ControlPayload::Completed) + } + ControlRequest::DecideApproval { + approval_id, + approve, + } => { + let id = ApprovalId::from_raw(approval_id).ok_or_else(|| { + control_error( + ControlErrorCode::InvalidRequest, + "invalid approval ID", + false, + ) + })?; + let principal = + AuthenticatedPrincipal::from_authenticated_control(context.principal.clone()) + .map_err(|_| { + control_error( + ControlErrorCode::PermissionDenied, + "invalid operator principal", + false, + ) + })?; + let reason = if approve { + self.policy.grant_approval(id, &principal, unix_seconds()) + } else { + self.policy.deny_approval(id, &principal, unix_seconds()) + } + .map_err(|_| { + control_error( + ControlErrorCode::Internal, + "approval decision failed", + false, + ) + })?; + match reason { + PolicyReasonCode::Allowed | PolicyReasonCode::ApprovalNotGranted => { + Ok(ControlPayload::Completed) + } + PolicyReasonCode::ApprovalUnknown => Err(control_error( + ControlErrorCode::NotFound, + "approval not found", + false, + )), + PolicyReasonCode::ApprovalExpired | PolicyReasonCode::ApprovalReplayed => { + Err(control_error( + ControlErrorCode::Conflict, + "approval is no longer pending", + false, + )) + } + _ => Err(control_error( + ControlErrorCode::Conflict, + "approval decision rejected", + false, + )), + } + } + ControlRequest::ForceReconnect => { + self.enqueue("reconnect", RuntimeControlCommand::ForceReconnect) + } + ControlRequest::ExpireConversation { avatar_id, channel } => { + let avatar_id = libremetaverse_types::UUID::parse(avatar_id).map_err(|_| { + control_error(ControlErrorCode::InvalidRequest, "invalid avatar ID", false) + })?; + let channel = match channel { + ConversationChannelView::PublicChat => ConversationChannel::PublicChat, + ConversationChannelView::DirectIm => ConversationChannel::DirectIm, + }; + let key = ConversationKey::new(avatar_id, channel).map_err(|_| { + control_error( + ControlErrorCode::InvalidRequest, + "invalid conversation key", + false, + ) + })?; + if self.conversations.expire(key) { + Ok(ControlPayload::Completed) + } else { + Err(control_error( + ControlErrorCode::NotFound, + "conversation not found", + false, + )) + } + } + ControlRequest::SetRoamingJob { job_id, enabled } => { + let mut jobs = lock(&self.jobs); + let Some(current) = jobs.get_mut(&job_id) else { + return Err(control_error( + ControlErrorCode::NotFound, + "roaming job not found", + false, + )); + }; + self.behavior.set_roaming(enabled).map_err(|_| { + control_error( + ControlErrorCode::Backpressure, + "behavior queue unavailable", + true, + ) + })?; + *current = enabled; + Ok(ControlPayload::Completed) + } + ControlRequest::InjectOperatorMessage { message } => self.enqueue( + "operator-message", + RuntimeControlCommand::OperatorMessage { message }, + ), + ControlRequest::GracefulShutdown => { + let response = self.enqueue("shutdown", RuntimeControlCommand::GracefulShutdown)?; + self.mark_stopping(); + Ok(response) + } + ControlRequest::SubscribeEvents { .. } | ControlRequest::CancelRequest { .. } => { + Err(control_error( + ControlErrorCode::Internal, + "transport request reached runtime", + false, + )) + } + } + } +} + +impl ControlTarget for AgentControlTarget { + fn execute( + &self, + context: ControlContext, + request: ControlRequest, + cancellation: CancellationToken, + ) -> ControlFuture<'_> { + Box::pin(async move { + if cancellation.is_cancellation_requested() { + return Err(control_error( + ControlErrorCode::Cancelled, + "control request cancelled", + false, + )); + } + self.execute_now(&context, request) + }) + } +} + +fn page_values(request: &PageRequest, values: Vec) -> Page { + let start = usize::try_from(request.cursor.unwrap_or(0)) + .unwrap_or(usize::MAX) + .min(values.len()); + let end = start + .saturating_add(usize::from(request.limit)) + .min(values.len()); + let next_cursor = (end < values.len()).then(|| u64::try_from(end).unwrap_or(u64::MAX)); + Page { + items: values.into_iter().skip(start).take(end - start).collect(), + next_cursor, + } +} + +const fn behavior_name(mode: BehaviorMode) -> &'static str { + match mode { + BehaviorMode::Offline => "offline", + BehaviorMode::Settling => "settling", + BehaviorMode::Available => "available", + BehaviorMode::Engaged => "engaged", + BehaviorMode::Executing => "executing", + BehaviorMode::Roaming => "roaming", + BehaviorMode::Paused => "paused", + BehaviorMode::Recovering => "recovering", + } +} + +const fn policy_outcome_name(outcome: PolicyFinalOutcome) -> &'static str { + match outcome { + PolicyFinalOutcome::Denied => "denied", + PolicyFinalOutcome::ApprovalRequired => "approval_required", + PolicyFinalOutcome::ApprovalGranted => "approval_granted", + PolicyFinalOutcome::Authorized => "authorized", + PolicyFinalOutcome::Completed => "completed", + PolicyFinalOutcome::Rejected => "rejected", + PolicyFinalOutcome::Failed => "failed", + PolicyFinalOutcome::AmbiguousMutation => "ambiguous_mutation", + } +} + +fn unix_seconds() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |duration| duration.as_secs()) +} + +const fn control_error( + code: ControlErrorCode, + message: &'static str, + retryable: bool, +) -> ControlError { + ControlError { + code, + message, + retryable, + } +} + +fn lock(value: &Mutex) -> MutexGuard<'_, T> { + value + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} diff --git a/crates/metacrate-grid-agent/src/control_runtime_tests.rs b/crates/metacrate-grid-agent/src/control_runtime_tests.rs new file mode 100644 index 0000000..7f74c69 --- /dev/null +++ b/crates/metacrate-grid-agent/src/control_runtime_tests.rs @@ -0,0 +1,167 @@ +use crate::behavior::{BehaviorController, BehaviorError, EmbodimentFuture, EmbodimentSink}; +use crate::control_plane::{ + CONTROL_PROTOCOL_VERSION, ControlLimits, ControlPayload, ControlPlane, ControlRequest, + ControlRequestEnvelope, +}; +use crate::control_runtime::{AgentControlTarget, RuntimeControlCommand}; +use crate::conversation::{ConversationLimits, ConversationStore}; +use crate::perception::WorldPosition; +use crate::policy::{MemoryPolicyAudit, PolicyAuditSink, PolicyGateway, PolicyLimits}; +use crate::{AgentConfig, EmbodiedPose, SecretString}; +use libremetaverse_types::UUID; +use libremetaverse_types::compat::CancellationToken; +use std::collections::BTreeSet; +use std::sync::Arc; +use std::time::Duration; + +struct Sink; + +impl EmbodimentSink for Sink { + fn current_pose( + &self, + generation: u64, + _cancellation: CancellationToken, + ) -> EmbodimentFuture<'_, EmbodiedPose> { + Box::pin(async move { + Ok(EmbodiedPose { + generation, + region_id: UUID::zero(), + position: WorldPosition { + x: 128.0, + y: 128.0, + z: 24.0, + }, + heading_degrees: 0.0, + sitting: false, + }) + }) + } + + fn resolve_avatar( + &self, + _generation: u64, + _avatar_id: UUID, + _cancellation: CancellationToken, + ) -> EmbodimentFuture<'_, WorldPosition> { + Box::pin(async { Err(BehaviorError::TargetUnavailable) }) + } + + fn face_point( + &self, + _generation: u64, + _point: WorldPosition, + _cancellation: CancellationToken, + ) -> EmbodimentFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } + + fn begin_walk( + &self, + _generation: u64, + _point: WorldPosition, + _cancellation: CancellationToken, + ) -> EmbodimentFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } + + fn validate_walk_target( + &self, + _generation: u64, + _point: WorldPosition, + _cancellation: CancellationToken, + ) -> EmbodimentFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } + + fn stop(&self, _generation: u64, _cancellation: CancellationToken) -> EmbodimentFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } + + fn sit(&self, _generation: u64, _cancellation: CancellationToken) -> EmbodimentFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } + + fn stand( + &self, + _generation: u64, + _cancellation: CancellationToken, + ) -> EmbodimentFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } +} + +fn envelope(id: &str, request: ControlRequest) -> ControlRequestEnvelope { + ControlRequestEnvelope { + version: CONTROL_PROTOCOL_VERSION, + request_id: id.to_owned(), + request, + } +} + +#[tokio::test] +async fn production_target_projects_state_and_routes_real_mutations() { + let settings = AgentConfig::offline("https://llm.invalid/chat", "llm-key") + .expect("offline config") + .behavior; + let behavior = BehaviorController::new(settings, Arc::new(Sink), 8, 8, Duration::from_secs(1)) + .expect("behavior") + .start(); + let audit = Arc::new(MemoryPolicyAudit::new(32).expect("audit")); + let audit_sink: Arc = audit.clone(); + let policy = Arc::new( + PolicyGateway::new( + BTreeSet::new(), + Vec::new(), + PolicyLimits::default(), + audit_sink, + ) + .expect("policy"), + ); + let conversations = Arc::new( + ConversationStore::open(ConversationLimits::default(), None).expect("conversations"), + ); + let (target, mut commands) = + AgentControlTarget::new(conversations, policy, audit, behavior.ingress(), 8) + .expect("target"); + target.update_region( + Some("00000000-0000-4000-8000-000000000001".into()), + Some("Test Region".into()), + Some([128.0, 128.0, 24.0]), + ); + let plane = ControlPlane::new( + target, + SecretString::new("test.operator", "operator-token").expect("token"), + Some(SecretString::new("test.observer", "observer-token").expect("token")), + ControlLimits::default(), + ) + .expect("plane"); + let operator = plane.connect("operator-token").expect("operator"); + let runtime = operator + .request(envelope("runtime", ControlRequest::Runtime)) + .await; + let Ok(ControlPayload::Runtime(runtime)) = runtime.result else { + panic!("runtime projection") + }; + assert_eq!(runtime.region_name.as_deref(), Some("Test Region")); + assert_eq!(runtime.control_queue_capacity, 8); + + assert!( + operator + .request(envelope("pause", ControlRequest::PauseAutonomy)) + .await + .result + .is_ok() + ); + assert!(matches!( + commands.recv().await, + Some(RuntimeControlCommand::Pause) + )); + + let observer = plane.connect("observer-token").expect("observer"); + let denied = observer + .request(envelope("shutdown", ControlRequest::GracefulShutdown)) + .await; + assert!(denied.result.is_err()); + assert!(commands.try_recv().is_err()); + behavior.shutdown().await.expect("shutdown behavior"); +} diff --git a/crates/metacrate-grid-agent/src/lib.rs b/crates/metacrate-grid-agent/src/lib.rs index 56bef74..2261300 100644 --- a/crates/metacrate-grid-agent/src/lib.rs +++ b/crates/metacrate-grid-agent/src/lib.rs @@ -7,6 +7,8 @@ pub mod backend; pub mod behavior; pub mod config; +pub mod control_plane; +pub mod control_runtime; pub mod conversation; pub mod interaction; pub mod llm; @@ -20,6 +22,10 @@ pub mod types; #[cfg(test)] mod behavior_tests; #[cfg(test)] +mod control_plane_tests; +#[cfg(test)] +mod control_runtime_tests; +#[cfg(test)] mod conversation_tests; #[cfg(test)] mod interaction_tests; @@ -47,10 +53,19 @@ pub use behavior::{ STOP_TOOL, WALK_SHORT_TOOL, behavior_policy_tools, }; pub use config::{ - AgentConfig, BehaviorSettings, ConfigError, ConfigLoader, ConversationSettings, EndpointUrl, - Environment, GridConnection, Limits, LlmConnection, MapEnvironment, OperatingMode, - SecretString, StdEnvironment, Timeouts, + AgentConfig, BehaviorSettings, ConfigError, ConfigLoader, ControlSettings, + ConversationSettings, EndpointUrl, Environment, GridConnection, Limits, LlmConnection, + MapEnvironment, OperatingMode, RemoteTlsSettings, SecretString, StdEnvironment, Timeouts, }; +pub use control_plane::{ + AuditEventView, CONTROL_PROTOCOL_VERSION, ControlContext, ControlError, ControlErrorBody, + ControlErrorCode, ControlEvent, ControlEventKind, ControlFuture, ControlLimits, ControlPayload, + ControlPlane, ControlRequest, ControlRequestEnvelope, ControlResponseEnvelope, ControlRole, + ControlSubscription, ControlTarget, ConversationChannelView, HealthView, + InProcessControlClient, Page, PageRequest, PendingApprovalView, RuntimeView, ScheduledJobView, + SessionMetadataView, TcpControlClient, TcpControlConfig, TcpControlServer, +}; +pub use control_runtime::{AgentControlTarget, RuntimeControlCommand}; pub use conversation::{ ConversationChannel, ConversationClock, ConversationContext, ConversationError, ConversationKey, ConversationLimits, ConversationMetadata, ConversationPersistence, @@ -81,10 +96,11 @@ pub use perception::{ pub use policy::{ ActionOrigin, AllowedOrigins, ApprovalId, ApprovalRule, AuthenticatedPrincipal, AuthorizedAction, BudgetLimits, Capability, FixedCost, Idempotency, MemoryPolicyAudit, - OriginClass, PolicyAuditError, PolicyAuditRecord, PolicyAuditSink, PolicyDisposition, - PolicyError, PolicyEvaluation, PolicyFinalOutcome, PolicyGateway, PolicyLimits, - PolicyReasonCode, PolicyRequestContext, PolicySnapshot, PolicyTool, PolicyToolExecutor, - ResourceCost, ResourceEstimator, Risk, SchedulerGrantId, UntrustedData, UntrustedSource, + OriginClass, PendingApprovalMetadata, PolicyAuditError, PolicyAuditRecord, PolicyAuditSink, + PolicyDisposition, PolicyError, PolicyEvaluation, PolicyFinalOutcome, PolicyGateway, + PolicyLimits, PolicyReasonCode, PolicyRequestContext, PolicySnapshot, PolicyTool, + PolicyToolExecutor, ResourceCost, ResourceEstimator, Risk, SchedulerGrantId, UntrustedData, + UntrustedSource, }; pub use service::{AgentService, ServiceError, ServiceHandle, ServiceState}; pub use session::{ diff --git a/crates/metacrate-grid-agent/src/main.rs b/crates/metacrate-grid-agent/src/main.rs index e61b4d0..0c872cb 100644 --- a/crates/metacrate-grid-agent/src/main.rs +++ b/crates/metacrate-grid-agent/src/main.rs @@ -147,8 +147,10 @@ async fn run_live( run_once: bool, ) -> Result<(), Box> { use metacrate_grid_agent::{ - GridSessionBackend, LibremetaverseClientOwner, SessionObservation, SessionState, - SessionSupervisor, + AgentControlTarget, BehaviorObservation, ControlEventKind, ControlPlane, ControlTarget, + GridSessionBackend, LibremetaverseClientOwner, OperatingMode, RuntimeControlCommand, + SessionControl, SessionObservation, SessionState, SessionSupervisor, TcpControlConfig, + TcpControlServer, }; let connection = config.grid.clone().ok_or_else(|| { @@ -230,8 +232,56 @@ async fn run_live( return Ok(()); } + let (control_target, mut control_commands) = AgentControlTarget::new( + live.conversations.clone(), + live.policy.clone(), + live.audit.clone(), + live.behavior.ingress(), + config.limits.control_queue, + )?; + control_target.update_session(handle.status()); + let erased_target: Arc = control_target.clone(); + let (control_plane, _integrated_client, control_server) = match config.mode { + OperatingMode::Integrated => { + let (plane, client) = ControlPlane::integrated(erased_target, config.control.limits)?; + (plane, Some(client), None) + } + OperatingMode::SplitService => { + let operator = config.control.operator_token.clone().ok_or_else(|| { + CliError("validated split configuration omitted its operator token".into()) + })?; + let plane = ControlPlane::new( + erased_target, + operator, + config.control.observer_token.clone(), + config.control.limits, + )?; + let transport = TcpControlConfig { + listen: config.control.listen, + limits: config.control.limits, + }; + let server = if let Some(tls) = &config.control.remote_tls { + eprintln!( + "WARNING: remote control is enabled with TLS; protect operator tokens and certificate keys" + ); + TcpControlServer::bind_tls(plane.clone(), transport, tls.server_config()?).await? + } else { + TcpControlServer::bind(plane.clone(), transport).await? + }; + println!( + "grid agent control plane listening on {}", + server.local_addr() + ); + (plane, None, Some(server)) + } + OperatingMode::OfflineFake => { + return Err(CliError("offline mode reached the live control plane".into()).into()); + } + }; + println!("grid agent session supervisor started; press Ctrl-C to stop"); let mut signal_error = None; + let mut control_failure = None; loop { tokio::select! { signal = tokio::signal::ctrl_c() => { @@ -243,6 +293,11 @@ async fn run_live( event = handle.next_observation() => { let Some(event) = event else { break; }; if let SessionObservation::Transition { status, reason, retry_in } = event { + control_target.update_session(status); + control_plane.publish(ControlEventKind::StateChanged { + component: "session".into(), + state: status.state.as_str().into(), + }); println!( "grid session state={} generation={} transport_connected={} agent_ready={} reason={reason:?} retry_in={retry_in:?}", status.state.as_str(), @@ -262,19 +317,66 @@ async fn run_live( } event = live.behavior.next_observation() => { let Some(event) = event else { break; }; + if let BehaviorObservation::Transition { to, .. } = &event { + control_target.update_behavior(*to); + control_plane.publish(ControlEventKind::StateChanged { + component: "behavior".into(), + state: format!("{to:?}").to_ascii_lowercase(), + }); + } println!("grid behavior event={event:?}"); } + command = control_commands.recv() => { + let Some(command) = command else { break; }; + match command { + RuntimeControlCommand::Pause => { + if let Err(error) = handle.control(SessionControl::Pause).await { + control_failure = Some(error.to_string()); + break; + } + } + RuntimeControlCommand::Resume => { + if let Err(error) = handle.control(SessionControl::Resume).await { + control_failure = Some(error.to_string()); + break; + } + } + RuntimeControlCommand::ForceReconnect => { + if let Err(error) = handle.control(SessionControl::ForceReconnect).await { + control_failure = Some(error.to_string()); + break; + } + } + RuntimeControlCommand::OperatorMessage { message } => { + control_plane.publish(ControlEventKind::StateChanged { + component: "operator_message".into(), + state: format!("accepted_{}_bytes", message.len()), + }); + } + RuntimeControlCommand::GracefulShutdown => break, + } + } } } + control_target.mark_stopping(); let session_result = handle.shutdown().await; let interaction_result = live.interaction.shutdown().await; let behavior_result = live.behavior.shutdown().await; + let control_result = if let Some(server) = control_server { + server.shutdown().await + } else { + Ok(()) + }; session_result?; interaction_result?; behavior_result?; + control_result?; if let Some(error) = signal_error { return Err(error.into()); } + if let Some(error) = control_failure { + return Err(CliError(format!("control command failed: {error}")).into()); + } println!("grid agent stopped cleanly"); Ok(()) } @@ -286,9 +388,13 @@ struct LiveInteractions { perception_observations: tokio::sync::mpsc::Receiver, behavior: metacrate_grid_agent::BehaviorHandle, + conversations: Arc, + policy: Arc, + audit: Arc, } #[cfg(feature = "live-grid")] +#[allow(clippy::too_many_lines)] fn start_live_interactions( config: &metacrate_grid_agent::AgentConfig, owner: &metacrate_grid_agent::LibremetaverseClientOwner, @@ -347,7 +453,7 @@ fn start_live_interactions( config.authorized_avatar_uuids.clone(), tools, PolicyLimits::default(), - audit, + audit.clone(), )?); let loop_limits = ToolLoopLimits { max_tool_calls_per_turn: config.limits.max_tool_calls, @@ -366,7 +472,7 @@ fn start_live_interactions( Arc::new(AuthorizedBackendRouter::new(routes)?); let responder = Arc::new(PolicyLlmResponder::new( client, - gateway, + gateway.clone(), routed_backend, loop_limits, now, @@ -377,7 +483,7 @@ fn start_live_interactions( config.interaction.clone(), libremetaverse_types::UUID::zero(), config.authorized_avatar_uuids.clone(), - conversation, + conversation.clone(), responder, sink, config.limits.grid_event_queue, @@ -391,5 +497,8 @@ fn start_live_interactions( perception: perception_ingress, perception_observations, behavior, + conversations: conversation, + policy: gateway, + audit, }) } diff --git a/crates/metacrate-grid-agent/src/policy.rs b/crates/metacrate-grid-agent/src/policy.rs index 7c7a761..e76414b 100644 --- a/crates/metacrate-grid-agent/src/policy.rs +++ b/crates/metacrate-grid-agent/src/policy.rs @@ -16,6 +16,7 @@ use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::error::Error; use std::fmt; use std::sync::{Arc, Mutex}; +use std::time::{SystemTime, UNIX_EPOCH}; const MAX_POLICY_TOOLS: usize = 64; const MAX_AUTHORIZED_AVATARS: usize = 1_024; @@ -238,6 +239,16 @@ pub struct ResourceCost { pub build_prims: u64, } +/// Secret-free operator projection of one pending approval. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct PendingApprovalMetadata { + pub id: ApprovalId, + pub principal: String, + pub tool: String, + pub expires_at: u64, + pub cost: ResourceCost, +} + impl ResourceCost { #[must_use] pub const fn one_call() -> Self { @@ -514,6 +525,7 @@ pub enum PolicyFinalOutcome { #[derive(Clone, Debug, Eq, PartialEq)] pub struct PolicyAuditRecord { + pub recorded_unix_millis: u64, pub authorization_id: Option, pub approval_id: Option, pub origin: OriginClass, @@ -581,6 +593,18 @@ impl PolicyAuditSink for MemoryPolicyAudit { #[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)] pub struct ApprovalId(u64); +impl ApprovalId { + #[must_use] + pub const fn from_raw(value: u64) -> Option { + if value == 0 { None } else { Some(Self(value)) } + } + + #[must_use] + pub const fn get(self) -> u64 { + self.0 + } +} + #[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)] pub struct SchedulerGrantId(u64); @@ -588,6 +612,7 @@ pub struct SchedulerGrantId(u64); enum ApprovalStatus { Pending, Granted, + Denied, Consumed, } @@ -820,6 +845,31 @@ impl PolicyGateway { PolicySnapshot(lock(&self.state).clone()) } + /// Returns bounded, content-free approval metadata for operator UIs. + #[must_use] + pub fn pending_approvals(&self, now: u64) -> Vec { + lock(&self.state) + .approvals + .values() + .filter(|approval| { + approval.status == ApprovalStatus::Pending && approval.expires_at > now + }) + .map(|approval| PendingApprovalMetadata { + id: approval.id, + principal: approval.principal.audit_label(), + tool: approval.tool.clone(), + expires_at: approval.expires_at, + cost: approval.cost, + }) + .collect() + } + + /// Current global budget use for a redacted control-plane projection. + #[must_use] + pub fn global_budget_usage(&self) -> ResourceCost { + lock(&self.state).global_used + } + #[cfg(test)] pub(crate) fn pending_approval_count(&self) -> usize { lock(&self.state) @@ -1064,7 +1114,9 @@ impl PolicyGateway { match approval.status { ApprovalStatus::Pending => Some(PolicyReasonCode::ApprovalNotGranted), ApprovalStatus::Granted => None, - ApprovalStatus::Consumed => Some(PolicyReasonCode::ApprovalReplayed), + ApprovalStatus::Denied | ApprovalStatus::Consumed => { + Some(PolicyReasonCode::ApprovalReplayed) + } } }; if let Some(reason) = reason { @@ -1121,8 +1173,12 @@ impl PolicyGateway { } if state.approvals.len() == MAX_APPROVALS { let removable = state.approvals.iter().find_map(|(id, approval)| { - (approval.expires_at <= now || approval.status == ApprovalStatus::Consumed) - .then_some(*id) + (approval.expires_at <= now + || matches!( + approval.status, + ApprovalStatus::Denied | ApprovalStatus::Consumed + )) + .then_some(*id) }); let Some(removable) = removable else { drop(state); @@ -1323,6 +1379,40 @@ impl PolicyGateway { Ok(PolicyReasonCode::Allowed) } + pub fn deny_approval( + &self, + id: ApprovalId, + _operator: &AuthenticatedPrincipal, + now: u64, + ) -> Result { + let mut state = lock(&self.state); + let Some(approval) = state.approvals.get_mut(&id) else { + return Ok(PolicyReasonCode::ApprovalUnknown); + }; + if approval.expires_at <= now { + return Ok(PolicyReasonCode::ApprovalExpired); + } + if approval.status != ApprovalStatus::Pending { + return Ok(PolicyReasonCode::ApprovalReplayed); + } + let audit = audit_record( + &approval.context, + approval.origin, + &approval.principal, + BoundedText::new("policy.tool", approval.tool.clone())?, + None, + Some(id), + PolicyDisposition::Denied, + PolicyReasonCode::ApprovalNotGranted, + approval.cost, + approval.arguments_hash, + PolicyFinalOutcome::Denied, + )?; + self.emit(audit)?; + approval.status = ApprovalStatus::Denied; + Ok(PolicyReasonCode::ApprovalNotGranted) + } + /// Consumes the originating authorization so it cannot be replayed to /// create more than one scheduler grant. #[allow(clippy::needless_pass_by_value)] @@ -1621,6 +1711,11 @@ fn audit_record( final_outcome: PolicyFinalOutcome, ) -> Result { Ok(PolicyAuditRecord { + recorded_unix_millis: SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |duration| { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) + }), authorization_id, approval_id, origin, diff --git a/crates/metacrate-grid-agent/src/policy_tests.rs b/crates/metacrate-grid-agent/src/policy_tests.rs index 03d9fed..b77dc60 100644 --- a/crates/metacrate-grid-agent/src/policy_tests.rs +++ b/crates/metacrate-grid-agent/src/policy_tests.rs @@ -411,6 +411,7 @@ fn approval_tool() -> PolicyTool { } #[test] +#[allow(clippy::too_many_lines)] // One approval lifecycle includes deny and expiry paths. fn approvals_bind_principal_canonical_arguments_expiry_and_one_execution() { let (gateway, audit) = gateway_with(vec![approval_tool()], PolicyLimits::default()); let requester = context(OriginClass::AuthorizedIm); @@ -507,6 +508,24 @@ fn approvals_bind_principal_canonical_arguments_expiry_and_one_execution() { .expect("expired result"), PolicyReasonCode::ApprovalExpired ); + let denied = gateway + .evaluate(&requester, &proposed, &arguments, None, 600) + .expect("denial pending") + .approval_id + .expect("approval ID"); + assert_eq!(gateway.pending_approvals(600).len(), 1); + assert_eq!( + gateway.deny_approval(denied, &operator, 601).expect("deny"), + PolicyReasonCode::ApprovalNotGranted + ); + assert!(gateway.pending_approvals(601).is_empty()); + assert_eq!( + gateway + .evaluate(&requester, &proposed, &arguments, Some(denied), 601) + .expect("denied approval cannot execute") + .reason, + PolicyReasonCode::ApprovalReplayed + ); assert!( audit .snapshot() diff --git a/crates/metacrate-grid-agent/tests/dependency_policy.rs b/crates/metacrate-grid-agent/tests/dependency_policy.rs index 5553852..905b032 100644 --- a/crates/metacrate-grid-agent/tests/dependency_policy.rs +++ b/crates/metacrate-grid-agent/tests/dependency_policy.rs @@ -2,14 +2,16 @@ use std::collections::BTreeSet; use std::fs; use std::path::{Path, PathBuf}; -const ALLOWED_DEPENDENCIES: [&str; 8] = [ +const ALLOWED_DEPENDENCIES: [&str; 10] = [ "libremetaverse", "libremetaverse-types", "reqwest", + "rustls", "serde", "serde_json", "sha2", "tokio", + "tokio-rustls", "url", ]; @@ -46,7 +48,7 @@ fn runtime_source_has_no_subprocess_or_native_abi_escape_hatch() { let mut files = Vec::with_capacity(20); collect_rust_files(&source, &mut files); assert!( - files.len() <= 20, + files.len() <= 24, "source-file count needs a reviewed bound update" ); for path in files { diff --git a/docs/grid-agent-control-plane.md b/docs/grid-agent-control-plane.md new file mode 100644 index 0000000..4f1cb0d --- /dev/null +++ b/docs/grid-agent-control-plane.md @@ -0,0 +1,91 @@ +# Grid-agent control plane v1 + +The grid agent exposes one transport-neutral request/response and event API. +Integrated clients receive an `InProcessControlClient`; split clients use the +same `ControlRequestEnvelope`, `ControlResponseEnvelope`, and `ControlEvent` +types over TCP. No terminal, Unix socket, named-pipe, permission-bit, or +platform-specific type appears in the core API. + +## Transport and negotiation + +TCP frames are a four-byte unsigned big-endian length followed by one UTF-8 +JSON value. Empty, partial, malformed, idle, and oversized frames close the +connection within configured deadlines. The first frame must be: + +```json +{"frame":"hello","body":{"version":1,"token":"separate-operator-capability"}} +``` + +The server answers with a `hello` containing version `1`, the authenticated +`operator` or `observer` role, or a typed error. Every later client frame is a +`request`; server frames are `response` or `event`. Request IDs contain at most +96 ASCII identifier bytes and cannot be replayed on a connection. A caller can +cancel an active request by its ID. Responses and events can arrive in either +order, so clients correlate responses by request ID. + +Plain TCP binds only to IPv4 or IPv6 loopback. Remote control is off by default +and a non-loopback address is rejected unless both DER certificate and private +key paths are explicitly configured. Remote mode uses Rustls before the hello +exchange and emits an operator warning at startup. The TLS client API requires +the embedding client to supply its certificate roots and server-name policy. +Every split connection must still authenticate with a dedicated control token; +observer and operator tokens must differ and configuration rejects reuse of an +LLM API key or grid password. Tokens are absent from protocol errors, audit, +serialization traits, and diagnostics. + +Remote service configuration uses the JSON fields +`control.remote_tls_certificate_der` and +`control.remote_tls_private_key_der` together with a non-loopback +`control.listen` value. Both files are bounded, non-symlink DER files. Omitting +either field, or selecting a non-loopback address without both, fails +configuration before a listener is created. + +## Requests + +The JSON request envelope is stable and versioned. For example: + +```json +{"version":1,"request_id":"health-1","request":{"method":"health"}} +``` + +Observers can call `health`, `runtime`, `list_sessions`, +`list_scheduled_jobs`, `list_pending_approvals`, `list_audit_events`, and +`subscribe_events`. Operators can additionally call `cancel_request`, +`pause_autonomy`, `resume_autonomy`, `cancel_action`, `decide_approval`, +`force_reconnect`, `expire_conversation`, `set_roaming_job`, +`inject_operator_message`, and `graceful_shutdown`. Cancellation is a mutation +and is operator-only. List requests use an opaque numeric cursor and a page +size of 1 through 100. + +Runtime projections contain lifecycle/readiness, session generation, safe +region and pose fields when known, behavior mode, control-queue utilization, +and aggregate budget use. Conversation responses contain metadata only. Audit +and approval responses exclude arguments, prompt contents, credentials, +authorization headers, capability URLs, model reasoning, and filesystem data. +`cancel_action` binds to the exact bounded action ID carried by behavior audit +observations; it can cancel a queued or executing embodied action without +preempting unrelated work. The built-in roaming job ID is `default-roaming`. + +Errors are typed as `authentication_failed`, `version_mismatch`, +`permission_denied`, `invalid_request`, `replay`, `not_found`, `conflict`, +`cancelled`, `timed_out`, `busy`, `backpressure`, `frame_too_large`, +`idle_timeout`, `transport_closed`, or `internal`, with a bounded safe message +and a retryable flag. + +## Events, bounds, and reconnect + +Events carry monotonically increasing sequence numbers. A subscription can +resume after its last observed sequence. If retained history no longer covers +that point, its first item is an explicit `gap` with `first_available` and +`last_missed`, followed by retained events. Mutation events record the +authenticated role/connection principal, operation name, and completed or +rejected outcome, never request contents. + +Connections, unauthenticated handshakes, in-flight request tasks, +subscriptions, command/event queues, replay history, event history, event +bytes, event rate, frames, pages, idle time, request time, and writes all have +validated hard bounds. A full event or writer queue disconnects the slow +consumer instead of blocking the headless service. With zero clients, event +publication retains only the configured history and creates no background +work. Server shutdown cancels and joins all listener, connection, request, and +writer tasks within its deadline.