From 1e1e95a58a8c80a3410167b0dc1b49a081f7eabd Mon Sep 17 00:00:00 2001 From: Chili Palmer Date: Mon, 17 Aug 2026 20:15:37 +0000 Subject: [PATCH] feat(grid-agent): establish architecture and config (#118) --- Cargo.lock | 12 + Cargo.toml | 2 + ci/dependency-policy.json | 1 + config/grid-agent.example.json | 29 + crates/metacrate-grid-agent/Cargo.toml | 31 + crates/metacrate-grid-agent/README.md | 40 + crates/metacrate-grid-agent/src/backend.rs | 156 +++ crates/metacrate-grid-agent/src/config.rs | 1121 +++++++++++++++++ crates/metacrate-grid-agent/src/lib.rs | 25 + crates/metacrate-grid-agent/src/main.rs | 135 ++ crates/metacrate-grid-agent/src/service.rs | 513 ++++++++ crates/metacrate-grid-agent/src/types.rs | 408 ++++++ .../tests/dependency_policy.rs | 87 ++ docs/grid-agent-architecture.md | 87 ++ 14 files changed, 2647 insertions(+) create mode 100644 config/grid-agent.example.json create mode 100644 crates/metacrate-grid-agent/Cargo.toml create mode 100644 crates/metacrate-grid-agent/README.md create mode 100644 crates/metacrate-grid-agent/src/backend.rs create mode 100644 crates/metacrate-grid-agent/src/config.rs create mode 100644 crates/metacrate-grid-agent/src/lib.rs create mode 100644 crates/metacrate-grid-agent/src/main.rs create mode 100644 crates/metacrate-grid-agent/src/service.rs create mode 100644 crates/metacrate-grid-agent/src/types.rs create mode 100644 crates/metacrate-grid-agent/tests/dependency_policy.rs create mode 100644 docs/grid-agent-architecture.md diff --git a/Cargo.lock b/Cargo.lock index 7dd372c..3f7fb67 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2194,6 +2194,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "metacrate-grid-agent" +version = "0.0.1" +dependencies = [ + "libremetaverse", + "libremetaverse-types", + "serde", + "serde_json", + "tokio", + "url", +] + [[package]] name = "metacrate-performance" version = "0.0.1" diff --git a/Cargo.toml b/Cargo.toml index 29358fc..fec0739 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,7 @@ members = [ "crates/libremetaverse-voice-vivox", "crates/libremetaverse-opus", "crates/libremetaverse-voice-webrtc", + "crates/metacrate-grid-agent", "programs", "tests/compat", "tools/codegen", @@ -38,6 +39,7 @@ default-members = [ "crates/libremetaverse-voice-vivox", "crates/libremetaverse-opus", "crates/libremetaverse-voice-webrtc", + "crates/metacrate-grid-agent", ] [workspace.package] diff --git a/ci/dependency-policy.json b/ci/dependency-policy.json index 725bbc6..02beb7a 100644 --- a/ci/dependency-policy.json +++ b/ci/dependency-policy.json @@ -36,6 +36,7 @@ { "name": "tar", "versions": ["0.4.46"], "purpose": "Bounded OAR, asset, and release-package archive traversal", "maintenance": "active", "transitive_cost": "low", "native": false, "rewrite_anchor": "`tar`" }, { "name": "tokio", "versions": ["1.53.1"], "purpose": "Shared asynchronous networking, timers, channels, and tasks", "maintenance": "active", "transitive_cost": "medium", "native": false, "rewrite_anchor": "`tokio`" }, { "name": "unicode-general-category", "versions": ["1.1.0"], "purpose": "Unicode category matching in the LSL lexer", "maintenance": "stable", "transitive_cost": "low", "native": false, "rewrite_anchor": "`unicode-general-category`" }, + { "name": "url", "versions": ["2.5.8"], "purpose": "Exact validation and credential-safe diagnostic redaction for grid-agent HTTP endpoints", "maintenance": "active", "transitive_cost": "medium", "native": false, "rewrite_anchor": "`url`" }, { "name": "uuid", "versions": ["1.24.0"], "purpose": "Random UUID generation behind protocol-compatible wrappers", "maintenance": "active", "transitive_cost": "low", "native": false, "rewrite_anchor": "`uuid`" }, { "name": "vcpkg", "versions": ["0.2.15"], "purpose": "Windows MSVC native library discovery in reviewed adapters", "maintenance": "stable", "transitive_cost": "low", "native": false, "rewrite_anchor": "`vcpkg`" }, { "name": "vorbis_rs", "versions": ["0.5.6"], "purpose": "Opt-in Ogg Vorbis asset encoding", "maintenance": "monitored-native", "transitive_cost": "medium", "native": true, "rewrite_anchor": "`vorbis_rs`" } diff --git a/config/grid-agent.example.json b/config/grid-agent.example.json new file mode 100644 index 0000000..01f1a7b --- /dev/null +++ b/config/grid-agent.example.json @@ -0,0 +1,29 @@ +{ + "integrated": false, + "split": false, + "llm": { + "endpoint_url": "https://llm.example.invalid/v1/chat/completions", + "api_key": "" + }, + "authorized_avatar_uuids": [], + "timeouts": { + "startup_seconds": 30, + "shutdown_seconds": 10, + "request_seconds": 60 + }, + "limits": { + "grid_event_queue": 256, + "control_queue": 32, + "observable_queue": 512, + "max_body_bytes": 1048576, + "max_message_bytes": 16384, + "max_conversation_messages": 64, + "max_tool_calls": 16, + "max_authorized_avatars": 128, + "max_background_tasks": 2 + }, + "storage_path": "data/grid-agent", + "behavior": { + "heartbeat_seconds": 30 + } +} diff --git a/crates/metacrate-grid-agent/Cargo.toml b/crates/metacrate-grid-agent/Cargo.toml new file mode 100644 index 0000000..23d44a4 --- /dev/null +++ b/crates/metacrate-grid-agent/Cargo.toml @@ -0,0 +1,31 @@ +[package] +name = "metacrate-grid-agent" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Bounded pure-Rust OpenSim grid-agent service foundation" +publish = false + +[dependencies] +libremetaverse = { version = "0.0.1", path = "../libremetaverse", default-features = false, optional = true } +libremetaverse-types = { version = "0.0.1", path = "../libremetaverse-types" } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tokio = { version = "1.53.1", features = ["macros", "rt", "sync", "time"] } +url = "2.5.8" + +[target.'cfg(any(unix, windows))'.dependencies] +tokio = { version = "1.53.1", features = ["rt-multi-thread", "signal"] } + +[features] +default = [] +live-grid = ["dep:libremetaverse"] + +[lints] +workspace = true + +[[bin]] +name = "metacrate-grid-agent" +path = "src/main.rs" diff --git a/crates/metacrate-grid-agent/README.md b/crates/metacrate-grid-agent/README.md new file mode 100644 index 0000000..492b901 --- /dev/null +++ b/crates/metacrate-grid-agent/README.md @@ -0,0 +1,40 @@ +# MetaCrate grid agent + +This package is the bounded, provider-neutral foundation for the MetaCrate +OpenSim grid agent. It contains a reusable library and the +`metacrate-grid-agent` service binary. The first implementation is deliberately +offline: it publishes a deterministic ready event, accepts control commands, +and shuts down both owned tasks without contacting a grid or LLM. The +`live-grid` feature exposes the side-effect-free owner for the existing +`libremetaverse::GridClient`; later live adapters must extend that manager graph +instead of adding a protocol client. + +The LLM connection identity has exactly two resolved fields: +`llm.endpoint_url` and `llm.api_key`. The endpoint is used exactly as supplied; +there are no providers, presets, base-URL rewrites, model catalogs, discovery, +or provider SDKs. `Debug`/`Display` output removes API keys, grid passwords, URL +user information, and URL query values. Secret wrappers are not serializable. + +Configuration precedence, from lowest to highest, is built-in defaults, an +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, +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. + +Run the focused offline gate with: + +```sh +cargo test --locked -p metacrate-grid-agent +cargo clippy --locked -p metacrate-grid-agent --all-targets -- -D warnings +cargo run --locked -p metacrate-grid-agent -- \ + --config config/grid-agent.example.json --check-config +cargo run --locked -p metacrate-grid-agent -- \ + --config config/grid-agent.example.json --run-once +``` + +See [`../../docs/grid-agent-architecture.md`](../../docs/grid-agent-architecture.md) +for queue/task ownership, shutdown, and trust boundaries. diff --git a/crates/metacrate-grid-agent/src/backend.rs b/crates/metacrate-grid-agent/src/backend.rs new file mode 100644 index 0000000..1726948 --- /dev/null +++ b/crates/metacrate-grid-agent/src/backend.rs @@ -0,0 +1,156 @@ +//! Narrow injected boundaries between orchestration and grid/world I/O. + +use crate::types::{GridEvent, GridEventKind, PolicyDecision, ProposedToolCall, ToolCallOutcome}; +use libremetaverse_types::compat::CancellationToken; +use std::error::Error; +use std::fmt; +use std::future::Future; +use std::pin::Pin; +use tokio::sync::mpsc; + +pub type BackendFuture<'a, T> = Pin + Send + 'a>>; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum BackendError { + Configuration { component: &'static str }, + EventQueueClosed, + RejectedMutation, + Operation { operation: &'static str }, +} + +impl fmt::Display for BackendError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Configuration { component } => { + write!(formatter, "backend configuration failed for {component}") + } + Self::EventQueueClosed => formatter.write_str("grid-event owner closed its queue"), + Self::RejectedMutation => { + formatter.write_str("world mutation lacks an approved policy decision") + } + Self::Operation { operation } => { + write!(formatter, "backend operation failed: {operation}") + } + } + } +} + +impl Error for BackendError {} + +/// One owned grid event source. Implementations send into the coordinator-owned +/// bounded queue and must finish when cancellation is requested. +pub trait GridBackend: Send + Sync + 'static { + fn name(&self) -> &'static str; + + fn run( + &self, + events: mpsc::Sender, + cancellation: CancellationToken, + ) -> BackendFuture<'_, Result<(), BackendError>>; +} + +/// The sole world-mutation boundary. Later tool implementations cannot bypass +/// the policy decision passed to this trait, and fake/live implementations use +/// the same call shape. +pub trait WorldMutator: Send + Sync + 'static { + fn apply( + &self, + call: ProposedToolCall, + decision: PolicyDecision, + cancellation: CancellationToken, + ) -> BackendFuture<'_, Result>; +} + +/// Inert deterministic backend used by the foundational offline service. +/// +/// It performs no login or network operation and is available without the +/// opt-in live-grid dependency graph. +#[derive(Clone, Copy, Debug, Default)] +pub struct OfflineGridBackend; + +impl OfflineGridBackend { + #[must_use] + pub const fn new() -> Self { + Self + } +} + +/// Live-feature composition owner that guarantees production adapters reuse +/// the existing `libremetaverse` manager/client graph. +#[cfg(feature = "live-grid")] +#[derive(Debug)] +pub struct LibremetaverseClientOwner { + client: libremetaverse::GridClient, +} + +#[cfg(feature = "live-grid")] +impl LibremetaverseClientOwner { + /// Builds the shared client composition root without starting login or I/O. + /// + /// # Errors + /// + /// Returns a backend configuration error if the shared client defaults are invalid. + pub fn new() -> Result { + let client = libremetaverse::GridClientBuilder::default() + .build() + .map_err(|_| BackendError::Configuration { + component: "libremetaverse client defaults", + })?; + Ok(Self { client }) + } + + #[must_use] + pub const fn client(&self) -> &libremetaverse::GridClient { + &self.client + } +} + +impl GridBackend for OfflineGridBackend { + fn name(&self) -> &'static str { + "offline-fake" + } + + fn run( + &self, + events: mpsc::Sender, + cancellation: CancellationToken, + ) -> BackendFuture<'_, Result<(), BackendError>> { + Box::pin(async move { + let ready = GridEvent { + sequence: 1, + kind: GridEventKind::BackendReady, + }; + tokio::select! { + () = cancellation.cancelled() => return Ok(()), + result = events.send(ready) => { + result.map_err(|_| BackendError::EventQueueClosed)?; + } + } + cancellation.cancelled().await; + Ok(()) + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use libremetaverse_types::compat::CancellationTokenSource; + + #[tokio::test] + async fn offline_backend_stops_on_cancellation() { + let backend = OfflineGridBackend::new(); + let (sender, mut receiver) = mpsc::channel(1); + let cancellation = CancellationTokenSource::new(); + let run = backend.run(sender, cancellation.token()); + tokio::pin!(run); + tokio::select! { + event = receiver.recv() => { + assert_eq!(event.expect("ready event").kind, GridEventKind::BackendReady); + } + result = &mut run => panic!("backend exited before ready: {result:?}"), + } + cancellation.cancel(); + run.await.expect("clean cancellation"); + } +} diff --git a/crates/metacrate-grid-agent/src/config.rs b/crates/metacrate-grid-agent/src/config.rs new file mode 100644 index 0000000..751b1b7 --- /dev/null +++ b/crates/metacrate-grid-agent/src/config.rs @@ -0,0 +1,1121 @@ +//! Layered configuration with validation and secret-safe formatting. + +use crate::types::{MAX_BODY_BYTES, MAX_CONVERSATION_MESSAGES, MAX_MESSAGE_BYTES, MAX_TOOL_CALLS}; +use libremetaverse_types::UUID; +use serde::Deserialize; +use std::collections::{BTreeMap, BTreeSet}; +use std::error::Error; +use std::fmt; +use std::fs; +use std::path::{Path, PathBuf}; +use std::time::Duration; +use url::Url; + +const MAX_CONFIG_BYTES: u64 = 64 * 1024; +const MAX_SECRET_BYTES: u64 = 16 * 1024; +const MAX_AUTHORIZED_AVATARS: usize = 1_024; +const MAX_QUEUE_CAPACITY: usize = 8_192; +const MAX_BACKGROUND_TASKS: usize = 2; + +const ENV_INTEGRATED: &str = "METACRATE_AGENT_INTEGRATED"; +const ENV_SPLIT: &str = "METACRATE_AGENT_SPLIT"; +const ENV_LLM_ENDPOINT: &str = "METACRATE_AGENT_LLM_ENDPOINT_URL"; +const ENV_LLM_API_KEY: &str = "METACRATE_AGENT_LLM_API_KEY"; +const ENV_LLM_API_KEY_FILE: &str = "METACRATE_AGENT_LLM_API_KEY_FILE"; +const ENV_GRID_LOGIN_URL: &str = "METACRATE_AGENT_GRID_LOGIN_URL"; +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"; + +/// Wrapper that never reveals its contents through `Debug` or `Display` and +/// deliberately does not implement serialization. +#[derive(Clone, Eq, PartialEq)] +pub struct SecretString(String); + +impl SecretString { + /// Admits one nonempty, single-line secret within the secret-file bound. + /// + /// # Errors + /// + /// Returns a typed configuration error for empty, multiline, NUL-containing, + /// or oversized values. + pub fn new(field: &'static str, value: impl Into) -> Result { + let value = value.into(); + validate_secret(field, &value)?; + Ok(Self(value)) + } + + /// Explicit secret-bearing accessor for authenticated transports. + #[must_use] + pub fn expose_secret(&self) -> &str { + &self.0 + } +} + +impl fmt::Debug for SecretString { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("SecretString([REDACTED])") + } +} + +impl fmt::Display for SecretString { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("[REDACTED]") + } +} + +/// Exact HTTP(S) URL whose diagnostic representation removes user info and query data. +#[derive(Clone, Eq, PartialEq)] +pub struct EndpointUrl(Url); + +impl EndpointUrl { + /// Parses a network endpoint without rewriting its path, host, or query. + /// + /// # Errors + /// + /// Rejects malformed, hostless, non-HTTP(S), or fragment-bearing URLs. + pub fn parse(field: &'static str, value: &str) -> Result { + let parsed = Url::parse(value).map_err(|error| ConfigError::InvalidUrl { + field, + reason: error.to_string(), + })?; + if !matches!(parsed.scheme(), "http" | "https") { + return Err(ConfigError::InvalidUrl { + field, + reason: "scheme must be http or https".into(), + }); + } + if parsed.host_str().is_none() { + return Err(ConfigError::InvalidUrl { + field, + reason: "URL must contain a host".into(), + }); + } + if parsed.fragment().is_some() { + return Err(ConfigError::InvalidUrl { + field, + reason: "URL fragments are not sent over HTTP and are not allowed".into(), + }); + } + Ok(Self(parsed)) + } + + /// Returns the exact configured URL. Callers must not log this value because + /// it can include user information or secret query parameters. + #[must_use] + pub fn expose_url(&self) -> &str { + self.0.as_str() + } + + fn redacted(&self) -> String { + let mut redacted = self.0.clone(); + if !redacted.username().is_empty() { + let _ = redacted.set_username("[REDACTED]"); + } + if redacted.password().is_some() { + let _ = redacted.set_password(Some("[REDACTED]")); + } + if redacted.query().is_some() { + redacted.set_query(Some("[REDACTED]")); + } + redacted.to_string() + } +} + +impl fmt::Debug for EndpointUrl { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_tuple("EndpointUrl") + .field(&self.redacted()) + .finish() + } +} + +impl fmt::Display for EndpointUrl { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(&self.redacted()) + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum OperatingMode { + OfflineFake, + Integrated, + SplitService, +} + +/// The LLM connection identity. These are deliberately the only two fields. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct LlmConnection { + pub endpoint_url: EndpointUrl, + pub api_key: SecretString, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct GridConnection { + pub login_url: EndpointUrl, + pub avatar_name: String, + pub password: SecretString, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Timeouts { + pub startup: Duration, + pub shutdown: Duration, + pub request: Duration, +} + +/// Runtime allocation and concurrency limits. Each value is validated against +/// a compile-time hard ceiling before service startup. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Limits { + pub grid_event_queue: usize, + pub control_queue: usize, + pub observable_queue: usize, + pub max_body_bytes: usize, + pub max_message_bytes: usize, + pub max_conversation_messages: usize, + pub max_tool_calls: usize, + pub max_authorized_avatars: usize, + pub max_background_tasks: usize, +} + +impl Default for Limits { + fn default() -> Self { + Self { + grid_event_queue: 256, + control_queue: 32, + observable_queue: 512, + max_body_bytes: 1024 * 1024, + max_message_bytes: 16 * 1024, + max_conversation_messages: 64, + max_tool_calls: 16, + max_authorized_avatars: 128, + max_background_tasks: MAX_BACKGROUND_TASKS, + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct BehaviorSettings { + pub heartbeat: Duration, +} + +/// Fully resolved configuration. It cannot be constructed without validation. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct AgentConfig { + pub mode: OperatingMode, + pub llm: LlmConnection, + pub grid: Option, + pub authorized_avatar_uuids: BTreeSet, + pub timeouts: Timeouts, + pub limits: Limits, + pub storage_path: PathBuf, + pub behavior: BehaviorSettings, +} + +impl AgentConfig { + /// Creates the smallest valid offline configuration with no network side effects. + /// + /// # Errors + /// + /// Returns a typed error when either required LLM field is invalid. + pub fn offline(endpoint_url: &str, api_key: &str) -> Result { + let raw = FileConfig { + llm: RawLlm { + endpoint_url: Some(endpoint_url.into()), + api_key: Some(api_key.into()), + }, + ..FileConfig::default() + }; + resolve(raw, &MapEnvironment::default(), None) + } + + /// Rechecks all dynamic limits and mode invariants. + /// + /// # Errors + /// + /// Returns the first invalid field without performing I/O. + pub fn validate(&self) -> Result<(), ConfigError> { + validate_limits(&self.limits)?; + if self.mode != OperatingMode::OfflineFake && self.grid.is_none() { + return Err(ConfigError::Missing { + field: "grid", + required_for: "integrated and split live modes", + }); + } + if self.authorized_avatar_uuids.len() > self.limits.max_authorized_avatars { + return Err(ConfigError::UnsafeLimit { + field: "authorized_avatar_uuids", + value: self.authorized_avatar_uuids.len(), + minimum: 0, + maximum: self.limits.max_authorized_avatars, + }); + } + Ok(()) + } +} + +/// Read-only environment boundary, injectable for deterministic precedence tests. +pub trait Environment { + fn get(&self, name: &str) -> Option; +} + +#[derive(Clone, Copy, Debug, Default)] +pub struct StdEnvironment; + +impl Environment for StdEnvironment { + fn get(&self, name: &str) -> Option { + std::env::var(name).ok() + } +} + +#[derive(Clone, Default)] +pub struct MapEnvironment(BTreeMap); + +impl fmt::Debug for MapEnvironment { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("MapEnvironment") + .field("variable_count", &self.0.len()) + .field("values", &"[REDACTED]") + .finish() + } +} + +impl MapEnvironment { + #[must_use] + pub fn from_pairs(pairs: [(&str, &str); N]) -> Self { + Self( + pairs + .into_iter() + .map(|(name, value)| (name.to_owned(), value.to_owned())) + .collect(), + ) + } + + pub fn insert(&mut self, name: impl Into, value: impl Into) { + self.0.insert(name.into(), value.into()); + } +} + +impl Environment for MapEnvironment { + fn get(&self, name: &str) -> Option { + self.0.get(name).cloned() + } +} + +/// Resolves built-in defaults, then an optional JSON file, secret files, and +/// finally environment variables (highest precedence). +#[derive(Clone)] +pub struct ConfigLoader { + file: Option, + environment: E, +} + +impl fmt::Debug for ConfigLoader { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("ConfigLoader") + .field("file", &self.file) + .field("environment", &"[REDACTED SOURCE]") + .finish() + } +} + +impl ConfigLoader { + #[must_use] + pub const fn new() -> Self { + Self { + file: None, + environment: StdEnvironment, + } + } +} + +impl Default for ConfigLoader { + fn default() -> Self { + Self::new() + } +} + +impl ConfigLoader { + #[must_use] + pub fn with_environment(self, environment: T) -> ConfigLoader { + ConfigLoader { + file: self.file, + environment, + } + } + + #[must_use] + pub fn with_file(mut self, path: impl Into) -> Self { + self.file = Some(path.into()); + self + } + + /// Loads and validates configuration before any runtime or network object exists. + /// + /// # Errors + /// + /// Returns typed I/O, schema, missing-field, secret, URL, UUID, mode, or + /// resource-limit errors. + pub fn load(&self) -> Result { + let raw = if let Some(path) = &self.file { + read_config(path)? + } else { + FileConfig::default() + }; + let config = resolve(raw, &self.environment, self.file.as_deref())?; + config.validate()?; + Ok(config) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ConfigError { + Io { + field: &'static str, + path: PathBuf, + reason: String, + }, + InvalidSchema { + path: PathBuf, + reason: String, + }, + Missing { + field: &'static str, + required_for: &'static str, + }, + InvalidUrl { + field: &'static str, + reason: String, + }, + InvalidUuid { + value: String, + reason: &'static str, + }, + InvalidSecret { + field: &'static str, + reason: &'static str, + }, + InvalidBoolean { + field: &'static str, + value: String, + }, + ConflictingModes, + UnsafeLimit { + field: &'static str, + value: usize, + minimum: usize, + maximum: usize, + }, +} + +impl fmt::Display for ConfigError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Io { + field, + path, + reason, + } => write!( + formatter, + "cannot read {field} at {}: {reason}", + path.display() + ), + Self::InvalidSchema { path, reason } => { + write!( + formatter, + "invalid configuration {}: {reason}", + path.display() + ) + } + Self::Missing { + field, + required_for, + } => write!(formatter, "missing {field}; required for {required_for}"), + Self::InvalidUrl { field, reason } => write!(formatter, "invalid {field}: {reason}"), + Self::InvalidUuid { value, reason } => { + write!( + formatter, + "invalid authorized avatar UUID {value:?}: {reason}" + ) + } + Self::InvalidSecret { field, reason } => { + write!(formatter, "invalid secret {field}: {reason}") + } + Self::InvalidBoolean { field, value } => { + write!(formatter, "{field} must be true or false, got {value:?}") + } + Self::ConflictingModes => formatter + .write_str("integrated and split modes are mutually exclusive; enable at most one"), + Self::UnsafeLimit { + field, + value, + minimum, + maximum, + } => write!( + formatter, + "unsafe {field}={value}; expected {minimum}..={maximum}" + ), + } + } +} + +impl Error for ConfigError {} + +#[derive(Clone, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +struct FileConfig { + integrated: Option, + split: Option, + llm: RawLlm, + grid: RawGrid, + secret_files: RawSecretFiles, + authorized_avatar_uuids: Option>, + timeouts: RawTimeouts, + limits: RawLimits, + storage_path: Option, + behavior: RawBehavior, +} + +#[derive(Clone, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +struct RawLlm { + endpoint_url: Option, + api_key: Option, +} + +#[derive(Clone, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +struct RawGrid { + login_url: Option, + avatar_name: Option, + password: Option, +} + +#[derive(Clone, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +struct RawSecretFiles { + llm_api_key: Option, + grid_password: Option, +} + +#[derive(Clone, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +struct RawTimeouts { + #[serde(rename = "startup_seconds")] + startup: Option, + #[serde(rename = "shutdown_seconds")] + shutdown: Option, + #[serde(rename = "request_seconds")] + request: Option, +} + +#[derive(Clone, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +struct RawLimits { + grid_event_queue: Option, + control_queue: Option, + observable_queue: Option, + max_body_bytes: Option, + max_message_bytes: Option, + max_conversation_messages: Option, + max_tool_calls: Option, + max_authorized_avatars: Option, + max_background_tasks: Option, +} + +#[derive(Clone, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +struct RawBehavior { + heartbeat_seconds: Option, +} + +fn read_config(path: &Path) -> Result { + let bytes = read_bounded_regular_file("configuration file", path, MAX_CONFIG_BYTES)?; + serde_json::from_slice(&bytes).map_err(|error| ConfigError::InvalidSchema { + path: path.to_owned(), + reason: error.to_string(), + }) +} + +#[allow(clippy::too_many_lines)] // One linear resolver makes the complete precedence order auditable. +fn resolve( + raw: FileConfig, + environment: &E, + config_path: Option<&Path>, +) -> Result { + let integrated = environment_boolean(environment, ENV_INTEGRATED)? + .or(raw.integrated) + .unwrap_or(false); + let split = environment_boolean(environment, ENV_SPLIT)? + .or(raw.split) + .unwrap_or(false); + let mode = match (integrated, split) { + (true, true) => return Err(ConfigError::ConflictingModes), + (true, false) => OperatingMode::Integrated, + (false, true) => OperatingMode::SplitService, + (false, false) => OperatingMode::OfflineFake, + }; + + let base = config_path + .and_then(Path::parent) + .unwrap_or_else(|| Path::new(".")); + let endpoint = environment + .get(ENV_LLM_ENDPOINT) + .or(raw.llm.endpoint_url) + .ok_or(ConfigError::Missing { + field: "llm.endpoint_url", + required_for: "all modes", + })?; + let api_key = secret_from_layers( + environment.get(ENV_LLM_API_KEY), + environment.get(ENV_LLM_API_KEY_FILE).map(PathBuf::from), + raw.llm.api_key, + raw.secret_files.llm_api_key, + base, + "llm.api_key", + )?; + let llm = LlmConnection { + endpoint_url: EndpointUrl::parse("llm.endpoint_url", &endpoint)?, + api_key, + }; + + let login_url = environment.get(ENV_GRID_LOGIN_URL).or(raw.grid.login_url); + let avatar_name = environment + .get(ENV_GRID_AVATAR_NAME) + .or(raw.grid.avatar_name); + let direct_password = environment.get(ENV_GRID_PASSWORD).or(raw.grid.password); + let environment_password_file = environment.get(ENV_GRID_PASSWORD_FILE).map(PathBuf::from); + let grid = if mode == OperatingMode::OfflineFake + && login_url.is_none() + && avatar_name.is_none() + && direct_password.is_none() + && environment_password_file.is_none() + && raw.secret_files.grid_password.is_none() + { + None + } else { + let login_url = login_url.ok_or(ConfigError::Missing { + field: "grid.login_url", + required_for: "configured grid connections", + })?; + let avatar_name = avatar_name.ok_or(ConfigError::Missing { + field: "grid.avatar_name", + required_for: "configured grid connections", + })?; + if avatar_name.trim().is_empty() || avatar_name.len() > 256 { + return Err(ConfigError::InvalidSecret { + field: "grid.avatar_name", + reason: "must contain 1..=256 bytes", + }); + } + Some(GridConnection { + login_url: EndpointUrl::parse("grid.login_url", &login_url)?, + avatar_name, + password: secret_from_layers( + direct_password, + environment_password_file, + None, + raw.secret_files.grid_password, + base, + "grid.password", + )?, + }) + }; + + let authorized_values = environment + .get(ENV_AUTHORIZED_AVATARS) + .map(|value| value.split(',').map(str::trim).map(str::to_owned).collect()) + .or(raw.authorized_avatar_uuids) + .unwrap_or_default(); + let authorized_avatar_uuids = parse_authorized_avatars(authorized_values)?; + + let defaults = Limits::default(); + let limits = Limits { + grid_event_queue: raw + .limits + .grid_event_queue + .unwrap_or(defaults.grid_event_queue), + control_queue: raw.limits.control_queue.unwrap_or(defaults.control_queue), + observable_queue: raw + .limits + .observable_queue + .unwrap_or(defaults.observable_queue), + max_body_bytes: raw.limits.max_body_bytes.unwrap_or(defaults.max_body_bytes), + max_message_bytes: raw + .limits + .max_message_bytes + .unwrap_or(defaults.max_message_bytes), + max_conversation_messages: raw + .limits + .max_conversation_messages + .unwrap_or(defaults.max_conversation_messages), + max_tool_calls: raw.limits.max_tool_calls.unwrap_or(defaults.max_tool_calls), + max_authorized_avatars: raw + .limits + .max_authorized_avatars + .unwrap_or(defaults.max_authorized_avatars), + max_background_tasks: raw + .limits + .max_background_tasks + .unwrap_or(defaults.max_background_tasks), + }; + validate_limits(&limits)?; + + let config = AgentConfig { + mode, + llm, + grid, + authorized_avatar_uuids, + timeouts: Timeouts { + startup: checked_duration( + "timeouts.startup_seconds", + raw.timeouts.startup.unwrap_or(30), + 1, + 300, + )?, + shutdown: checked_duration( + "timeouts.shutdown_seconds", + raw.timeouts.shutdown.unwrap_or(10), + 1, + 60, + )?, + request: checked_duration( + "timeouts.request_seconds", + raw.timeouts.request.unwrap_or(60), + 1, + 300, + )?, + }, + limits, + storage_path: raw + .storage_path + .unwrap_or_else(|| PathBuf::from("data/grid-agent")), + behavior: BehaviorSettings { + heartbeat: checked_duration( + "behavior.heartbeat_seconds", + raw.behavior.heartbeat_seconds.unwrap_or(30), + 1, + 300, + )?, + }, + }; + config.validate()?; + Ok(config) +} + +fn parse_authorized_avatars(values: Vec) -> Result, ConfigError> { + if values.len() > MAX_AUTHORIZED_AVATARS { + return Err(ConfigError::UnsafeLimit { + field: "authorized_avatar_uuids", + value: values.len(), + minimum: 0, + maximum: MAX_AUTHORIZED_AVATARS, + }); + } + let mut parsed = BTreeSet::new(); + for value in values { + if value.trim() == "*" { + return Err(ConfigError::InvalidUuid { + value, + reason: "wildcard authorization is forbidden", + }); + } + let id = UUID::new_with_string(value.clone()).map_err(|_| ConfigError::InvalidUuid { + value: value.clone(), + reason: "expected a canonical UUID", + })?; + if id == UUID::zero() || id.to_string() != value.to_ascii_lowercase() { + return Err(ConfigError::InvalidUuid { + value, + reason: "expected a nonzero canonical lowercase hyphenated UUID", + }); + } + if !parsed.insert(id) { + return Err(ConfigError::InvalidUuid { + value, + reason: "duplicate authorization entries are forbidden", + }); + } + } + Ok(parsed) +} + +fn secret_from_layers( + environment_value: Option, + environment_file: Option, + file_value: Option, + file_path: Option, + base: &Path, + field: &'static str, +) -> Result { + if let Some(value) = environment_value { + return SecretString::new(field, value); + } + if let Some(path) = environment_file { + return read_secret(field, &resolve_path(base, &path)); + } + if let Some(path) = file_path { + return read_secret(field, &resolve_path(base, &path)); + } + if let Some(value) = file_value { + return SecretString::new(field, value); + } + Err(ConfigError::Missing { + field, + required_for: "the selected connection", + }) +} + +fn resolve_path(base: &Path, path: &Path) -> PathBuf { + if path.is_absolute() { + path.to_owned() + } else { + base.join(path) + } +} + +fn read_secret(field: &'static str, path: &Path) -> Result { + let bytes = read_bounded_regular_file(field, path, MAX_SECRET_BYTES)?; + let value = String::from_utf8(bytes).map_err(|_| ConfigError::InvalidSecret { + field, + reason: "secret file must be UTF-8", + })?; + let value = value + .strip_suffix("\r\n") + .or_else(|| value.strip_suffix('\n')) + .unwrap_or(&value) + .to_owned(); + SecretString::new(field, value) +} + +fn read_bounded_regular_file( + field: &'static str, + path: &Path, + maximum: u64, +) -> Result, ConfigError> { + let metadata = fs::symlink_metadata(path).map_err(|error| ConfigError::Io { + field, + path: path.to_owned(), + reason: error.to_string(), + })?; + if !metadata.file_type().is_file() || metadata.file_type().is_symlink() { + return Err(ConfigError::Io { + field, + path: path.to_owned(), + reason: "must be a regular non-symlink file".into(), + }); + } + if metadata.len() > maximum { + return Err(ConfigError::Io { + field, + path: path.to_owned(), + reason: format!("file is {} bytes; maximum is {maximum}", metadata.len()), + }); + } + fs::read(path).map_err(|error| ConfigError::Io { + field, + path: path.to_owned(), + reason: error.to_string(), + }) +} + +fn validate_secret(field: &'static str, value: &str) -> Result<(), ConfigError> { + if value.is_empty() { + return Err(ConfigError::InvalidSecret { + field, + reason: "must not be empty", + }); + } + if value.len() > usize::try_from(MAX_SECRET_BYTES).unwrap_or(usize::MAX) { + return Err(ConfigError::InvalidSecret { + field, + reason: "exceeds the 16 KiB secret bound", + }); + } + if value.contains(['\n', '\r', '\0']) { + return Err(ConfigError::InvalidSecret { + field, + reason: "must be a single line without NUL bytes", + }); + } + Ok(()) +} + +fn environment_boolean( + environment: &E, + field: &'static str, +) -> Result, ConfigError> { + environment + .get(field) + .map(|value| match value.to_ascii_lowercase().as_str() { + "true" | "1" => Ok(true), + "false" | "0" => Ok(false), + _ => Err(ConfigError::InvalidBoolean { field, value }), + }) + .transpose() +} + +fn checked_duration( + field: &'static str, + seconds: u64, + minimum: usize, + maximum: usize, +) -> Result { + let value = usize::try_from(seconds).unwrap_or(usize::MAX); + check_limit(field, value, minimum, maximum)?; + Ok(Duration::from_secs(seconds)) +} + +fn validate_limits(limits: &Limits) -> Result<(), ConfigError> { + check_limit( + "limits.grid_event_queue", + limits.grid_event_queue, + 1, + MAX_QUEUE_CAPACITY, + )?; + check_limit("limits.control_queue", limits.control_queue, 1, 256)?; + check_limit( + "limits.observable_queue", + limits.observable_queue, + 1, + MAX_QUEUE_CAPACITY, + )?; + check_limit( + "limits.max_body_bytes", + limits.max_body_bytes, + 1_024, + MAX_BODY_BYTES, + )?; + check_limit( + "limits.max_message_bytes", + limits.max_message_bytes, + 1, + MAX_MESSAGE_BYTES, + )?; + check_limit( + "limits.max_conversation_messages", + limits.max_conversation_messages, + 1, + MAX_CONVERSATION_MESSAGES, + )?; + check_limit( + "limits.max_tool_calls", + limits.max_tool_calls, + 1, + MAX_TOOL_CALLS, + )?; + check_limit( + "limits.max_authorized_avatars", + limits.max_authorized_avatars, + 1, + MAX_AUTHORIZED_AVATARS, + )?; + check_limit( + "limits.max_background_tasks", + limits.max_background_tasks, + MAX_BACKGROUND_TASKS, + MAX_BACKGROUND_TASKS, + ) +} + +fn check_limit( + field: &'static str, + value: usize, + minimum: usize, + maximum: usize, +) -> Result<(), ConfigError> { + if (minimum..=maximum).contains(&value) { + Ok(()) + } else { + Err(ConfigError::UnsafeLimit { + field, + value, + minimum, + maximum, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write as _; + use std::sync::atomic::{AtomicU64, Ordering}; + + static NEXT_TEMPORARY: AtomicU64 = AtomicU64::new(1); + + fn offline_environment() -> MapEnvironment { + MapEnvironment::from_pairs([ + ( + ENV_LLM_ENDPOINT, + "https://llm.example.invalid/v1/chat/completions", + ), + (ENV_LLM_API_KEY, "environment-key"), + ]) + } + + fn temporary_file(name: &str, contents: &str) -> PathBuf { + let directory = std::env::temp_dir().join(format!( + "metacrate-grid-agent-{}-{}", + NEXT_TEMPORARY.fetch_add(1, Ordering::Relaxed), + name + )); + fs::create_dir_all(&directory).expect("create test directory"); + let path = directory.join(name); + let mut file = fs::File::create(&path).expect("create test file"); + file.write_all(contents.as_bytes()) + .expect("write test file"); + path + } + + #[test] + fn environment_overrides_file_and_secrets_are_redacted() { + let path = temporary_file( + "precedence.json", + r#"{ + "llm": {"endpoint_url":"https://file-user:file-pass@file.invalid/chat?api_key=file-query", "api_key":"file-key"} + }"#, + ); + let loader = ConfigLoader::new() + .with_file(&path) + .with_environment(offline_environment()); + let loader_diagnostic = format!("{loader:?}"); + assert!(!loader_diagnostic.contains("environment-key")); + let config = loader.load().expect("valid layered configuration"); + assert_eq!(config.llm.api_key.expose_secret(), "environment-key"); + assert!( + config + .llm + .endpoint_url + .expose_url() + .contains("llm.example.invalid") + ); + let diagnostic = format!("{config:?}"); + assert!(!diagnostic.contains("environment-key")); + assert!(!diagnostic.contains("file-key")); + let _ = fs::remove_file(path); + } + + #[test] + fn secret_file_overrides_inline_secret_and_direct_environment_wins_last() { + let secret = temporary_file("precedence-api-key", "secret-file-key\n"); + let document = serde_json::json!({ + "llm": { + "endpoint_url": "https://file.invalid/chat", + "api_key": "inline-key" + }, + "secret_files": {"llm_api_key": secret.clone()} + }); + let config_path = temporary_file("all-layers.json", &document.to_string()); + let file_environment = + MapEnvironment::from_pairs([(ENV_LLM_ENDPOINT, "https://environment.invalid/chat")]); + let config = ConfigLoader::new() + .with_file(&config_path) + .with_environment(file_environment) + .load() + .expect("secret file layer resolves"); + assert_eq!(config.llm.api_key.expose_secret(), "secret-file-key"); + + let direct_environment = MapEnvironment::from_pairs([ + (ENV_LLM_ENDPOINT, "https://environment.invalid/chat"), + (ENV_LLM_API_KEY, "direct-environment-key"), + ]); + let config = ConfigLoader::new() + .with_file(&config_path) + .with_environment(direct_environment) + .load() + .expect("direct environment layer resolves"); + assert_eq!(config.llm.api_key.expose_secret(), "direct-environment-key"); + let _ = fs::remove_file(secret); + let _ = fs::remove_file(config_path); + } + + #[test] + fn endpoint_debug_redacts_credentials_and_query() { + let endpoint = EndpointUrl::parse( + "llm.endpoint_url", + "https://alice:password@example.invalid/chat?api_key=query-secret", + ) + .expect("valid HTTPS endpoint"); + let diagnostic = format!("{endpoint:?}"); + assert!(!diagnostic.contains("alice")); + assert!(!diagnostic.contains("password")); + assert!(!diagnostic.contains("query-secret")); + assert!(diagnostic.contains("REDACTED")); + } + + #[test] + fn invalid_urls_uuids_limits_and_wildcards_fail_fast() { + assert!(AgentConfig::offline("file:///tmp/socket", "key").is_err()); + + let mut wildcard = offline_environment(); + wildcard.insert(ENV_AUTHORIZED_AVATARS, "*"); + assert!(matches!( + ConfigLoader::new().with_environment(wildcard).load(), + Err(ConfigError::InvalidUuid { .. }) + )); + + let mut malformed = offline_environment(); + malformed.insert(ENV_AUTHORIZED_AVATARS, "not-a-uuid"); + assert!(matches!( + ConfigLoader::new().with_environment(malformed).load(), + Err(ConfigError::InvalidUuid { .. }) + )); + + let path = temporary_file("unsafe-limit.json", r#"{"limits":{"control_queue":0}}"#); + assert!(matches!( + ConfigLoader::new() + .with_file(&path) + .with_environment(offline_environment()) + .load(), + Err(ConfigError::UnsafeLimit { .. }) + )); + let _ = fs::remove_file(path); + } + + #[test] + fn conflicting_live_modes_and_missing_live_fields_are_rejected() { + let both = MapEnvironment::from_pairs([ + (ENV_INTEGRATED, "true"), + (ENV_SPLIT, "true"), + (ENV_LLM_ENDPOINT, "https://llm.invalid/chat"), + (ENV_LLM_API_KEY, "key"), + ]); + assert_eq!( + ConfigLoader::new().with_environment(both).load(), + Err(ConfigError::ConflictingModes) + ); + + let missing_grid = MapEnvironment::from_pairs([ + (ENV_INTEGRATED, "true"), + (ENV_LLM_ENDPOINT, "https://llm.invalid/chat"), + (ENV_LLM_API_KEY, "key"), + ]); + assert!(matches!( + ConfigLoader::new().with_environment(missing_grid).load(), + Err(ConfigError::Missing { + field: "grid.login_url", + .. + }) + )); + } + + #[test] + fn secret_file_is_bounded_regular_and_trailing_newline_is_removed() { + let secret = temporary_file("api-key", "secret-from-file\n"); + let mut environment = + MapEnvironment::from_pairs([(ENV_LLM_ENDPOINT, "https://llm.example.invalid/chat")]); + environment.insert(ENV_LLM_API_KEY_FILE, secret.to_string_lossy()); + let config = ConfigLoader::new() + .with_environment(environment) + .load() + .expect("secret file loads"); + assert_eq!(config.llm.api_key.expose_secret(), "secret-from-file"); + let _ = fs::remove_file(secret); + } +} diff --git a/crates/metacrate-grid-agent/src/lib.rs b/crates/metacrate-grid-agent/src/lib.rs new file mode 100644 index 0000000..b7bac5d --- /dev/null +++ b/crates/metacrate-grid-agent/src/lib.rs @@ -0,0 +1,25 @@ +//! Bounded service foundation for the native `MetaCrate` `OpenSim` grid agent. +//! +//! Configuration is resolved and validated before [`AgentService::start`] +//! creates tasks or calls a backend. The core has no signal, terminal, path, +//! subprocess, provider-SDK, or platform-specific dependency. + +pub mod backend; +pub mod config; +pub mod service; +pub mod types; + +#[cfg(feature = "live-grid")] +pub use backend::LibremetaverseClientOwner; +pub use backend::{BackendError, BackendFuture, GridBackend, OfflineGridBackend, WorldMutator}; +pub use config::{ + AgentConfig, BehaviorSettings, ConfigError, ConfigLoader, EndpointUrl, Environment, + GridConnection, Limits, LlmConnection, MapEnvironment, OperatingMode, SecretString, + StdEnvironment, Timeouts, +}; +pub use service::{AgentService, ServiceError, ServiceHandle, ServiceState}; +pub use types::{ + BoundaryError, BoundedText, BoundedVec, ControlCommand, Conversation, ConversationMessage, + GridEvent, GridEventKind, LlmRequest, LlmResult, MessageRole, ObservableEvent, PolicyDecision, + ProposedToolCall, ToolCallOutcome, +}; diff --git a/crates/metacrate-grid-agent/src/main.rs b/crates/metacrate-grid-agent/src/main.rs new file mode 100644 index 0000000..b02251a --- /dev/null +++ b/crates/metacrate-grid-agent/src/main.rs @@ -0,0 +1,135 @@ +use metacrate_grid_agent::{ + AgentService, ConfigLoader, GridEventKind, ObservableEvent, OperatingMode, +}; +use std::error::Error; +use std::fmt; +use std::path::PathBuf; + +const MAX_ARGUMENTS: usize = 8; + +#[derive(Debug)] +struct CliError(String); + +impl fmt::Display for CliError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(&self.0) + } +} + +impl Error for CliError {} + +#[derive(Default)] +struct Options { + config: Option, + check_config: bool, + run_once: bool, +} + +fn options() -> Result, CliError> { + let mut result = Options::default(); + let mut arguments = std::env::args_os().skip(1); + let mut count = 0; + while let Some(argument) = arguments.next() { + count += 1; + if count > MAX_ARGUMENTS { + return Err(CliError(format!( + "at most {MAX_ARGUMENTS} command-line arguments are accepted" + ))); + } + if argument == "--help" || argument == "-h" { + println!( + "metacrate-grid-agent [--config PATH] [--check-config | --run-once]\n\ + Configuration precedence: defaults < JSON < secret files < environment." + ); + return Ok(None); + } + if argument == "--check-config" { + result.check_config = true; + } else if argument == "--run-once" { + result.run_once = true; + } else if argument == "--config" { + let path = arguments + .next() + .ok_or_else(|| CliError("--config requires a path".into()))?; + count += 1; + result.config = Some(PathBuf::from(path)); + } else { + let argument = PathBuf::from(argument); + return Err(CliError(format!( + "unknown argument {}; use --help", + argument.display() + ))); + } + } + if result.check_config && result.run_once { + return Err(CliError( + "--check-config and --run-once are mutually exclusive".into(), + )); + } + Ok(Some(result)) +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let Some(options) = options()? else { + return Ok(()); + }; + let mut loader = ConfigLoader::new(); + if let Some(path) = options.config { + loader = loader.with_file(path); + } + let config = loader.load()?; + if options.check_config { + println!("configuration is valid for {:?} mode", config.mode); + return Ok(()); + } + if config.mode != OperatingMode::OfflineFake { + return Err(CliError( + "this architecture issue starts only the offline backend; live login is owned by a later milestone issue" + .into(), + ) + .into()); + } + + let startup_timeout = config.timeouts.startup; + let mut handle = AgentService::offline(config)?.start()?; + if options.run_once { + tokio::time::timeout(startup_timeout, async { + loop { + match handle.next_event().await { + Some(ObservableEvent::Grid(event)) + if event.kind == GridEventKind::BackendReady => + { + return Ok::<(), CliError>(()); + } + Some(_) => {} + None => { + return Err(CliError("service stopped before backend readiness".into())); + } + } + } + }) + .await + .map_err(|_| CliError("timed out waiting for offline backend readiness".into()))??; + handle.shutdown().await?; + println!("grid agent completed one offline startup/shutdown cycle"); + return Ok(()); + } + println!("grid agent started in offline/fake mode; press Ctrl-C to stop"); + loop { + tokio::select! { + signal = tokio::signal::ctrl_c() => { + signal?; + break; + } + event = handle.next_event() => { + if event.is_none() { + break; + } + } + } + } + handle.shutdown().await?; + println!("grid agent stopped cleanly"); + Ok(()) +} diff --git a/crates/metacrate-grid-agent/src/service.rs b/crates/metacrate-grid-agent/src/service.rs new file mode 100644 index 0000000..7b2fc66 --- /dev/null +++ b/crates/metacrate-grid-agent/src/service.rs @@ -0,0 +1,513 @@ +//! Two-task, cancellation-safe orchestration skeleton. + +use crate::backend::{BackendError, GridBackend, OfflineGridBackend}; +use crate::config::{AgentConfig, ConfigError, OperatingMode}; +use crate::types::{ + BoundedText, ControlCommand, GridEventKind, MAX_OBSERVABLE_DETAIL_BYTES, ObservableEvent, +}; +use libremetaverse_types::compat::CancellationTokenSource; +use std::error::Error; +use std::fmt; +use std::sync::Arc; +use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering}; +use std::time::Duration; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; + +const OWNED_TASKS: usize = 2; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[repr(u8)] +pub enum ServiceState { + Starting = 0, + Running = 1, + Paused = 2, + Stopping = 3, + Stopped = 4, + Failed = 5, +} + +impl ServiceState { + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Starting => "starting", + Self::Running => "running", + Self::Paused => "paused", + Self::Stopping => "stopping", + Self::Stopped => "stopped", + Self::Failed => "failed", + } + } + + fn from_atomic(value: u8) -> Self { + match value { + 0 => Self::Starting, + 1 => Self::Running, + 2 => Self::Paused, + 3 => Self::Stopping, + 4 => Self::Stopped, + _ => Self::Failed, + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ServiceError { + Configuration(ConfigError), + Backend(BackendError), + ControlQueueClosed, + ObservableQueueClosed, + TaskPanicked, + ShutdownTimedOut { task: &'static str }, +} + +impl fmt::Display for ServiceError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Configuration(error) => write!(formatter, "configuration rejected: {error}"), + Self::Backend(error) => write!(formatter, "grid backend failed: {error}"), + Self::ControlQueueClosed => formatter.write_str("control queue is closed"), + Self::ObservableQueueClosed => formatter.write_str("observable queue is closed"), + Self::TaskPanicked => formatter.write_str("owned task panicked or was cancelled"), + Self::ShutdownTimedOut { task } => { + write!(formatter, "timed out while joining owned {task} task") + } + } + } +} + +impl Error for ServiceError {} + +impl From for ServiceError { + fn from(value: ConfigError) -> Self { + Self::Configuration(value) + } +} + +impl From for ServiceError { + fn from(value: BackendError) -> Self { + Self::Backend(value) + } +} + +/// Validated service composition. Construction and validation are inert; +/// `start` is the only task-creation point. +pub struct AgentService { + config: AgentConfig, + backend: Arc, +} + +impl AgentService { + #[must_use] + pub fn new(config: AgentConfig, backend: Arc) -> Self { + Self { config, backend } + } + + /// Composes the deterministic offline backend. + /// + /// # Errors + /// + /// Rejects non-offline configuration. + pub fn offline(config: AgentConfig) -> Result { + if config.mode != OperatingMode::OfflineFake { + return Err(ConfigError::Missing { + field: "offline mode", + required_for: "the offline backend", + } + .into()); + } + Ok(Self::new(config, Arc::new(OfflineGridBackend::new()))) + } + + /// Validates all invariants, creates three bounded queues, and starts the + /// exactly two configured owner tasks. + /// + /// # Errors + /// + /// Returns before task creation for invalid configuration. + pub fn start(self) -> Result { + self.config.validate()?; + if self.config.limits.max_background_tasks != OWNED_TASKS { + return Err(ConfigError::UnsafeLimit { + field: "limits.max_background_tasks", + value: self.config.limits.max_background_tasks, + minimum: OWNED_TASKS, + maximum: OWNED_TASKS, + } + .into()); + } + + let (grid_sender, grid_receiver) = mpsc::channel(self.config.limits.grid_event_queue); + let (control_sender, control_receiver) = mpsc::channel(self.config.limits.control_queue); + let (observable_sender, observable_receiver) = + mpsc::channel(self.config.limits.observable_queue); + let cancellation = CancellationTokenSource::new(); + let state = Arc::new(AtomicU8::new(ServiceState::Starting as u8)); + let active_tasks = Arc::new(AtomicUsize::new(0)); + + let backend = Arc::clone(&self.backend); + let backend_cancellation = cancellation.clone(); + let backend_observable = observable_sender.clone(); + let backend_guard = TaskCountGuard::new(Arc::clone(&active_tasks)); + let backend_state = Arc::clone(&state); + let backend_task = tokio::spawn(async move { + let _task_guard = backend_guard; + let result = backend.run(grid_sender, backend_cancellation.token()).await; + if let Err(error) = &result { + if let Ok(diagnostic) = BoundedText::::new( + "observable.backend_failure", + "grid backend stopped with a typed error", + ) { + let _ = send_observable( + &backend_observable, + ObservableEvent::Diagnostic { detail: diagnostic }, + &backend_cancellation, + ) + .await; + } + backend_state.store(ServiceState::Failed as u8, Ordering::Release); + backend_cancellation.cancel(); + return Err(ServiceError::Backend(error.clone())); + } + Ok(()) + }); + + let coordinator_cancellation = cancellation.clone(); + let coordinator_state = Arc::clone(&state); + let coordinator_guard = TaskCountGuard::new(Arc::clone(&active_tasks)); + let coordinator_task = tokio::spawn(async move { + let _task_guard = coordinator_guard; + coordinator_loop( + grid_receiver, + control_receiver, + observable_sender, + coordinator_cancellation, + coordinator_state, + ) + .await + }); + + Ok(ServiceHandle { + control_sender, + observable_receiver, + cancellation, + tasks: [Some(backend_task), Some(coordinator_task)], + state, + active_tasks, + shutdown_timeout: self.config.timeouts.shutdown, + }) + } +} + +/// Exclusive owner of service controls, observations, cancellation, and both tasks. +pub struct ServiceHandle { + control_sender: mpsc::Sender, + observable_receiver: mpsc::Receiver, + cancellation: CancellationTokenSource, + tasks: [Option>>; OWNED_TASKS], + state: Arc, + active_tasks: Arc, + shutdown_timeout: Duration, +} + +impl ServiceHandle { + #[must_use] + pub fn state(&self) -> ServiceState { + ServiceState::from_atomic(self.state.load(Ordering::Acquire)) + } + + #[must_use] + pub fn active_task_count(&self) -> usize { + self.active_tasks.load(Ordering::Acquire) + } + + /// Applies backpressure at the bounded coordinator-owned control queue. + /// + /// # Errors + /// + /// Returns when the service has stopped accepting control commands. + pub async fn command(&self, command: ControlCommand) -> Result<(), ServiceError> { + self.control_sender + .send(command) + .await + .map_err(|_| ServiceError::ControlQueueClosed) + } + + /// Receives the next event from the handle-owned bounded observation queue. + pub async fn next_event(&mut self) -> Option { + self.observable_receiver.recv().await + } + + /// Requests cancellation and joins backend first, then coordinator. If a + /// task exceeds its bound it is aborted and still awaited before return. + /// + /// # Errors + /// + /// Returns the first backend, panic, or timeout error after both task slots + /// have been reclaimed. + pub async fn shutdown(&mut self) -> Result<(), ServiceError> { + self.state + .store(ServiceState::Stopping as u8, Ordering::Release); + self.cancellation.cancel(); + let names = ["backend", "coordinator"]; + let mut first_error = None; + for (slot, name) in self.tasks.iter_mut().zip(names) { + let Some(task) = slot.as_mut() else { + continue; + }; + let task_error = match tokio::time::timeout(self.shutdown_timeout, &mut *task).await { + Ok(Ok(Ok(()))) => None, + Ok(Ok(Err(error))) => Some(error), + Ok(Err(_)) => Some(ServiceError::TaskPanicked), + Err(_) => { + task.abort(); + let _ = (&mut *task).await; + Some(ServiceError::ShutdownTimedOut { task: name }) + } + }; + *slot = None; + if first_error.is_none() { + first_error = task_error; + } + } + self.state.store( + if first_error.is_some() { + ServiceState::Failed as u8 + } else { + ServiceState::Stopped as u8 + }, + Ordering::Release, + ); + first_error.map_or(Ok(()), Err) + } +} + +impl Drop for ServiceHandle { + fn drop(&mut self) { + self.cancellation.cancel(); + for task in self.tasks.iter().flatten() { + task.abort(); + } + } +} + +struct TaskCountGuard(Arc); + +impl TaskCountGuard { + fn new(counter: Arc) -> Self { + counter.fetch_add(1, Ordering::AcqRel); + Self(counter) + } +} + +impl Drop for TaskCountGuard { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::AcqRel); + } +} + +async fn coordinator_loop( + mut grid_events: mpsc::Receiver, + mut controls: mpsc::Receiver, + observable: mpsc::Sender, + cancellation: CancellationTokenSource, + state: Arc, +) -> Result<(), ServiceError> { + send_state(&observable, ServiceState::Starting, &cancellation).await?; + loop { + tokio::select! { + () = cancellation.token().cancelled() => break, + control = controls.recv() => match control { + Some(ControlCommand::Pause) => { + state.store(ServiceState::Paused as u8, Ordering::Release); + send_state(&observable, ServiceState::Paused, &cancellation).await?; + } + Some(ControlCommand::Resume | ControlCommand::ReloadBehavior) => { + state.store(ServiceState::Running as u8, Ordering::Release); + send_state(&observable, ServiceState::Running, &cancellation).await?; + } + Some(ControlCommand::Shutdown) | None => { + cancellation.cancel(); + break; + } + }, + event = grid_events.recv() => if let Some(event) = event { + if event.kind == GridEventKind::BackendReady { + state.store(ServiceState::Running as u8, Ordering::Release); + send_state(&observable, ServiceState::Running, &cancellation).await?; + } + send_observable(&observable, ObservableEvent::Grid(event), &cancellation).await?; + } else { + cancellation.cancel(); + break; + } + } + } + if ServiceState::from_atomic(state.load(Ordering::Acquire)) != ServiceState::Failed { + state.store(ServiceState::Stopped as u8, Ordering::Release); + } + Ok(()) +} + +async fn send_state( + observable: &mpsc::Sender, + state: ServiceState, + cancellation: &CancellationTokenSource, +) -> Result<(), ServiceError> { + send_observable( + observable, + ObservableEvent::StateChanged { + state: state.as_str(), + }, + cancellation, + ) + .await +} + +async fn send_observable( + observable: &mpsc::Sender, + event: ObservableEvent, + cancellation: &CancellationTokenSource, +) -> Result<(), ServiceError> { + tokio::select! { + () = cancellation.token().cancelled() => Ok(()), + result = observable.send(event) => result.map_err(|_| ServiceError::ObservableQueueClosed), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::backend::BackendFuture; + use crate::config::AgentConfig; + use crate::types::GridEvent; + use libremetaverse_types::compat::CancellationToken; + + fn offline_config() -> AgentConfig { + AgentConfig::offline("https://llm.example.invalid/chat", "test-key") + .expect("valid offline config") + } + + #[tokio::test] + async fn offline_skeleton_starts_and_shuts_down_without_orphan_tasks() { + let mut handle = AgentService::offline(offline_config()) + .expect("offline composition") + .start() + .expect("service starts"); + assert_eq!(handle.active_task_count(), OWNED_TASKS); + + let mut saw_ready = false; + for _ in 0..3 { + let event = tokio::time::timeout(Duration::from_secs(1), handle.next_event()) + .await + .expect("observable event arrives") + .expect("observable queue remains open"); + if matches!( + event, + ObservableEvent::Grid(crate::types::GridEvent { + kind: GridEventKind::BackendReady, + .. + }) + ) { + saw_ready = true; + break; + } + } + assert!(saw_ready); + handle.shutdown().await.expect("ordered shutdown"); + assert_eq!(handle.state(), ServiceState::Stopped); + assert_eq!(handle.active_task_count(), 0); + } + + #[tokio::test] + async fn pause_resume_and_shutdown_controls_are_bounded_and_observable() { + let mut handle = AgentService::offline(offline_config()) + .expect("offline composition") + .start() + .expect("service starts"); + handle + .command(ControlCommand::Pause) + .await + .expect("pause queued"); + let mut paused = false; + for _ in 0..5 { + if matches!( + handle.next_event().await, + Some(ObservableEvent::StateChanged { state: "paused" }) + ) { + paused = true; + break; + } + } + assert!(paused); + handle + .command(ControlCommand::Resume) + .await + .expect("resume queued"); + handle.shutdown().await.expect("clean shutdown"); + } + + #[tokio::test] + async fn dropping_handle_cancels_and_aborts_owned_tasks() { + let handle = AgentService::offline(offline_config()) + .expect("offline composition") + .start() + .expect("service starts"); + let counter = Arc::clone(&handle.active_tasks); + drop(handle); + tokio::time::timeout(Duration::from_secs(1), async { + while counter.load(Ordering::Acquire) != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("aborted task guards complete"); + } + + struct StubbornBackend; + + impl GridBackend for StubbornBackend { + fn name(&self) -> &'static str { + "stubborn-test-backend" + } + + fn run( + &self, + _events: mpsc::Sender, + _cancellation: CancellationToken, + ) -> BackendFuture<'_, Result<(), BackendError>> { + Box::pin(async { + std::future::pending::<()>().await; + Ok(()) + }) + } + } + + #[tokio::test] + async fn cancelled_shutdown_future_keeps_join_handles_owned() { + let mut handle = AgentService::new(offline_config(), Arc::new(StubbornBackend)) + .start() + .expect("service starts"); + let counter = Arc::clone(&handle.active_tasks); + let mut shutdown = Box::pin(handle.shutdown()); + tokio::select! { + result = &mut shutdown => panic!("stubborn backend unexpectedly stopped: {result:?}"), + () = tokio::task::yield_now() => {} + } + drop(shutdown); + assert!(handle.tasks.iter().all(Option::is_some)); + assert!(matches!( + handle.state(), + ServiceState::Stopping | ServiceState::Stopped + )); + drop(handle); + tokio::time::timeout(Duration::from_secs(1), async { + while counter.load(Ordering::Acquire) != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("handle drop reclaims tasks retained after shutdown cancellation"); + } +} diff --git a/crates/metacrate-grid-agent/src/types.rs b/crates/metacrate-grid-agent/src/types.rs new file mode 100644 index 0000000..9be6d84 --- /dev/null +++ b/crates/metacrate-grid-agent/src/types.rs @@ -0,0 +1,408 @@ +//! Runtime-neutral, size-bounded messages shared by agent subsystems. + +use libremetaverse_types::UUID; +use std::error::Error; +use std::fmt; +use std::ops::Deref; + +/// Absolute body bound accepted at any agent boundary (8 MiB). +pub const MAX_BODY_BYTES: usize = 8 * 1024 * 1024; +/// Absolute UTF-8 message bound accepted at any agent boundary (64 KiB). +pub const MAX_MESSAGE_BYTES: usize = 64 * 1024; +/// Absolute messages retained in one normalized conversation. +pub const MAX_CONVERSATION_MESSAGES: usize = 256; +/// Absolute number of proposed calls returned for one LLM request. +pub const MAX_TOOL_CALLS: usize = 64; +/// Bound for a tool name or stable identifier. +pub const MAX_IDENTIFIER_BYTES: usize = 128; +/// Bound for JSON arguments attached to one proposed tool call (256 KiB). +pub const MAX_TOOL_ARGUMENT_BYTES: usize = 256 * 1024; +/// Bound for human-readable diagnostics published through observable events. +pub const MAX_OBSERVABLE_DETAIL_BYTES: usize = 2 * 1024; + +/// Typed failure produced before an oversized boundary value is admitted. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct BoundaryError { + field: &'static str, + actual: usize, + maximum: usize, + problem: BoundaryProblem, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum BoundaryProblem { + Empty, + TooLarge, + InvalidJson, +} + +impl BoundaryError { + const fn empty(field: &'static str, maximum: usize) -> Self { + Self { + field, + actual: 0, + maximum, + problem: BoundaryProblem::Empty, + } + } + + const fn too_large(field: &'static str, actual: usize, maximum: usize) -> Self { + Self { + field, + actual, + maximum, + problem: BoundaryProblem::TooLarge, + } + } + + const fn invalid_json(field: &'static str, actual: usize, maximum: usize) -> Self { + Self { + field, + actual, + maximum, + problem: BoundaryProblem::InvalidJson, + } + } +} + +impl fmt::Display for BoundaryError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self.problem { + BoundaryProblem::Empty => write!(formatter, "{} must not be empty", self.field), + BoundaryProblem::TooLarge => write!( + formatter, + "{} contains {} items/bytes; maximum is {}", + self.field, self.actual, self.maximum + ), + BoundaryProblem::InvalidJson => write!( + formatter, + "{} must contain one valid JSON value within {} bytes", + self.field, self.maximum + ), + } + } +} + +impl Error for BoundaryError {} + +/// UTF-8 text whose allocation can never exceed `MAX` bytes. +#[derive(Clone, Eq, Hash, Ord, PartialEq, PartialOrd)] +pub struct BoundedText(String); + +impl BoundedText { + /// Creates nonempty bounded text. + /// + /// # Errors + /// + /// Returns a typed boundary error when `value` is empty or exceeds `MAX`. + pub fn new(field: &'static str, value: impl Into) -> Result { + let value = value.into(); + if value.is_empty() { + return Err(BoundaryError::empty(field, MAX)); + } + if value.len() > MAX { + return Err(BoundaryError::too_large(field, value.len(), MAX)); + } + Ok(Self(value)) + } + + /// Creates bounded text that may be empty. + /// + /// # Errors + /// + /// Returns a typed boundary error when `value` exceeds `MAX`. + pub fn new_allow_empty( + field: &'static str, + value: impl Into, + ) -> Result { + let value = value.into(); + if value.len() > MAX { + return Err(BoundaryError::too_large(field, value.len(), MAX)); + } + Ok(Self(value)) + } + + #[must_use] + pub fn as_str(&self) -> &str { + &self.0 + } + + #[must_use] + pub fn into_inner(self) -> String { + self.0 + } +} + +impl fmt::Debug for BoundedText { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("BoundedText") + .field("bytes", &self.0.len()) + .field("maximum", &MAX) + .finish() + } +} + +impl Deref for BoundedText { + type Target = str; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +/// Collection whose element count can never exceed `MAX`. +#[derive(Clone, Eq, PartialEq)] +pub struct BoundedVec(Vec); + +impl BoundedVec { + #[must_use] + pub const fn new() -> Self { + Self(Vec::new()) + } + + /// Admits an existing vector only when its length is within the bound. + /// + /// # Errors + /// + /// Returns a typed boundary error when the vector exceeds `MAX` elements. + pub fn try_from_vec(field: &'static str, values: Vec) -> Result { + if values.len() > MAX { + return Err(BoundaryError::too_large(field, values.len(), MAX)); + } + Ok(Self(values)) + } + + /// Adds one item without allowing the collection to grow past `MAX`. + /// + /// # Errors + /// + /// Returns a typed boundary error when the collection is already full. + pub fn try_push(&mut self, field: &'static str, value: T) -> Result<(), BoundaryError> { + if self.0.len() == MAX { + return Err(BoundaryError::too_large(field, self.0.len() + 1, MAX)); + } + self.0.push(value); + Ok(()) + } + + #[must_use] + pub fn as_slice(&self) -> &[T] { + &self.0 + } + + #[must_use] + pub fn len(&self) -> usize { + self.0.len() + } + + #[must_use] + pub fn is_empty(&self) -> bool { + self.0.is_empty() + } + + #[must_use] + pub fn into_inner(self) -> Vec { + self.0 + } +} + +impl Default for BoundedVec { + fn default() -> Self { + Self::new() + } +} + +impl fmt::Debug for BoundedVec { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("BoundedVec") + .field("length", &self.0.len()) + .field("maximum", &MAX) + .finish() + } +} + +/// Grid input normalized before the coordinator consumes it. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct GridEvent { + pub sequence: u64, + pub kind: GridEventKind, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum GridEventKind { + BackendReady, + Connected, + Disconnected { + reason: BoundedText, + }, + PublicChat { + avatar_id: UUID, + body: BoundedText, + }, + InstantMessage { + avatar_id: UUID, + body: BoundedText, + }, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum MessageRole { + System, + Avatar, + Agent, + Tool, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ConversationMessage { + pub role: MessageRole, + pub body: BoundedText, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Conversation { + pub avatar_id: UUID, + pub messages: BoundedVec, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct LlmRequest { + pub request_id: u64, + pub conversation: Conversation, + pub max_output_bytes: usize, +} + +impl LlmRequest { + /// Validates the request-specific output limit. + /// + /// # Errors + /// + /// Returns a boundary error when the limit is zero or above the hard body bound. + pub fn validate(&self) -> Result<(), BoundaryError> { + if self.max_output_bytes == 0 { + return Err(BoundaryError::empty( + "llm_request.max_output_bytes", + MAX_BODY_BYTES, + )); + } + if self.max_output_bytes > MAX_BODY_BYTES { + return Err(BoundaryError::too_large( + "llm_request.max_output_bytes", + self.max_output_bytes, + MAX_BODY_BYTES, + )); + } + Ok(()) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ProposedToolCall { + pub call_id: BoundedText, + pub name: BoundedText, + pub arguments_json: BoundedText, +} + +impl ProposedToolCall { + /// Creates a proposed call only from a bounded, syntactically valid JSON value. + /// + /// # Errors + /// + /// Returns a boundary error for invalid identifiers, size, or JSON syntax. + pub fn new( + call_id: impl Into, + name: impl Into, + arguments_json: impl Into, + ) -> Result { + let arguments_json = BoundedText::new("tool_call.arguments_json", arguments_json)?; + if serde_json::from_str::(arguments_json.as_str()).is_err() { + return Err(BoundaryError::invalid_json( + "tool_call.arguments_json", + arguments_json.len(), + MAX_TOOL_ARGUMENT_BYTES, + )); + } + Ok(Self { + call_id: BoundedText::new("tool_call.call_id", call_id)?, + name: BoundedText::new("tool_call.name", name)?, + arguments_json, + }) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum PolicyDecision { + Approved { + authorization_id: u64, + }, + Denied { + reason: BoundedText, + }, + NeedsOperatorApproval { + prompt: BoundedText, + }, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct LlmResult { + pub request_id: u64, + pub response: BoundedText, + pub proposed_calls: BoundedVec, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ToolCallOutcome { + Completed { + call_id: BoundedText, + result: BoundedText, + }, + Rejected { + call_id: BoundedText, + reason: BoundedText, + }, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ControlCommand { + Pause, + Resume, + Shutdown, + ReloadBehavior, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ObservableEvent { + StateChanged { + state: &'static str, + }, + Grid(GridEvent), + Policy { + call_id: BoundedText, + decision: PolicyDecision, + }, + Tool(ToolCallOutcome), + Diagnostic { + detail: BoundedText, + }, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn bounded_text_and_collections_reject_overflow() { + assert!(BoundedText::<3>::new("text", "four").is_err()); + let mut values = BoundedVec::::new(); + values.try_push("values", 1).expect("first item fits"); + assert!(values.try_push("values", 2).is_err()); + } + + #[test] + fn tool_arguments_must_be_valid_bounded_json() { + assert!(ProposedToolCall::new("1", "look", "{\"range\": 10}").is_ok()); + assert!(ProposedToolCall::new("1", "look", "not-json").is_err()); + } +} diff --git a/crates/metacrate-grid-agent/tests/dependency_policy.rs b/crates/metacrate-grid-agent/tests/dependency_policy.rs new file mode 100644 index 0000000..183e4c1 --- /dev/null +++ b/crates/metacrate-grid-agent/tests/dependency_policy.rs @@ -0,0 +1,87 @@ +use std::collections::BTreeSet; +use std::fs; +use std::path::{Path, PathBuf}; + +const ALLOWED_DEPENDENCIES: [&str; 6] = [ + "libremetaverse", + "libremetaverse-types", + "serde", + "serde_json", + "tokio", + "url", +]; + +#[test] +fn package_has_only_reviewed_rust_dependencies_and_no_build_script() { + let root = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + assert!( + !root.join("build.rs").exists(), + "agent must not have a build script" + ); + let manifest = fs::read_to_string(root.join("Cargo.toml")).expect("read package manifest"); + let dependency_section = manifest + .split("[dependencies]") + .nth(1) + .expect("dependency section") + .split("\n[") + .next() + .expect("end of dependency section"); + let observed = dependency_section + .lines() + .filter_map(|line| line.split_once('=').map(|(name, _)| name.trim())) + .filter(|name| !name.is_empty()) + .collect::>(); + assert_eq!( + observed, + ALLOWED_DEPENDENCIES.into_iter().collect(), + "every direct dependency needs explicit policy review" + ); +} + +#[test] +fn runtime_source_has_no_subprocess_or_native_abi_escape_hatch() { + let source = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("src"); + let mut files = Vec::with_capacity(8); + collect_rust_files(&source, &mut files); + assert!( + files.len() <= 8, + "source-file count needs a reviewed bound update" + ); + for path in files { + let text = fs::read_to_string(&path).expect("read runtime source"); + for forbidden in [ + "std::process::Command", + "tokio::process", + "Command::new(", + "extern \"C\"", + "#[link(", + "unsafe fn ", + "unsafe impl ", + ] { + assert!( + !text.contains(forbidden), + "{} contains forbidden runtime boundary {forbidden:?}", + path.display() + ); + } + assert!( + !text.lines().any(|line| { + let line = line.trim_start(); + line.starts_with("unsafe {") || line.contains("= unsafe {") + }), + "{} contains an unsafe block", + path.display() + ); + } +} + +fn collect_rust_files(directory: &Path, output: &mut Vec) { + for entry in fs::read_dir(directory).expect("read source directory") { + let path = entry.expect("source entry").path(); + if path.is_dir() { + collect_rust_files(&path, output); + } else if path.extension().is_some_and(|extension| extension == "rs") { + output.push(path); + } + } +} diff --git a/docs/grid-agent-architecture.md b/docs/grid-agent-architecture.md new file mode 100644 index 0000000..1c6bb1d --- /dev/null +++ b/docs/grid-agent-architecture.md @@ -0,0 +1,87 @@ +# Grid-agent architecture foundation + +The grid agent is a workspace-owned Rust library and small service binary. Its +core depends on Tokio for scheduling and bounded channels. Its `live-grid` +feature owns the existing `libremetaverse` composition root for grid protocol +managers; the default offline graph does not compile live transports. It does +not create a second login, UDP, capabilities, inventory, or world client. It contains no +CLR/.NET loading, sidecar, subprocess adapter, provider SDK, native ABI, or +platform-specific core path. + +## Ownership and bounds + +`AgentService::start` is the sole task-creation point. It validates the complete +configuration before allocating channels, calling a backend, or permitting +network access. `ServiceHandle` then exclusively owns cancellation, both join +handles, the control sender, and observable receiver. + +| Resource | Owner | Hard/configured bound | Backpressure/termination | +| --- | --- | --- | --- | +| Grid-event queue | coordinator receives; backend sends | 8,192 / `grid_event_queue` | async send or cancellation | +| Control queue | coordinator receives; handle sends | 256 / `control_queue` | async send; closed after stop | +| Observable queue | handle receives; coordinator/backend send | 8,192 / `observable_queue` | async send or cancellation | +| Backend task | `ServiceHandle.tasks[0]` | exactly one | shared cancellation token, joined first | +| Coordinator task | `ServiceHandle.tasks[1]` | exactly one | shared cancellation token, joined second | +| Body / message | typed boundary owners | 8 MiB / 64 KiB hard ceilings, with lower configured limits | rejected before enqueue | +| Conversation / tool calls | request owner | 256 messages / 64 calls, with lower configured limits | rejected before request | +| Authorized avatars | immutable `AgentConfig` set | 1,024 hard ceiling, lower configured limit | malformed, nil, duplicate, and wildcard input rejected | +| Configuration / secret file | loader | 64 KiB / 16 KiB | regular non-symlink file only | + +`BoundedText` and `BoundedVec` make message and collection ceilings part of the +type. Dynamic configuration can lower these absolute ceilings but cannot raise +them. Backend error text is not forwarded; observations publish a fixed bounded +diagnostic. + +## State machine and shutdown + +The explicit service states are `starting -> running <-> paused -> stopping -> +stopped`, with `failed` reserved for task failure. The offline backend publishes +`BackendReady`, after which the coordinator enters `running`. A shutdown control, +direct handle shutdown, backend failure, closed owner queue, or handle drop +triggers the same cancellation token. + +Orderly shutdown first cancels, joins the backend, and then joins the +coordinator. Each join has the validated shutdown timeout. A late task is +aborted and awaited before return. A partially polled shutdown future only +borrows each join handle, so cancelling that future leaves every task in the +handle's fixed ownership slots for a retry or final drop. Dropping the handle +cancels and aborts all remaining owned tasks, so no task is detached. A +feature-enabled live adapter drops its +`LibremetaverseClientOwner` last, invoking the existing client ownership +shutdown. + +## Trust boundaries + +- JSON configuration and environment text are untrusted. Unknown fields, + oversized files, invalid booleans, conflicting modes, non-HTTP(S) URLs, + URL fragments, malformed/noncanonical/nil UUIDs, wildcard authorization, + multiline secrets, and unsafe limits fail before startup. +- `SecretString` and `EndpointUrl` are the only credential-bearing value types. + Secrets are never serializable or printable. Endpoint diagnostics replace + user information, passwords, and the complete query. +- Grid input crosses `GridBackend` only through bounded `GridEvent` values. + The `live-grid` feature supplies `LibremetaverseClientOwner`; live + implementations must own it and reuse its client and managers. +- World changes cross only `WorldMutator::apply`, which always receives the + proposed call and an explicit `PolicyDecision`. This issue supplies no live + mutation implementation. +- Signals and console output belong to the binary. The reusable core relies on + no terminal, Unix socket, Unix signal, separator, or fixed platform path. + +## Dependency diagram + +```text +metacrate-grid-agent binary (portable Ctrl-C + config path) + | + v +AgentService -> bounded Tokio channels/tasks -> injected GridBackend + | | + v v +typed config/events/policy boundaries live-grid feature boundary + | | + +-----------------> libremetaverse-types +--> libremetaverse::GridClient +``` + +The package has no build script or direct native dependency. The focused +`dependency_policy` test rejects subprocess launch sites, unsafe/native ABI +source, build scripts, and unreviewed direct dependency names in this package.