feat(grid-agent): establish architecture and config (#118)
Some checks failed
CI / rust-skia (Rust only) (push) Successful in 2m47s
CI / required (push) Failing after 2m44s

This commit is contained in:
2026-08-17 20:15:37 +00:00
parent 1254cf24e1
commit 1e1e95a58a
14 changed files with 2647 additions and 0 deletions

12
Cargo.lock generated
View File

@@ -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"

View File

@@ -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]

View File

@@ -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`" }

View File

@@ -0,0 +1,29 @@
{
"integrated": false,
"split": false,
"llm": {
"endpoint_url": "https://llm.example.invalid/v1/chat/completions",
"api_key": "<set-via-environment-or-restricted-secret-file>"
},
"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
}
}

View File

@@ -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"

View File

@@ -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.

View File

@@ -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<Box<dyn Future<Output = T> + 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<GridEvent>,
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<ToolCallOutcome, BackendError>>;
}
/// 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<Self, BackendError> {
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<GridEvent>,
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");
}
}

File diff suppressed because it is too large Load Diff

View File

@@ -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,
};

View File

@@ -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<PathBuf>,
check_config: bool,
run_once: bool,
}
fn options() -> Result<Option<Options>, 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<dyn Error>> {
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(())
}

View File

@@ -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<ConfigError> for ServiceError {
fn from(value: ConfigError) -> Self {
Self::Configuration(value)
}
}
impl From<BackendError> 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<dyn GridBackend>,
}
impl AgentService {
#[must_use]
pub fn new(config: AgentConfig, backend: Arc<dyn GridBackend>) -> Self {
Self { config, backend }
}
/// Composes the deterministic offline backend.
///
/// # Errors
///
/// Rejects non-offline configuration.
pub fn offline(config: AgentConfig) -> Result<Self, ServiceError> {
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<ServiceHandle, ServiceError> {
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::<MAX_OBSERVABLE_DETAIL_BYTES>::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<ControlCommand>,
observable_receiver: mpsc::Receiver<ObservableEvent>,
cancellation: CancellationTokenSource,
tasks: [Option<JoinHandle<Result<(), ServiceError>>>; OWNED_TASKS],
state: Arc<AtomicU8>,
active_tasks: Arc<AtomicUsize>,
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<ObservableEvent> {
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<AtomicUsize>);
impl TaskCountGuard {
fn new(counter: Arc<AtomicUsize>) -> 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<crate::types::GridEvent>,
mut controls: mpsc::Receiver<ControlCommand>,
observable: mpsc::Sender<ObservableEvent>,
cancellation: CancellationTokenSource,
state: Arc<AtomicU8>,
) -> 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<ObservableEvent>,
state: ServiceState,
cancellation: &CancellationTokenSource,
) -> Result<(), ServiceError> {
send_observable(
observable,
ObservableEvent::StateChanged {
state: state.as_str(),
},
cancellation,
)
.await
}
async fn send_observable(
observable: &mpsc::Sender<ObservableEvent>,
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<GridEvent>,
_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");
}
}

View File

@@ -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<const MAX: usize>(String);
impl<const MAX: usize> BoundedText<MAX> {
/// 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<String>) -> Result<Self, BoundaryError> {
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<String>,
) -> Result<Self, BoundaryError> {
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<const MAX: usize> fmt::Debug for BoundedText<MAX> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("BoundedText")
.field("bytes", &self.0.len())
.field("maximum", &MAX)
.finish()
}
}
impl<const MAX: usize> Deref for BoundedText<MAX> {
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<T, const MAX: usize>(Vec<T>);
impl<T, const MAX: usize> BoundedVec<T, MAX> {
#[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<T>) -> Result<Self, BoundaryError> {
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<T> {
self.0
}
}
impl<T, const MAX: usize> Default for BoundedVec<T, MAX> {
fn default() -> Self {
Self::new()
}
}
impl<T, const MAX: usize> fmt::Debug for BoundedVec<T, MAX> {
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<MAX_OBSERVABLE_DETAIL_BYTES>,
},
PublicChat {
avatar_id: UUID,
body: BoundedText<MAX_MESSAGE_BYTES>,
},
InstantMessage {
avatar_id: UUID,
body: BoundedText<MAX_MESSAGE_BYTES>,
},
}
#[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<MAX_MESSAGE_BYTES>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Conversation {
pub avatar_id: UUID,
pub messages: BoundedVec<ConversationMessage, MAX_CONVERSATION_MESSAGES>,
}
#[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<MAX_IDENTIFIER_BYTES>,
pub name: BoundedText<MAX_IDENTIFIER_BYTES>,
pub arguments_json: BoundedText<MAX_TOOL_ARGUMENT_BYTES>,
}
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<String>,
name: impl Into<String>,
arguments_json: impl Into<String>,
) -> Result<Self, BoundaryError> {
let arguments_json = BoundedText::new("tool_call.arguments_json", arguments_json)?;
if serde_json::from_str::<serde_json::Value>(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<MAX_OBSERVABLE_DETAIL_BYTES>,
},
NeedsOperatorApproval {
prompt: BoundedText<MAX_OBSERVABLE_DETAIL_BYTES>,
},
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LlmResult {
pub request_id: u64,
pub response: BoundedText<MAX_BODY_BYTES>,
pub proposed_calls: BoundedVec<ProposedToolCall, MAX_TOOL_CALLS>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ToolCallOutcome {
Completed {
call_id: BoundedText<MAX_IDENTIFIER_BYTES>,
result: BoundedText<MAX_BODY_BYTES>,
},
Rejected {
call_id: BoundedText<MAX_IDENTIFIER_BYTES>,
reason: BoundedText<MAX_OBSERVABLE_DETAIL_BYTES>,
},
}
#[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<MAX_IDENTIFIER_BYTES>,
decision: PolicyDecision,
},
Tool(ToolCallOutcome),
Diagnostic {
detail: BoundedText<MAX_OBSERVABLE_DETAIL_BYTES>,
},
}
#[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::<u8, 1>::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());
}
}

View File

@@ -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::<BTreeSet<_>>();
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<PathBuf>) {
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);
}
}
}

View File

@@ -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.