feat(grid-agent): establish architecture and config (#118)
This commit is contained in:
12
Cargo.lock
generated
12
Cargo.lock
generated
@@ -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"
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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`" }
|
||||
|
||||
29
config/grid-agent.example.json
Normal file
29
config/grid-agent.example.json
Normal 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
|
||||
}
|
||||
}
|
||||
31
crates/metacrate-grid-agent/Cargo.toml
Normal file
31
crates/metacrate-grid-agent/Cargo.toml
Normal 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"
|
||||
40
crates/metacrate-grid-agent/README.md
Normal file
40
crates/metacrate-grid-agent/README.md
Normal 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.
|
||||
156
crates/metacrate-grid-agent/src/backend.rs
Normal file
156
crates/metacrate-grid-agent/src/backend.rs
Normal 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");
|
||||
}
|
||||
}
|
||||
1121
crates/metacrate-grid-agent/src/config.rs
Normal file
1121
crates/metacrate-grid-agent/src/config.rs
Normal file
File diff suppressed because it is too large
Load Diff
25
crates/metacrate-grid-agent/src/lib.rs
Normal file
25
crates/metacrate-grid-agent/src/lib.rs
Normal 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,
|
||||
};
|
||||
135
crates/metacrate-grid-agent/src/main.rs
Normal file
135
crates/metacrate-grid-agent/src/main.rs
Normal 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(())
|
||||
}
|
||||
513
crates/metacrate-grid-agent/src/service.rs
Normal file
513
crates/metacrate-grid-agent/src/service.rs
Normal 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");
|
||||
}
|
||||
}
|
||||
408
crates/metacrate-grid-agent/src/types.rs
Normal file
408
crates/metacrate-grid-agent/src/types.rs
Normal 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());
|
||||
}
|
||||
}
|
||||
87
crates/metacrate-grid-agent/tests/dependency_policy.rs
Normal file
87
crates/metacrate-grid-agent/tests/dependency_policy.rs
Normal 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
87
docs/grid-agent-architecture.md
Normal file
87
docs/grid-agent-architecture.md
Normal 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.
|
||||
Reference in New Issue
Block a user