From a46bc42a8fdd4b4c7db977952a631a51ebdc84b5 Mon Sep 17 00:00:00 2001 From: Chili Palmer Date: Mon, 17 Aug 2026 20:45:56 +0000 Subject: [PATCH] feat(grid-agent): add generic LLM tool loop (#119) --- Cargo.lock | 1 + crates/metacrate-grid-agent/Cargo.toml | 3 +- crates/metacrate-grid-agent/README.md | 11 +- crates/metacrate-grid-agent/src/lib.rs | 10 + crates/metacrate-grid-agent/src/llm.rs | 764 ++++++++++++++ crates/metacrate-grid-agent/src/tool_loop.rs | 534 ++++++++++ crates/metacrate-grid-agent/src/types.rs | 15 + .../tests/dependency_policy.rs | 3 +- .../tests/llm_transport.rs | 950 ++++++++++++++++++ docs/grid-agent-architecture.md | 12 + docs/grid-agent-llm.md | 68 ++ 11 files changed, 2365 insertions(+), 6 deletions(-) create mode 100644 crates/metacrate-grid-agent/src/llm.rs create mode 100644 crates/metacrate-grid-agent/src/tool_loop.rs create mode 100644 crates/metacrate-grid-agent/tests/llm_transport.rs create mode 100644 docs/grid-agent-llm.md diff --git a/Cargo.lock b/Cargo.lock index 3f7fb67..a9a4fe1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2200,6 +2200,7 @@ version = "0.0.1" dependencies = [ "libremetaverse", "libremetaverse-types", + "reqwest", "serde", "serde_json", "tokio", diff --git a/crates/metacrate-grid-agent/Cargo.toml b/crates/metacrate-grid-agent/Cargo.toml index 23d44a4..1bd9eec 100644 --- a/crates/metacrate-grid-agent/Cargo.toml +++ b/crates/metacrate-grid-agent/Cargo.toml @@ -11,13 +11,14 @@ publish = false [dependencies] libremetaverse = { version = "0.0.1", path = "../libremetaverse", default-features = false, optional = true } libremetaverse-types = { version = "0.0.1", path = "../libremetaverse-types" } +reqwest = { version = "0.13.4", default-features = false, features = ["rustls"] } 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"] } +tokio = { version = "1.53.1", features = ["io-util", "net", "rt-multi-thread", "signal"] } [features] default = [] diff --git a/crates/metacrate-grid-agent/README.md b/crates/metacrate-grid-agent/README.md index 492b901..220b0a2 100644 --- a/crates/metacrate-grid-agent/README.md +++ b/crates/metacrate-grid-agent/README.md @@ -2,10 +2,11 @@ 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 +`metacrate-grid-agent` service binary. The default service graph 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 +and shuts down both owned tasks without contacting a grid or LLM. The library +also provides the bounded exact-endpoint LLM transport and tool loop for live +adapters. The `live-grid` feature exposes the side-effect-free owner for the existing `libremetaverse::GridClient`; later live adapters must extend that manager graph instead of adding a protocol client. @@ -37,4 +38,6 @@ cargo run --locked -p metacrate-grid-agent -- \ ``` See [`../../docs/grid-agent-architecture.md`](../../docs/grid-agent-architecture.md) -for queue/task ownership, shutdown, and trust boundaries. +for queue/task ownership, shutdown, and trust boundaries. See +[`../../docs/grid-agent-llm.md`](../../docs/grid-agent-llm.md) for the LLM wire +compatibility envelope, retry rules, and tool-loop safety contract. diff --git a/crates/metacrate-grid-agent/src/lib.rs b/crates/metacrate-grid-agent/src/lib.rs index b7bac5d..70407cb 100644 --- a/crates/metacrate-grid-agent/src/lib.rs +++ b/crates/metacrate-grid-agent/src/lib.rs @@ -6,7 +6,9 @@ pub mod backend; pub mod config; +pub mod llm; pub mod service; +pub mod tool_loop; pub mod types; #[cfg(feature = "live-grid")] @@ -17,7 +19,15 @@ pub use config::{ GridConnection, Limits, LlmConnection, MapEnvironment, OperatingMode, SecretString, StdEnvironment, Timeouts, }; +pub use llm::{ + Completion, CompletionMessage, ContentPart, ImageDetail, LlmClient, LlmError, + LlmTransportLimits, ToolDefinition, ToolSchema, Usage, +}; pub use service::{AgentService, ServiceError, ServiceHandle, ServiceState}; +pub use tool_loop::{ + HistorySummarizer, SessionGeneration, ToolExecution, ToolExecutor, ToolFuture, ToolLoop, + ToolLoopError, ToolLoopLimits, ToolLoopOutcome, +}; pub use types::{ BoundaryError, BoundedText, BoundedVec, ControlCommand, Conversation, ConversationMessage, GridEvent, GridEventKind, LlmRequest, LlmResult, MessageRole, ObservableEvent, PolicyDecision, diff --git a/crates/metacrate-grid-agent/src/llm.rs b/crates/metacrate-grid-agent/src/llm.rs new file mode 100644 index 0000000..748dec5 --- /dev/null +++ b/crates/metacrate-grid-agent/src/llm.rs @@ -0,0 +1,764 @@ +//! Provider-neutral exact-endpoint LLM transport and bounded wire envelope. + +#![allow(clippy::missing_errors_doc)] // Public methods share the exhaustive typed LlmError boundary. + +use crate::config::LlmConnection; +use crate::types::{ + BoundedText, BoundedVec, MAX_BODY_BYTES, MAX_CONVERSATION_MESSAGES, MAX_IDENTIFIER_BYTES, + MAX_MESSAGE_BYTES, MAX_TOOL_CALLS, MessageRole, ProposedToolCall, +}; +use libremetaverse_types::compat::CancellationToken; +use serde::Deserialize; +use serde_json::{Map, Value, json}; +use std::collections::{BTreeMap, BTreeSet}; +use std::error::Error; +use std::fmt; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; +use tokio::sync::Semaphore; + +const MAX_TOOLS: usize = 64; +const MAX_SCHEMA_PROPERTIES: usize = 128; +const MAX_SCHEMA_DEPTH: usize = 16; +const MAX_SCHEMA_NODES: usize = 1_024; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ContentPart { + Text(BoundedText), + Image { + url: BoundedText, + detail: ImageDetail, + }, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ImageDetail { + Auto, + Low, + High, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct CompletionMessage { + pub role: MessageRole, + pub content: BoundedVec, + pub tool_call_id: Option>, + pub proposed_calls: BoundedVec, +} + +impl CompletionMessage { + pub fn text(role: MessageRole, text: impl Into) -> Result { + let mut content = BoundedVec::new(); + content + .try_push( + "completion_message.content", + ContentPart::Text(BoundedText::new("completion_message.text", text)?), + ) + .map_err(LlmError::Boundary)?; + Ok(Self { + role, + content, + tool_call_id: None, + proposed_calls: BoundedVec::new(), + }) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ToolSchema { + Object { + properties: BTreeMap, + required: BTreeSet, + additional_properties: bool, + }, + String, + Integer, + Number, + Boolean, + Array { + items: Box, + max_items: usize, + }, +} + +impl ToolSchema { + pub fn validate_schema(&self) -> Result<(), LlmError> { + let mut remaining_nodes = MAX_SCHEMA_NODES; + self.validate_schema_inner(0, &mut remaining_nodes) + } + + fn validate_schema_inner( + &self, + depth: usize, + remaining_nodes: &mut usize, + ) -> Result<(), LlmError> { + if depth > MAX_SCHEMA_DEPTH || *remaining_nodes == 0 { + return Err(LlmError::InvalidToolSchema); + } + *remaining_nodes -= 1; + match self { + Self::Object { + properties, + required, + .. + } => { + if properties.len() > MAX_SCHEMA_PROPERTIES + || required.len() > MAX_SCHEMA_PROPERTIES + || required.iter().any(|name| !properties.contains_key(name)) + || properties + .keys() + .any(|name| name.is_empty() || name.len() > MAX_IDENTIFIER_BYTES) + { + return Err(LlmError::InvalidToolSchema); + } + for schema in properties.values() { + schema.validate_schema_inner(depth + 1, remaining_nodes)?; + } + Ok(()) + } + Self::Array { items, max_items } => { + if *max_items == 0 || *max_items > 256 { + return Err(LlmError::InvalidToolSchema); + } + items.validate_schema_inner(depth + 1, remaining_nodes) + } + _ => Ok(()), + } + } + + pub fn validate_value(&self, value: &Value) -> Result<(), LlmError> { + match self { + Self::Object { + properties, + required, + additional_properties, + } => { + let object = value.as_object().ok_or(LlmError::InvalidToolArguments)?; + if required.iter().any(|name| !object.contains_key(name)) + || (!additional_properties + && object.keys().any(|name| !properties.contains_key(name))) + { + return Err(LlmError::InvalidToolArguments); + } + for (name, value) in object { + if let Some(schema) = properties.get(name) { + schema.validate_value(value)?; + } + } + Ok(()) + } + Self::String if value.is_string() => Ok(()), + Self::Integer if value.as_i64().is_some() || value.as_u64().is_some() => Ok(()), + Self::Number if value.is_number() => Ok(()), + Self::Boolean if value.is_boolean() => Ok(()), + Self::Array { items, max_items } => { + let values = value.as_array().ok_or(LlmError::InvalidToolArguments)?; + if values.len() > *max_items { + return Err(LlmError::InvalidToolArguments); + } + for value in values { + items.validate_value(value)?; + } + Ok(()) + } + _ => Err(LlmError::InvalidToolArguments), + } + } + + fn wire_value(&self) -> Value { + match self { + Self::Object { + properties, + required, + additional_properties, + } => { + let properties = properties + .iter() + .map(|(name, schema)| (name.clone(), schema.wire_value())) + .collect::>(); + json!({ + "type": "object", + "properties": properties, + "required": required, + "additionalProperties": additional_properties + }) + } + Self::String => json!({"type":"string"}), + Self::Integer => json!({"type":"integer"}), + Self::Number => json!({"type":"number"}), + Self::Boolean => json!({"type":"boolean"}), + Self::Array { items, max_items } => { + json!({"type":"array", "items":items.wire_value(), "maxItems":max_items}) + } + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ToolDefinition { + pub name: BoundedText, + pub description: BoundedText, + pub schema: ToolSchema, + pub mutating: bool, +} + +impl ToolDefinition { + pub fn validate(&self) -> Result<(), LlmError> { + if !self + .name + .chars() + .all(|character| character.is_ascii_alphanumeric() || matches!(character, '_' | '-')) + { + return Err(LlmError::InvalidToolSchema); + } + self.schema.validate_schema() + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Usage { + pub prompt_tokens: Option, + pub completion_tokens: Option, + pub total_tokens: Option, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Completion { + pub request_id: u64, + pub correlation_id: BoundedText, + pub message: CompletionMessage, + pub usage: Option, + pub latency: Duration, + pub attempts: usize, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct LlmTransportLimits { + pub connect_timeout: Duration, + pub request_timeout: Duration, + pub read_idle_timeout: Duration, + pub pool_idle_timeout: Duration, + pub total_timeout: Duration, + pub max_prompt_bytes: usize, + pub max_response_bytes: usize, + pub max_concurrent_requests: usize, + pub max_retries: usize, + pub max_retry_delay: Duration, +} + +impl Default for LlmTransportLimits { + fn default() -> Self { + Self { + connect_timeout: Duration::from_secs(10), + request_timeout: Duration::from_mins(1), + read_idle_timeout: Duration::from_secs(15), + pool_idle_timeout: Duration::from_secs(30), + total_timeout: Duration::from_secs(90), + max_prompt_bytes: 1024 * 1024, + max_response_bytes: 1024 * 1024, + max_concurrent_requests: 8, + max_retries: 2, + max_retry_delay: Duration::from_secs(5), + } + } +} + +impl LlmTransportLimits { + pub fn validate(&self) -> Result<(), LlmError> { + if self.connect_timeout.is_zero() + || self.request_timeout.is_zero() + || self.read_idle_timeout.is_zero() + || self.pool_idle_timeout.is_zero() + || self.total_timeout.is_zero() + || self.max_prompt_bytes == 0 + || self.max_prompt_bytes > MAX_BODY_BYTES + || self.max_response_bytes == 0 + || self.max_response_bytes > MAX_BODY_BYTES + || self.max_concurrent_requests == 0 + || self.max_concurrent_requests > 256 + || self.max_retries > 8 + || self.max_retry_delay > Duration::from_mins(1) + { + return Err(LlmError::UnsafeLimits); + } + Ok(()) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum LlmError { + Boundary(crate::types::BoundaryError), + UnsafeLimits, + InvalidToolSchema, + InvalidToolArguments, + PromptTooLarge, + ResponseTooLarge, + Cancelled, + Timeout, + Transport, + RedirectRefused, + HttpStatus(u16), + MalformedJson, + UnsupportedResponse, + DuplicateToolCallId, +} + +impl fmt::Display for LlmError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Boundary(error) => write!(formatter, "LLM boundary rejected input: {error}"), + Self::UnsafeLimits => formatter.write_str("unsafe LLM transport limits"), + Self::InvalidToolSchema => formatter.write_str("invalid registered tool schema"), + Self::InvalidToolArguments => formatter.write_str("tool arguments do not match schema"), + Self::PromptTooLarge => formatter.write_str("LLM prompt exceeds its byte bound"), + Self::ResponseTooLarge => formatter.write_str("LLM response exceeds its byte bound"), + Self::Cancelled => formatter.write_str("LLM request cancelled"), + Self::Timeout => formatter.write_str("LLM request timed out"), + Self::Transport => formatter.write_str("LLM transport failed"), + Self::RedirectRefused => formatter.write_str("LLM endpoint redirect refused"), + Self::HttpStatus(status) => write!(formatter, "LLM endpoint returned HTTP {status}"), + Self::MalformedJson => formatter.write_str("LLM endpoint returned malformed JSON"), + Self::UnsupportedResponse => formatter.write_str("unsupported LLM response shape"), + Self::DuplicateToolCallId => { + formatter.write_str("LLM response repeated a tool-call ID") + } + } + } +} + +impl Error for LlmError {} + +impl From for LlmError { + fn from(value: crate::types::BoundaryError) -> Self { + Self::Boundary(value) + } +} + +pub struct LlmClient { + connection: LlmConnection, + client: reqwest::Client, + limits: LlmTransportLimits, + slots: Arc, + next_request_id: AtomicU64, +} + +impl fmt::Debug for LlmClient { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("LlmClient") + .field("endpoint", &self.connection.endpoint_url) + .field("limits", &self.limits) + .field("available_slots", &self.slots.available_permits()) + .finish_non_exhaustive() + } +} + +impl LlmClient { + pub fn new(connection: LlmConnection, limits: LlmTransportLimits) -> Result { + limits.validate()?; + let client = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .connect_timeout(limits.connect_timeout) + .timeout(limits.request_timeout) + .read_timeout(limits.read_idle_timeout) + .pool_idle_timeout(limits.pool_idle_timeout) + .pool_max_idle_per_host(limits.max_concurrent_requests) + .build() + .map_err(|_| LlmError::Transport)?; + Ok(Self { + connection, + client, + slots: Arc::new(Semaphore::new(limits.max_concurrent_requests)), + limits, + next_request_id: AtomicU64::new(1), + }) + } + + pub async fn complete( + &self, + messages: &[CompletionMessage], + tools: &[ToolDefinition], + cancellation: &CancellationToken, + ) -> Result { + if messages.is_empty() + || messages.len() > MAX_CONVERSATION_MESSAGES + || tools.len() > MAX_TOOLS + { + return Err(LlmError::PromptTooLarge); + } + for tool in tools { + tool.validate()?; + } + let body = request_body(messages, tools)?; + if body.len() > self.limits.max_prompt_bytes { + return Err(LlmError::PromptTooLarge); + } + let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed); + let correlation_id = BoundedText::new( + "llm.correlation_id", + format!("agent-request-{request_id:016x}"), + )?; + let started = Instant::now(); + let operation = async { + let permit = tokio::select! { + () = cancellation.cancelled() => return Err(LlmError::Cancelled), + permit = Arc::clone(&self.slots).acquire_owned() => permit.map_err(|_| LlmError::Cancelled)?, + }; + let result = self + .complete_with_retries(request_id, correlation_id, body, cancellation, started) + .await; + drop(permit); + result + }; + tokio::time::timeout(self.limits.total_timeout, operation) + .await + .map_err(|_| LlmError::Timeout)? + } + + async fn complete_with_retries( + &self, + request_id: u64, + correlation_id: BoundedText, + body: Vec, + cancellation: &CancellationToken, + started: Instant, + ) -> Result { + for attempt in 0..=self.limits.max_retries { + match self + .complete_once( + request_id, + correlation_id.clone(), + &body, + cancellation, + started, + attempt + 1, + ) + .await + { + Ok(completion) => return Ok(completion), + Err(AttemptError { error, retry_after }) + if attempt < self.limits.max_retries && transient(&error) => + { + let delay = retry_after + .unwrap_or_else(|| deterministic_backoff(request_id, attempt)) + .min(self.limits.max_retry_delay); + tokio::select! { + () = cancellation.cancelled() => return Err(LlmError::Cancelled), + () = tokio::time::sleep(delay) => {} + } + } + Err(AttemptError { error, .. }) => return Err(error), + } + } + Err(LlmError::Transport) + } + + async fn complete_once( + &self, + request_id: u64, + correlation_id: BoundedText, + body: &[u8], + cancellation: &CancellationToken, + started: Instant, + attempts: usize, + ) -> Result { + let response = tokio::select! { + () = cancellation.cancelled() => return Err(AttemptError::new(LlmError::Cancelled)), + response = self.client + .post(self.connection.endpoint_url.expose_url()) + .bearer_auth(self.connection.api_key.expose_secret()) + .header("content-type", "application/json") + .header("accept", "application/json") + .header("x-correlation-id", correlation_id.as_str()) + .body(body.to_vec()) + .send() => response.map_err(|error| AttemptError::new(classify_reqwest(&error)))?, + }; + let status = response.status(); + if status.is_redirection() { + return Err(AttemptError::new(LlmError::RedirectRefused)); + } + if !status.is_success() { + let retry_after = response + .headers() + .get("retry-after") + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()) + .map(Duration::from_secs); + return Err(AttemptError { + error: LlmError::HttpStatus(status.as_u16()), + retry_after, + }); + } + let bytes = read_bounded(response, self.limits.max_response_bytes, cancellation).await?; + parse_completion( + request_id, + correlation_id, + &bytes, + started.elapsed(), + attempts, + ) + .map_err(AttemptError::new) + } +} + +struct AttemptError { + error: LlmError, + retry_after: Option, +} + +impl AttemptError { + const fn new(error: LlmError) -> Self { + Self { + error, + retry_after: None, + } + } +} + +async fn read_bounded( + mut response: reqwest::Response, + maximum: usize, + cancellation: &CancellationToken, +) -> Result, AttemptError> { + if response + .content_length() + .is_some_and(|length| usize::try_from(length).map_or(true, |length| length > maximum)) + { + return Err(AttemptError::new(LlmError::ResponseTooLarge)); + } + let mut body = Vec::with_capacity( + response + .content_length() + .and_then(|value| usize::try_from(value).ok()) + .unwrap_or(0) + .min(maximum), + ); + loop { + let chunk = tokio::select! { + () = cancellation.cancelled() => return Err(AttemptError::new(LlmError::Cancelled)), + chunk = response.chunk() => chunk.map_err(|error| AttemptError::new(classify_reqwest(&error)))?, + }; + let Some(chunk) = chunk else { break }; + if body.len().saturating_add(chunk.len()) > maximum { + return Err(AttemptError::new(LlmError::ResponseTooLarge)); + } + body.extend_from_slice(&chunk); + } + Ok(body) +} + +fn request_body( + messages: &[CompletionMessage], + tools: &[ToolDefinition], +) -> Result, LlmError> { + let messages = messages.iter().map(message_wire_value).collect::>(); + let tools = tools + .iter() + .map(|tool| { + json!({ + "type":"function", + "function": { + "name": tool.name.as_str(), + "description": tool.description.as_str(), + "parameters": tool.schema.wire_value() + } + }) + }) + .collect::>(); + serde_json::to_vec(&json!({"messages":messages, "tools":tools})) + .map_err(|_| LlmError::MalformedJson) +} + +fn message_wire_value(message: &CompletionMessage) -> Value { + let role = match message.role { + MessageRole::System => "system", + MessageRole::Avatar => "user", + MessageRole::Agent => "assistant", + MessageRole::Tool => "tool", + }; + let content = message.content.as_slice().iter().map(|part| match part { + ContentPart::Text(text) => json!({"type":"text", "text":text.as_str()}), + ContentPart::Image { url, detail } => json!({ + "type":"image_url", + "image_url":{"url":url.as_str(), "detail":match detail { ImageDetail::Auto=>"auto", ImageDetail::Low=>"low", ImageDetail::High=>"high"}} + }), + }).collect::>(); + let mut value = json!({"role":role,"content":content}); + if let Some(call_id) = &message.tool_call_id { + value["tool_call_id"] = Value::String(call_id.as_str().to_owned()); + } + if !message.proposed_calls.is_empty() { + value["tool_calls"] = Value::Array(message.proposed_calls.as_slice().iter().map(|call| json!({ + "id":call.call_id.as_str(), "type":"function", "function":{"name":call.name.as_str(),"arguments":call.arguments_json.as_str()} + })).collect()); + } + value +} + +#[derive(Deserialize)] +struct WireResponse { + choices: Vec, + usage: Option, +} + +#[derive(Deserialize)] +struct WireChoice { + message: WireMessage, +} + +#[derive(Deserialize)] +struct WireMessage { + content: Option, + #[serde(default)] + tool_calls: Vec, +} + +#[derive(Deserialize)] +struct WireToolCall { + id: String, + #[serde(rename = "type")] + kind: Option, + function: WireFunction, +} + +#[derive(Deserialize)] +struct WireFunction { + name: String, + arguments: String, +} + +#[derive(Deserialize)] +struct WireUsage { + #[serde(rename = "prompt_tokens")] + prompt: Option, + #[serde(rename = "completion_tokens")] + completion: Option, + #[serde(rename = "total_tokens")] + total: Option, +} + +fn parse_completion( + request_id: u64, + correlation_id: BoundedText, + body: &[u8], + latency: Duration, + attempts: usize, +) -> Result { + let wire: WireResponse = serde_json::from_slice(body).map_err(|_| LlmError::MalformedJson)?; + let choice = wire + .choices + .into_iter() + .next() + .ok_or(LlmError::UnsupportedResponse)?; + let mut proposed_calls = BoundedVec::new(); + let mut ids = BTreeSet::new(); + for call in choice.message.tool_calls { + if call.kind.as_deref().is_some_and(|kind| kind != "function") { + return Err(LlmError::UnsupportedResponse); + } + if !ids.insert(call.id.clone()) { + return Err(LlmError::DuplicateToolCallId); + } + proposed_calls + .try_push( + "completion.tool_calls", + ProposedToolCall::from_model_output( + call.id, + call.function.name, + call.function.arguments, + )?, + ) + .map_err(LlmError::Boundary)?; + } + let mut content = BoundedVec::new(); + if let Some(text) = choice.message.content { + content + .try_push( + "completion.content", + ContentPart::Text(BoundedText::new_allow_empty("completion.text", text)?), + ) + .map_err(LlmError::Boundary)?; + } + if content.is_empty() && proposed_calls.is_empty() { + return Err(LlmError::UnsupportedResponse); + } + Ok(Completion { + request_id, + correlation_id, + message: CompletionMessage { + role: MessageRole::Agent, + content, + tool_call_id: None, + proposed_calls, + }, + usage: wire.usage.map(|usage| Usage { + prompt_tokens: usage.prompt, + completion_tokens: usage.completion, + total_tokens: usage.total, + }), + latency, + attempts, + }) +} + +fn classify_reqwest(error: &reqwest::Error) -> LlmError { + if error.is_timeout() { + LlmError::Timeout + } else { + LlmError::Transport + } +} + +fn transient(error: &LlmError) -> bool { + matches!(error, LlmError::Transport | LlmError::Timeout) + || matches!( + error, + LlmError::HttpStatus(408 | 425 | 429 | 500 | 502 | 503 | 504) + ) +} + +fn deterministic_backoff(request_id: u64, attempt: usize) -> Duration { + let exponent = u32::try_from(attempt).unwrap_or(8).min(8); + let base = 100_u64.saturating_mul(1_u64 << exponent); + let jitter = request_id + .wrapping_mul(6_364_136_223_846_793_005) + .rotate_left(exponent) + % 97; + Duration::from_millis(base.saturating_add(jitter)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::AgentConfig; + + #[tokio::test] + async fn total_timeout_includes_waiting_for_a_concurrency_slot() { + let limits = LlmTransportLimits { + total_timeout: Duration::from_millis(20), + max_concurrent_requests: 1, + ..LlmTransportLimits::default() + }; + let connection = AgentConfig::offline("http://127.0.0.1:9/exact", "test-key") + .expect("test config") + .llm; + let client = LlmClient::new(connection, limits).expect("test client"); + let permit = Arc::clone(&client.slots) + .acquire_owned() + .await + .expect("semaphore open"); + assert_eq!( + client + .complete( + &[CompletionMessage::text(MessageRole::Avatar, "hello").expect("message")], + &[], + &CancellationToken::default(), + ) + .await + .expect_err("slot wait must time out"), + LlmError::Timeout + ); + drop(permit); + } +} diff --git a/crates/metacrate-grid-agent/src/tool_loop.rs b/crates/metacrate-grid-agent/src/tool_loop.rs new file mode 100644 index 0000000..0718f4d --- /dev/null +++ b/crates/metacrate-grid-agent/src/tool_loop.rs @@ -0,0 +1,534 @@ +//! Bounded multi-turn tool orchestration above the provider-neutral transport. + +use crate::llm::{CompletionMessage, ContentPart, LlmClient, LlmError, ToolDefinition, Usage}; +use crate::types::{ + BoundedText, BoundedVec, MAX_BODY_BYTES, MAX_CONVERSATION_MESSAGES, MAX_MESSAGE_BYTES, + MAX_TOOL_CALLS, MessageRole, ProposedToolCall, +}; +use libremetaverse_types::compat::{CancellationToken, CancellationTokenSource}; +use serde_json::Value; +use std::error::Error; +use std::fmt; +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; + +const MAX_REGISTERED_TOOLS: usize = 64; +const MAX_LOOP_TURNS: usize = 32; +const MAX_SESSION_TOOL_CALLS: usize = 256; + +pub type ToolFuture<'a> = Pin + Send + 'a>>; + +/// Downstream execution boundary. Issue #120 can implement policy evaluation +/// here; this loop guarantees its input already passed name/schema checks. +pub trait ToolExecutor: Send + Sync { + fn execute<'a>( + &'a self, + definition: &'a ToolDefinition, + call: &'a ProposedToolCall, + arguments: &'a Value, + cancellation: &'a CancellationToken, + ) -> ToolFuture<'a>; +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ToolExecution { + Completed(BoundedText), + Rejected(BoundedText), + Failed(BoundedText), + /// A mutating operation may have reached the world, so it must never be + /// repeated or converted into another model request automatically. + AmbiguousMutation, +} + +/// Optional deterministic history summarizer. Failure is explicitly safe: the +/// loop inserts a bounded truncation marker instead. +pub trait HistorySummarizer: Send + Sync { + /// Returns a deterministic bounded summary, or `None` to request the safe + /// truncation marker fallback. + fn summarize(&self, omitted: &[CompletionMessage]) -> Option; +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ToolLoopLimits { + pub max_turns: usize, + pub max_tool_calls_per_turn: usize, + pub max_tool_calls_per_session: usize, + pub max_history_messages: usize, + pub max_history_bytes: usize, + pub wall_clock_timeout: Duration, +} + +impl Default for ToolLoopLimits { + fn default() -> Self { + Self { + max_turns: 8, + max_tool_calls_per_turn: 8, + max_tool_calls_per_session: 32, + max_history_messages: 64, + max_history_bytes: 512 * 1024, + wall_clock_timeout: Duration::from_mins(2), + } + } +} + +impl ToolLoopLimits { + /// # Errors + /// + /// Rejects zero or above-hard-ceiling loop limits. + pub fn validate(&self) -> Result<(), ToolLoopError> { + if self.max_turns == 0 + || self.max_turns > MAX_LOOP_TURNS + || self.max_tool_calls_per_turn == 0 + || self.max_tool_calls_per_turn > MAX_TOOL_CALLS + || self.max_tool_calls_per_session == 0 + || self.max_tool_calls_per_session > MAX_SESSION_TOOL_CALLS + || self.max_history_messages < 2 + || self.max_history_messages > MAX_CONVERSATION_MESSAGES + || self.max_history_bytes < 1024 + || self.max_history_bytes > MAX_BODY_BYTES + || self.wall_clock_timeout.is_zero() + || self.wall_clock_timeout > Duration::from_mins(10) + { + return Err(ToolLoopError::UnsafeLimits); + } + Ok(()) + } +} + +/// Monotonic session epoch. Disconnect, expiry, or operator replacement calls +/// `supersede`; late inference and tool results then fail closed. +#[derive(Debug)] +struct SessionState { + generation: u64, + cancellation: CancellationTokenSource, +} + +/// Session generation and its owned cancellation source are updated under one +/// lock so a caller can never obtain an uncancelled token for a stale epoch. +#[derive(Debug)] +pub struct SessionGeneration(Mutex); + +impl Default for SessionGeneration { + fn default() -> Self { + Self(Mutex::new(SessionState { + generation: 0, + cancellation: CancellationTokenSource::new(), + })) + } +} + +impl SessionGeneration { + #[must_use] + pub fn current(&self) -> u64 { + self.state().generation + } + + pub fn supersede(&self) -> u64 { + let mut state = self.state(); + state.cancellation.cancel(); + state.generation = state.generation.saturating_add(1); + state.cancellation = CancellationTokenSource::new(); + state.generation + } + + #[must_use] + pub fn is_current(&self, generation: u64) -> bool { + self.current() == generation + } + + fn cancellation_for(&self, generation: u64) -> Option { + let state = self.state(); + (state.generation == generation).then(|| state.cancellation.token()) + } + + fn state(&self) -> std::sync::MutexGuard<'_, SessionState> { + self.0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ToolLoopOutcome { + pub final_message: CompletionMessage, + pub turns: usize, + pub tool_calls: usize, + pub usage: BoundedVec, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum ToolLoopError { + Transport(LlmError), + UnsafeLimits, + DuplicateToolName, + DuplicateToolCallId, + TooManyTools, + EmptyHistory, + HistoryLimit, + EndlessToolLoop, + ToolCallLimit, + AmbiguousMutation, + Cancelled, + WallClockTimeout, + Superseded, + Boundary(crate::types::BoundaryError), +} + +impl fmt::Display for ToolLoopError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Transport(error) => write!(formatter, "LLM tool loop transport failed: {error}"), + Self::UnsafeLimits => formatter.write_str("unsafe tool-loop limits"), + Self::DuplicateToolName => formatter.write_str("duplicate registered tool name"), + Self::DuplicateToolCallId => { + formatter.write_str("model repeated a tool-call ID in the session") + } + Self::TooManyTools => formatter.write_str("too many registered tools"), + Self::EmptyHistory => formatter.write_str("tool loop requires initial history"), + Self::HistoryLimit => formatter.write_str("tool-loop history exceeds its hard bound"), + Self::EndlessToolLoop => formatter.write_str("model exceeded the bounded tool turns"), + Self::ToolCallLimit => { + formatter.write_str("model exceeded the session tool-call bound") + } + Self::AmbiguousMutation => formatter.write_str("mutating tool outcome is ambiguous"), + Self::Cancelled => formatter.write_str("tool loop cancelled"), + Self::WallClockTimeout => formatter.write_str("tool loop wall-clock bound elapsed"), + Self::Superseded => formatter.write_str("tool loop session was superseded"), + Self::Boundary(error) => write!(formatter, "tool loop boundary failed: {error}"), + } + } +} + +impl Error for ToolLoopError {} + +impl From for ToolLoopError { + fn from(value: LlmError) -> Self { + if value == LlmError::Cancelled { + Self::Cancelled + } else { + Self::Transport(value) + } + } +} + +impl From for ToolLoopError { + fn from(value: crate::types::BoundaryError) -> Self { + Self::Boundary(value) + } +} + +pub struct ToolLoop { + client: Arc, + tools: BoundedVec, + limits: ToolLoopLimits, + summarizer: Option>, +} + +impl fmt::Debug for ToolLoop { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("ToolLoop") + .field("client", &self.client) + .field("tool_count", &self.tools.len()) + .field("limits", &self.limits) + .field("has_summarizer", &self.summarizer.is_some()) + .finish() + } +} + +impl ToolLoop { + /// # Errors + /// + /// Rejects invalid limits, schemas, counts, or duplicate tool names. + pub fn new( + client: Arc, + tools: Vec, + limits: ToolLoopLimits, + ) -> Result { + limits.validate()?; + if tools.len() > MAX_REGISTERED_TOOLS { + return Err(ToolLoopError::TooManyTools); + } + let mut names = std::collections::BTreeSet::new(); + for tool in &tools { + tool.validate()?; + if !names.insert(tool.name.as_str()) { + return Err(ToolLoopError::DuplicateToolName); + } + } + Ok(Self { + client, + tools: BoundedVec::try_from_vec("tool_loop.tools", tools)?, + limits, + summarizer: None, + }) + } + + #[must_use] + pub fn with_summarizer(mut self, summarizer: Arc) -> Self { + self.summarizer = Some(summarizer); + self + } + + /// # Errors + /// + /// Returns typed transport, bound, cancellation, supersession, endless-loop, + /// or ambiguous-mutation failures. + pub async fn run( + &self, + mut history: Vec, + generation_owner: &SessionGeneration, + generation: u64, + cancellation: &CancellationToken, + executor: &dyn ToolExecutor, + ) -> Result { + if history.is_empty() { + return Err(ToolLoopError::EmptyHistory); + } + if history.len() > MAX_CONVERSATION_MESSAGES { + return Err(ToolLoopError::HistoryLimit); + } + let session_cancellation = generation_owner + .cancellation_for(generation) + .ok_or(ToolLoopError::Superseded)?; + let linked_cancellation = + CancellationTokenSource::new_linked(&[cancellation.clone(), session_cancellation]); + let linked_token = linked_cancellation.token(); + let future = self.run_inner( + &mut history, + generation_owner, + generation, + &linked_token, + executor, + ); + let result = tokio::time::timeout(self.limits.wall_clock_timeout, future) + .await + .map_err(|_| ToolLoopError::WallClockTimeout)?; + drop(linked_cancellation); + result + } + + async fn run_inner( + &self, + history: &mut Vec, + generation_owner: &SessionGeneration, + generation: u64, + cancellation: &CancellationToken, + executor: &dyn ToolExecutor, + ) -> Result { + let mut total_calls = 0; + let mut usage = BoundedVec::new(); + let mut seen_call_ids = std::collections::BTreeSet::new(); + for turn in 1..=self.limits.max_turns { + ensure_live(generation_owner, generation, cancellation)?; + compact_history(history, &self.limits, self.summarizer.as_deref())?; + let completion_result = self + .client + .complete(history, self.tools.as_slice(), cancellation) + .await; + ensure_live(generation_owner, generation, cancellation)?; + let completion = completion_result?; + if let Some(item) = completion.usage.clone() { + usage.try_push("tool_loop.usage", item)?; + } + let calls = completion.message.proposed_calls.clone().into_inner(); + if calls.len() > self.limits.max_tool_calls_per_turn { + return Err(ToolLoopError::ToolCallLimit); + } + if calls + .iter() + .any(|call| !seen_call_ids.insert(call.call_id.as_str().to_owned())) + { + return Err(ToolLoopError::DuplicateToolCallId); + } + history.push(completion.message.clone()); + if calls.is_empty() { + return Ok(ToolLoopOutcome { + final_message: completion.message, + turns: turn, + tool_calls: total_calls, + usage, + }); + } + total_calls = total_calls.saturating_add(calls.len()); + if total_calls > self.limits.max_tool_calls_per_session { + return Err(ToolLoopError::ToolCallLimit); + } + for call in calls { + ensure_live(generation_owner, generation, cancellation)?; + let observation = self + .evaluate_call(&call, generation_owner, generation, cancellation, executor) + .await?; + history.push(observation); + } + } + Err(ToolLoopError::EndlessToolLoop) + } + + async fn evaluate_call( + &self, + call: &ProposedToolCall, + generation_owner: &SessionGeneration, + generation: u64, + cancellation: &CancellationToken, + executor: &dyn ToolExecutor, + ) -> Result { + let Some(definition) = self + .tools + .as_slice() + .iter() + .find(|definition| definition.name.as_str() == call.name.as_str()) + else { + return tool_observation(call, "unknown tool; no execution occurred"); + }; + let Ok(arguments) = serde_json::from_str::(call.arguments_json.as_str()) else { + return tool_observation(call, "arguments are malformed JSON; no execution occurred"); + }; + if definition.schema.validate_value(&arguments).is_err() { + return tool_observation(call, "arguments rejected by registered schema"); + } + ensure_live(generation_owner, generation, cancellation)?; + let execution = tokio::select! { + () = cancellation.cancelled() => { + ensure_live(generation_owner, generation, cancellation)?; + return Err(ToolLoopError::Cancelled); + } + execution = executor.execute(definition, call, &arguments, cancellation) => execution, + }; + ensure_live(generation_owner, generation, cancellation)?; + match execution { + ToolExecution::Completed(result) => tool_observation(call, result.as_str()), + ToolExecution::Rejected(reason) => { + tool_observation(call, &format!("tool rejected: {}", reason.as_str())) + } + ToolExecution::Failed(reason) => { + tool_observation(call, &format!("tool failed safely: {}", reason.as_str())) + } + ToolExecution::AmbiguousMutation => Err(ToolLoopError::AmbiguousMutation), + } + } +} + +fn ensure_live( + owner: &SessionGeneration, + generation: u64, + cancellation: &CancellationToken, +) -> Result<(), ToolLoopError> { + if !owner.is_current(generation) { + Err(ToolLoopError::Superseded) + } else if cancellation.is_cancellation_requested() { + Err(ToolLoopError::Cancelled) + } else { + Ok(()) + } +} + +fn tool_observation( + call: &ProposedToolCall, + body: &str, +) -> Result { + const TRUNCATION_MARKER: &str = "\n[tool observation truncated]"; + let body = if body.len() <= MAX_MESSAGE_BYTES { + body.to_owned() + } else { + let mut end = MAX_MESSAGE_BYTES - TRUNCATION_MARKER.len(); + while !body.is_char_boundary(end) { + end -= 1; + } + format!("{}{TRUNCATION_MARKER}", &body[..end]) + }; + let mut message = CompletionMessage::text(MessageRole::Tool, body)?; + message.tool_call_id = Some(call.call_id.clone()); + Ok(message) +} + +fn compact_history( + history: &mut Vec, + limits: &ToolLoopLimits, + summarizer: Option<&dyn HistorySummarizer>, +) -> Result<(), ToolLoopError> { + let mut omitted = Vec::new(); + while history.len() > 1 + && (history.len() >= limits.max_history_messages + || history_bytes(history) > limits.max_history_bytes) + { + omitted.push(history.remove(0)); + } + if omitted.is_empty() { + return (history_bytes(history) <= limits.max_history_bytes) + .then_some(()) + .ok_or(ToolLoopError::HistoryLimit); + } + if history_bytes(history) > limits.max_history_bytes { + return Err(ToolLoopError::HistoryLimit); + } + let fallback = CompletionMessage::text( + MessageRole::System, + format!( + "[history safely truncated: {} earlier messages omitted]", + omitted.len() + ), + )?; + let summary = summarizer + .and_then(|summarizer| summarizer.summarize(&omitted)) + .filter(valid_summary) + .filter(|summary| { + history_bytes(history).saturating_add(message_bytes(summary)) + <= limits.max_history_bytes + }) + .or_else(|| { + (history_bytes(history).saturating_add(message_bytes(&fallback)) + <= limits.max_history_bytes) + .then_some(fallback) + }); + if let Some(summary) = summary { + history.insert(0, summary); + } + while history.len() > limits.max_history_messages { + history.remove(1); + } + Ok(()) +} + +fn history_bytes(history: &[CompletionMessage]) -> usize { + history + .iter() + .map(message_bytes) + .fold(0, usize::saturating_add) +} + +fn message_bytes(message: &CompletionMessage) -> usize { + let content = message + .content + .as_slice() + .iter() + .map(|part| match part { + ContentPart::Text(text) => text.len(), + ContentPart::Image { url, .. } => url.len(), + }) + .fold(0, usize::saturating_add); + let call_id = message + .tool_call_id + .as_ref() + .map_or(0, |call_id| call_id.len()); + message + .proposed_calls + .as_slice() + .iter() + .map(|call| { + call.call_id + .len() + .saturating_add(call.name.len()) + .saturating_add(call.arguments_json.len()) + }) + .fold(content.saturating_add(call_id), usize::saturating_add) +} + +fn valid_summary(message: &CompletionMessage) -> bool { + message.role == MessageRole::System + && message.tool_call_id.is_none() + && message.proposed_calls.is_empty() +} diff --git a/crates/metacrate-grid-agent/src/types.rs b/crates/metacrate-grid-agent/src/types.rs index 9be6d84..ebb9094 100644 --- a/crates/metacrate-grid-agent/src/types.rs +++ b/crates/metacrate-grid-agent/src/types.rs @@ -330,6 +330,21 @@ impl ProposedToolCall { arguments_json, }) } + + /// Admits bounded model output before schema/JSON validation in the tool + /// loop. This is crate-private so application-created calls still use + /// [`Self::new`]. + pub(crate) fn from_model_output( + call_id: impl Into, + name: impl Into, + arguments_json: impl Into, + ) -> Result { + Ok(Self { + call_id: BoundedText::new("tool_call.call_id", call_id)?, + name: BoundedText::new("tool_call.name", name)?, + arguments_json: BoundedText::new("tool_call.arguments_json", arguments_json)?, + }) + } } #[derive(Clone, Debug, Eq, PartialEq)] diff --git a/crates/metacrate-grid-agent/tests/dependency_policy.rs b/crates/metacrate-grid-agent/tests/dependency_policy.rs index 183e4c1..1103a36 100644 --- a/crates/metacrate-grid-agent/tests/dependency_policy.rs +++ b/crates/metacrate-grid-agent/tests/dependency_policy.rs @@ -2,9 +2,10 @@ use std::collections::BTreeSet; use std::fs; use std::path::{Path, PathBuf}; -const ALLOWED_DEPENDENCIES: [&str; 6] = [ +const ALLOWED_DEPENDENCIES: [&str; 7] = [ "libremetaverse", "libremetaverse-types", + "reqwest", "serde", "serde_json", "tokio", diff --git a/crates/metacrate-grid-agent/tests/llm_transport.rs b/crates/metacrate-grid-agent/tests/llm_transport.rs new file mode 100644 index 0000000..2692063 --- /dev/null +++ b/crates/metacrate-grid-agent/tests/llm_transport.rs @@ -0,0 +1,950 @@ +use libremetaverse_types::compat::{CancellationToken, CancellationTokenSource}; +use metacrate_grid_agent::{ + AgentConfig, BoundedText, BoundedVec, CompletionMessage, ContentPart, HistorySummarizer, + ImageDetail, LlmClient, LlmError, LlmTransportLimits, MessageRole, SessionGeneration, + ToolDefinition, ToolExecution, ToolExecutor, ToolFuture, ToolLoop, ToolLoopError, + ToolLoopLimits, ToolSchema, +}; +use serde_json::{Value, json}; +use std::collections::{BTreeMap, BTreeSet}; +use std::fmt::Write as _; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; +use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; +use tokio::net::TcpListener; +use tokio::sync::{Notify, mpsc}; + +const MAX_REQUEST_BYTES: usize = 2 * 1024 * 1024; + +#[derive(Clone)] +struct ResponsePlan { + status: u16, + headers: Vec<(String, String)>, + chunks: Vec>, + delay: Duration, +} + +impl ResponsePlan { + #[allow(clippy::needless_pass_by_value)] // Keeps nested JSON fixtures readable at call sites. + fn json(value: Value) -> Self { + Self { + status: 200, + headers: vec![("content-type".into(), "application/json".into())], + chunks: vec![serde_json::to_vec(&value).expect("fixture JSON")], + delay: Duration::ZERO, + } + } + + fn status(status: u16) -> Self { + Self { + status, + headers: Vec::new(), + chunks: vec![b"bounded error".to_vec()], + delay: Duration::ZERO, + } + } + + fn delayed(mut self, delay: Duration) -> Self { + self.delay = delay; + self + } +} + +#[derive(Debug)] +struct CapturedRequest { + head: String, + body: Value, +} + +struct FakeServer { + url: String, + requests: mpsc::Receiver, + task: tokio::task::JoinHandle<()>, + max_active: Arc, +} + +impl Drop for FakeServer { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn fake_server(plans: Vec) -> FakeServer { + let listener = TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)) + .await + .expect("bind fake LLM"); + let address = listener.local_addr().expect("fake address"); + let (request_sender, requests) = mpsc::channel(plans.len().max(1)); + let active = Arc::new(AtomicUsize::new(0)); + let max_active = Arc::new(AtomicUsize::new(0)); + let task_active = Arc::clone(&active); + let task_max = Arc::clone(&max_active); + let task = tokio::spawn(async move { + let mut handlers = Vec::with_capacity(plans.len()); + for plan in plans { + let Ok((stream, _)) = listener.accept().await else { + break; + }; + let sender = request_sender.clone(); + let active = Arc::clone(&task_active); + let max_active = Arc::clone(&task_max); + handlers.push(tokio::spawn(async move { + handle_connection(stream, plan, sender, active, max_active).await; + })); + } + for handler in handlers { + let _ = handler.await; + } + }); + FakeServer { + url: format!("http://{address}/exact/chat?route=operator"), + requests, + task, + max_active, + } +} + +async fn handle_connection( + mut stream: tokio::net::TcpStream, + plan: ResponsePlan, + sender: mpsc::Sender, + active: Arc, + max_active: Arc, +) { + let now = active.fetch_add(1, Ordering::AcqRel) + 1; + max_active.fetch_max(now, Ordering::AcqRel); + let request = read_request(&mut stream).await.expect("bounded request"); + sender.send(request).await.expect("capture owner"); + tokio::time::sleep(plan.delay).await; + let body_len = plan.chunks.iter().map(Vec::len).sum::(); + let reason = match plan.status { + 200 => "OK", + 302 => "Found", + 400 => "Bad Request", + 429 => "Too Many Requests", + 500 => "Internal Server Error", + 503 => "Service Unavailable", + _ => "Response", + }; + let mut head = format!( + "HTTP/1.1 {} {}\r\nContent-Length: {}\r\nConnection: close\r\n", + plan.status, reason, body_len + ); + for (name, value) in plan.headers { + write!(head, "{name}: {value}\r\n").expect("write response header"); + } + head.push_str("\r\n"); + stream + .write_all(head.as_bytes()) + .await + .expect("response head"); + for chunk in plan.chunks { + stream.write_all(&chunk).await.expect("response chunk"); + tokio::task::yield_now().await; + } + let _ = stream.shutdown().await; + active.fetch_sub(1, Ordering::AcqRel); +} + +async fn read_request(stream: &mut tokio::net::TcpStream) -> Result { + let mut bytes = Vec::with_capacity(4096); + let header_end = loop { + if bytes.len() >= MAX_REQUEST_BYTES { + return Err(()); + } + let mut chunk = [0_u8; 1024]; + let read = stream.read(&mut chunk).await.map_err(|_| ())?; + if read == 0 { + return Err(()); + } + bytes.extend_from_slice(&chunk[..read]); + if let Some(offset) = bytes.windows(4).position(|window| window == b"\r\n\r\n") { + break offset + 4; + } + }; + let head = String::from_utf8(bytes[..header_end].to_vec()).map_err(|_| ())?; + let content_length = head + .lines() + .find_map(|line| { + line.split_once(':').and_then(|(name, value)| { + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().ok()) + .flatten() + }) + }) + .ok_or(())?; + if content_length > MAX_REQUEST_BYTES { + return Err(()); + } + while bytes.len() - header_end < content_length { + let mut chunk = [0_u8; 1024]; + let read = stream.read(&mut chunk).await.map_err(|_| ())?; + if read == 0 { + return Err(()); + } + bytes.extend_from_slice(&chunk[..read]); + } + let body = + serde_json::from_slice(&bytes[header_end..header_end + content_length]).map_err(|_| ())?; + Ok(CapturedRequest { head, body }) +} + +fn response_text(text: &str) -> Value { + json!({ + "id":"ignored-provider-field", + "choices":[{"message":{"role":"assistant","content":text},"unknown":"kept-compatible"}], + "usage":{"prompt_tokens":3,"completion_tokens":2,"total_tokens":5}, + "unknown_root":{"harmless":true} + }) +} + +#[allow(clippy::needless_pass_by_value)] // Keeps nested JSON fixtures readable at call sites. +fn response_calls(calls: Value) -> Value { + json!({"choices":[{"message":{"role":"assistant","content":null,"tool_calls":calls}}]}) +} + +fn call(id: &str, name: &str, arguments: &str) -> Value { + json!({"id":id,"type":"function","function":{"name":name,"arguments":arguments}}) +} + +fn limits() -> LlmTransportLimits { + LlmTransportLimits { + connect_timeout: Duration::from_secs(1), + request_timeout: Duration::from_secs(2), + read_idle_timeout: Duration::from_secs(1), + pool_idle_timeout: Duration::from_secs(1), + total_timeout: Duration::from_secs(4), + max_prompt_bytes: 64 * 1024, + max_response_bytes: 64 * 1024, + max_concurrent_requests: 2, + max_retries: 0, + max_retry_delay: Duration::from_millis(20), + } +} + +fn client(url: &str, limits: LlmTransportLimits) -> Arc { + let connection = AgentConfig::offline(url, "super-secret-api-key") + .expect("valid fake endpoint") + .llm; + Arc::new(LlmClient::new(connection, limits).expect("valid client")) +} + +fn message() -> CompletionMessage { + CompletionMessage::text(MessageRole::Avatar, "hello").expect("bounded message") +} + +fn system_message() -> CompletionMessage { + CompletionMessage::text(MessageRole::System, "use registered tools only") + .expect("bounded system message") +} + +fn image_message() -> CompletionMessage { + CompletionMessage { + role: MessageRole::Avatar, + content: BoundedVec::try_from_vec( + "image.content", + vec![ContentPart::Image { + url: BoundedText::new("image.url", "https://assets.invalid/object.png") + .expect("image URL"), + detail: ImageDetail::Low, + }], + ) + .expect("image content"), + tool_call_id: None, + proposed_calls: BoundedVec::new(), + } +} + +fn tool(name: &str, mutating: bool) -> ToolDefinition { + let mut properties = BTreeMap::new(); + properties.insert("target".into(), ToolSchema::String); + ToolDefinition { + name: BoundedText::new("tool.name", name).expect("tool name"), + description: BoundedText::new("tool.description", "test tool").expect("description"), + schema: ToolSchema::Object { + properties, + required: BTreeSet::from(["target".into()]), + additional_properties: false, + }, + mutating, + } +} + +#[tokio::test] +async fn exact_url_auth_schema_fragmentation_and_metadata_are_verified() { + let body = serde_json::to_vec(&response_text("done")).expect("response"); + let midpoint = body.len() / 2; + let mut server = fake_server(vec![ResponsePlan { + status: 200, + headers: vec![("content-type".into(), "application/json".into())], + chunks: vec![body[..midpoint].to_vec(), body[midpoint..].to_vec()], + delay: Duration::ZERO, + }]) + .await; + let client = client(&server.url, limits()); + let source = CancellationTokenSource::new(); + let messages = [system_message(), message(), image_message()]; + let completion = client + .complete(&messages, &[tool("look", false)], &source.token()) + .await + .expect("completion"); + assert_eq!(completion.request_id, 1); + assert_eq!(completion.attempts, 1); + assert_eq!(completion.usage.expect("usage").total_tokens, Some(5)); + let request = server.requests.recv().await.expect("captured request"); + assert!( + request + .head + .starts_with("POST /exact/chat?route=operator HTTP/1.1") + ); + assert!( + request + .head + .to_ascii_lowercase() + .contains("authorization: bearer super-secret-api-key") + ); + assert!( + request + .head + .to_ascii_lowercase() + .contains("x-correlation-id: agent-request-0000000000000001") + ); + assert!(request.body.get("model").is_none()); + assert!(request.body.get("provider").is_none()); + assert_eq!(request.body["tools"][0]["type"], "function"); + assert_eq!(request.body["messages"][0]["role"], "system"); + assert_eq!( + request.body["messages"][2]["content"][0]["type"], + "image_url" + ); + assert_eq!( + request.body["messages"][2]["content"][0]["image_url"]["detail"], + "low" + ); + let diagnostic = format!("{client:?}"); + assert!(!diagnostic.contains("super-secret-api-key")); + assert!(!diagnostic.contains("route=operator")); +} + +#[tokio::test] +async fn prompt_bytes_and_schema_complexity_fail_before_network_access() { + let mut tiny = limits(); + tiny.max_prompt_bytes = 32; + let client = client("http://127.0.0.1:9/exact", tiny); + assert_eq!( + client + .complete(&[message()], &[], &CancellationToken::default()) + .await + .expect_err("prompt cap"), + LlmError::PromptTooLarge + ); + + let mut schema = ToolSchema::String; + for _ in 0..18 { + schema = ToolSchema::Array { + items: Box::new(schema), + max_items: 1, + }; + } + let definition = ToolDefinition { + name: BoundedText::new("tool.name", "deep").expect("name"), + description: BoundedText::new("tool.description", "too deep").expect("description"), + schema, + mutating: false, + }; + assert_eq!(definition.validate(), Err(LlmError::InvalidToolSchema)); +} + +#[tokio::test] +async fn malformed_html_oversize_and_duplicate_calls_are_typed() { + let cases = [ + ( + ResponsePlan::json(json!({"not_choices":[]})), + LlmError::MalformedJson, + ), + ( + ResponsePlan { + status: 200, + headers: vec![("content-type".into(), "text/html".into())], + chunks: vec![b"not JSON".to_vec()], + delay: Duration::ZERO, + }, + LlmError::MalformedJson, + ), + (ResponsePlan::status(500), LlmError::HttpStatus(500)), + ( + ResponsePlan::json(response_calls(json!([{ + "id":"unsupported-1", + "type":"not-a-function", + "function":{"name":"look","arguments":"{}"} + }]))), + LlmError::UnsupportedResponse, + ), + ( + ResponsePlan::json(response_calls(json!([ + call("same", "look", "{}"), + call("same", "look", "{}") + ]))), + LlmError::DuplicateToolCallId, + ), + ]; + for (plan, expected) in cases { + let server = fake_server(vec![plan]).await; + let error = client(&server.url, limits()) + .complete(&[message()], &[], &CancellationToken::default()) + .await + .expect_err("typed failure"); + assert_eq!(error, expected); + } + + let mut small = limits(); + small.max_response_bytes = 32; + let server = fake_server(vec![ResponsePlan::json(response_text( + "this response is deliberately larger than thirty-two bytes", + ))]) + .await; + assert_eq!( + client(&server.url, small) + .complete(&[message()], &[], &CancellationToken::default()) + .await + .expect_err("oversize rejected"), + LlmError::ResponseTooLarge + ); +} + +#[tokio::test] +async fn redirects_never_forward_bearer_credentials() { + let mut target = fake_server(vec![ResponsePlan::json(response_text("leaked"))]).await; + let redirect = ResponsePlan { + status: 302, + headers: vec![("location".into(), target.url.clone())], + chunks: Vec::new(), + delay: Duration::ZERO, + }; + let mut source = fake_server(vec![redirect]).await; + let error = client(&source.url, limits()) + .complete(&[message()], &[], &CancellationToken::default()) + .await + .expect_err("redirect refused"); + assert_eq!(error, LlmError::RedirectRefused); + source.requests.recv().await.expect("source request"); + assert!( + tokio::time::timeout(Duration::from_millis(100), target.requests.recv()) + .await + .is_err(), + "redirect target must receive no request" + ); +} + +#[tokio::test] +async fn retry_classification_retry_after_timeout_and_cancellation_are_bounded() { + let retry = ResponsePlan { + status: 503, + headers: vec![("retry-after".into(), "0".into())], + chunks: Vec::new(), + delay: Duration::ZERO, + }; + let mut server = fake_server(vec![retry, ResponsePlan::json(response_text("recovered"))]).await; + let mut retry_limits = limits(); + retry_limits.max_retries = 1; + let completion = client(&server.url, retry_limits) + .complete(&[message()], &[], &CancellationToken::default()) + .await + .expect("transient retry"); + assert_eq!(completion.attempts, 2); + server.requests.recv().await.expect("first attempt"); + server.requests.recv().await.expect("second attempt"); + + let server = fake_server(vec![ + ResponsePlan::status(400), + ResponsePlan::json(response_text("must not retry")), + ]) + .await; + let mut retry_limits = limits(); + retry_limits.max_retries = 1; + assert_eq!( + client(&server.url, retry_limits) + .complete(&[message()], &[], &CancellationToken::default()) + .await + .expect_err("400 not retried"), + LlmError::HttpStatus(400) + ); + + let mut slow_limits = limits(); + slow_limits.request_timeout = Duration::from_millis(50); + slow_limits.total_timeout = Duration::from_millis(100); + let server = fake_server(vec![ + ResponsePlan::json(response_text("late")).delayed(Duration::from_secs(1)), + ]) + .await; + assert_eq!( + client(&server.url, slow_limits) + .complete(&[message()], &[], &CancellationToken::default()) + .await + .expect_err("timeout"), + LlmError::Timeout + ); + + let server = fake_server(vec![ + ResponsePlan::json(response_text("late")).delayed(Duration::from_secs(1)), + ]) + .await; + let client = client(&server.url, limits()); + let source = CancellationTokenSource::new(); + let token = source.token(); + let messages = [message()]; + let request = client.complete(&messages, &[], &token); + tokio::pin!(request); + tokio::select! { + result = &mut request => panic!("request unexpectedly completed: {result:?}"), + () = tokio::time::sleep(Duration::from_millis(20)) => source.cancel(), + } + assert_eq!(request.await.expect_err("cancelled"), LlmError::Cancelled); +} + +#[tokio::test] +async fn concurrent_slow_sessions_respect_the_shared_semaphore() { + let plans = (0..6) + .map(|_| ResponsePlan::json(response_text("done")).delayed(Duration::from_millis(80))) + .collect(); + let server = fake_server(plans).await; + let client = client(&server.url, limits()); + let mut tasks = Vec::with_capacity(6); + for _ in 0..6 { + let client = Arc::clone(&client); + tasks.push(tokio::spawn(async move { + client + .complete(&[message()], &[], &CancellationToken::default()) + .await + })); + } + for task in tasks { + task.await + .expect("session task") + .expect("session completion"); + } + assert_eq!(server.max_active.load(Ordering::Acquire), 2); +} + +struct RecordingExecutor { + calls: AtomicUsize, + result: ToolExecution, +} + +impl ToolExecutor for RecordingExecutor { + fn execute<'a>( + &'a self, + _definition: &'a ToolDefinition, + _call: &'a metacrate_grid_agent::ProposedToolCall, + _arguments: &'a Value, + _cancellation: &'a CancellationToken, + ) -> ToolFuture<'a> { + self.calls.fetch_add(1, Ordering::AcqRel); + let result = self.result.clone(); + Box::pin(async move { result }) + } +} + +fn loop_limits() -> ToolLoopLimits { + ToolLoopLimits { + max_turns: 3, + max_tool_calls_per_turn: 4, + max_tool_calls_per_session: 4, + max_history_messages: 8, + max_history_bytes: 4096, + wall_clock_timeout: Duration::from_secs(2), + } +} + +#[tokio::test] +async fn valid_tool_round_trip_and_model_authored_summary_complete() { + let mut server = fake_server(vec![ + ResponsePlan::json(response_calls(json!([call( + "call-1", + "look", + r#"{"target":"tree"}"#, + )]))), + ResponsePlan::json(response_text("I looked at the tree.")), + ]) + .await; + let executor = RecordingExecutor { + calls: AtomicUsize::new(0), + result: ToolExecution::Completed( + BoundedText::new("result", "tree is nearby").expect("result"), + ), + }; + let loop_ = ToolLoop::new( + client(&server.url, limits()), + vec![tool("look", false)], + loop_limits(), + ) + .expect("loop"); + let generation = SessionGeneration::default(); + let outcome = loop_ + .run( + vec![message()], + &generation, + generation.current(), + &CancellationToken::default(), + &executor, + ) + .await + .expect("tool loop"); + assert_eq!(outcome.turns, 2); + assert_eq!(outcome.tool_calls, 1); + assert_eq!(executor.calls.load(Ordering::Acquire), 1); + server.requests.recv().await.expect("tool request"); + let round_trip = server.requests.recv().await.expect("observation request"); + let serialized = round_trip.body.to_string(); + assert!(serialized.contains("tree is nearby")); + assert!(serialized.contains("call-1")); +} + +#[tokio::test] +async fn unknown_and_malformed_calls_become_observations_without_execution() { + let mut server = fake_server(vec![ + ResponsePlan::json(response_calls(json!([ + call("unknown-1", "missing", "{}"), + call("bad-1", "look", "not-json"), + call("schema-1", "look", r#"{"wrong":true}"#) + ]))), + ResponsePlan::json(response_text("handled safely")), + ]) + .await; + let executor = RecordingExecutor { + calls: AtomicUsize::new(0), + result: ToolExecution::Completed(BoundedText::new("result", "unused").expect("result")), + }; + let loop_ = ToolLoop::new( + client(&server.url, limits()), + vec![tool("look", false)], + loop_limits(), + ) + .expect("loop"); + let generation = SessionGeneration::default(); + loop_ + .run( + vec![message()], + &generation, + 0, + &CancellationToken::default(), + &executor, + ) + .await + .expect("safe observations"); + assert_eq!(executor.calls.load(Ordering::Acquire), 0); + server.requests.recv().await.expect("first request"); + let observations = server + .requests + .recv() + .await + .expect("second request") + .body + .to_string(); + assert!(observations.contains("unknown tool")); + assert!(observations.contains("malformed JSON")); + assert!(observations.contains("registered schema")); +} + +#[tokio::test] +async fn endless_loops_and_ambiguous_mutations_fail_without_reexecution() { + let server = fake_server( + (1..=3) + .map(|number| { + ResponsePlan::json(response_calls(json!([call( + &format!("call-{number}"), + "look", + r#"{"target":"tree"}"#, + )]))) + }) + .collect(), + ) + .await; + let executor = RecordingExecutor { + calls: AtomicUsize::new(0), + result: ToolExecution::Completed(BoundedText::new("result", "again").expect("result")), + }; + let loop_ = ToolLoop::new( + client(&server.url, limits()), + vec![tool("look", false)], + loop_limits(), + ) + .expect("loop"); + assert_eq!( + loop_ + .run( + vec![message()], + &SessionGeneration::default(), + 0, + &CancellationToken::default(), + &executor, + ) + .await + .expect_err("endless loop"), + ToolLoopError::EndlessToolLoop + ); + + let server = fake_server(vec![ResponsePlan::json(response_calls(json!([call( + "mutate-1", + "rez", + r#"{"target":"cube"}"#, + )])))]) + .await; + let executor = RecordingExecutor { + calls: AtomicUsize::new(0), + result: ToolExecution::AmbiguousMutation, + }; + let loop_ = ToolLoop::new( + client(&server.url, limits()), + vec![tool("rez", true)], + loop_limits(), + ) + .expect("loop"); + assert_eq!( + loop_ + .run( + vec![message()], + &SessionGeneration::default(), + 0, + &CancellationToken::default(), + &executor, + ) + .await + .expect_err("ambiguous mutation"), + ToolLoopError::AmbiguousMutation + ); + assert_eq!(executor.calls.load(Ordering::Acquire), 1); +} + +#[tokio::test] +async fn repeated_call_id_across_turns_is_rejected_before_reexecution() { + let duplicate = ResponsePlan::json(response_calls(json!([call( + "same-call", + "look", + r#"{"target":"tree"}"#, + )]))); + let server = fake_server(vec![duplicate.clone(), duplicate]).await; + let executor = RecordingExecutor { + calls: AtomicUsize::new(0), + result: ToolExecution::Completed(BoundedText::new("result", "first").expect("result")), + }; + let loop_ = ToolLoop::new( + client(&server.url, limits()), + vec![tool("look", false)], + loop_limits(), + ) + .expect("loop"); + assert_eq!( + loop_ + .run( + vec![message()], + &SessionGeneration::default(), + 0, + &CancellationToken::default(), + &executor, + ) + .await + .expect_err("duplicate call ID"), + ToolLoopError::DuplicateToolCallId + ); + assert_eq!(executor.calls.load(Ordering::Acquire), 1); +} + +#[tokio::test] +async fn session_call_and_wall_clock_budgets_are_enforced() { + let server = fake_server(vec![ResponsePlan::json(response_calls(json!([ + call("one", "look", r#"{"target":"tree"}"#), + call("two", "look", r#"{"target":"rock"}"#) + ])))]) + .await; + let executor = RecordingExecutor { + calls: AtomicUsize::new(0), + result: ToolExecution::Completed(BoundedText::new("result", "unused").expect("result")), + }; + let mut one_call = loop_limits(); + one_call.max_tool_calls_per_turn = 1; + one_call.max_tool_calls_per_session = 1; + let loop_ = ToolLoop::new( + client(&server.url, limits()), + vec![tool("look", false)], + one_call, + ) + .expect("loop"); + assert_eq!( + loop_ + .run( + vec![message()], + &SessionGeneration::default(), + 0, + &CancellationToken::default(), + &executor, + ) + .await + .expect_err("call budget"), + ToolLoopError::ToolCallLimit + ); + assert_eq!(executor.calls.load(Ordering::Acquire), 0); + + let server = fake_server(vec![ + ResponsePlan::json(response_text("too late")).delayed(Duration::from_secs(1)), + ]) + .await; + let mut short_loop = loop_limits(); + short_loop.wall_clock_timeout = Duration::from_millis(30); + let loop_ = ToolLoop::new(client(&server.url, limits()), vec![], short_loop).expect("loop"); + assert_eq!( + loop_ + .run( + vec![message()], + &SessionGeneration::default(), + 0, + &CancellationToken::default(), + &executor, + ) + .await + .expect_err("wall-clock budget"), + ToolLoopError::WallClockTimeout + ); +} + +#[tokio::test] +async fn superseded_session_discards_late_model_result_before_execution() { + let mut server = fake_server(vec![ + ResponsePlan::json(response_calls(json!([call( + "late-1", + "look", + r#"{"target":"tree"}"#, + )]))) + .delayed(Duration::from_millis(100)), + ]) + .await; + let executor = Arc::new(RecordingExecutor { + calls: AtomicUsize::new(0), + result: ToolExecution::Completed(BoundedText::new("result", "late").expect("result")), + }); + let loop_ = ToolLoop::new( + client(&server.url, limits()), + vec![tool("look", false)], + loop_limits(), + ) + .expect("loop"); + let generation = Arc::new(SessionGeneration::default()); + let token = CancellationToken::default(); + let run = loop_.run(vec![message()], &generation, 0, &token, executor.as_ref()); + tokio::pin!(run); + tokio::select! { + request = server.requests.recv() => { + request.expect("in-flight request"); + } + result = &mut run => panic!("session completed before supersession: {result:?}"), + } + generation.supersede(); + assert_eq!( + run.await.expect_err("superseded"), + ToolLoopError::Superseded + ); + assert_eq!(executor.calls.load(Ordering::Acquire), 0); +} + +struct BlockingExecutor { + started: Notify, + completed: AtomicUsize, +} + +impl ToolExecutor for BlockingExecutor { + fn execute<'a>( + &'a self, + _definition: &'a ToolDefinition, + _call: &'a metacrate_grid_agent::ProposedToolCall, + _arguments: &'a Value, + _cancellation: &'a CancellationToken, + ) -> ToolFuture<'a> { + Box::pin(async move { + self.started.notify_one(); + std::future::pending::<()>().await; + self.completed.fetch_add(1, Ordering::AcqRel); + ToolExecution::AmbiguousMutation + }) + } +} + +#[tokio::test] +async fn superseding_session_cancels_an_in_flight_tool_executor() { + let server = fake_server(vec![ResponsePlan::json(response_calls(json!([call( + "in-flight-1", + "look", + r#"{"target":"tree"}"#, + )])))]) + .await; + let executor = Arc::new(BlockingExecutor { + started: Notify::new(), + completed: AtomicUsize::new(0), + }); + let loop_ = ToolLoop::new( + client(&server.url, limits()), + vec![tool("look", false)], + loop_limits(), + ) + .expect("loop"); + let generation = SessionGeneration::default(); + let token = CancellationToken::default(); + let run = loop_.run(vec![message()], &generation, 0, &token, executor.as_ref()); + tokio::pin!(run); + tokio::select! { + () = executor.started.notified() => {} + result = &mut run => panic!("tool loop completed before supersession: {result:?}"), + } + generation.supersede(); + assert_eq!( + run.await.expect_err("superseded executor"), + ToolLoopError::Superseded + ); + assert_eq!(executor.completed.load(Ordering::Acquire), 0); +} + +struct FailingSummarizer; + +impl HistorySummarizer for FailingSummarizer { + fn summarize(&self, _omitted: &[CompletionMessage]) -> Option { + None + } +} + +#[tokio::test] +async fn failed_history_summary_degrades_to_bounded_safe_truncation() { + let mut server = fake_server(vec![ResponsePlan::json(response_text("done"))]).await; + let mut compact = loop_limits(); + compact.max_history_messages = 2; + let loop_ = ToolLoop::new(client(&server.url, limits()), vec![], compact) + .expect("loop") + .with_summarizer(Arc::new(FailingSummarizer)); + let history = vec![ + CompletionMessage::text(MessageRole::System, "old system").expect("message"), + CompletionMessage::text(MessageRole::Avatar, "old question").expect("message"), + CompletionMessage::text(MessageRole::Agent, "old answer").expect("message"), + message(), + ]; + loop_ + .run( + history, + &SessionGeneration::default(), + 0, + &CancellationToken::default(), + &RecordingExecutor { + calls: AtomicUsize::new(0), + result: ToolExecution::AmbiguousMutation, + }, + ) + .await + .expect("compacted completion"); + let request = server + .requests + .recv() + .await + .expect("request") + .body + .to_string(); + assert!(request.contains("history safely truncated")); + assert!(!request.contains("old question")); +} diff --git a/docs/grid-agent-architecture.md b/docs/grid-agent-architecture.md index 1c6bb1d..2c5cb96 100644 --- a/docs/grid-agent-architecture.md +++ b/docs/grid-agent-architecture.md @@ -24,6 +24,8 @@ handles, the control sender, and observable receiver. | 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 | +| LLM request slots | shared `LlmClient` semaphore | 256 hard / configured concurrent requests | async acquire or cancellation | +| Reasoning/tool session | `ToolLoop` caller | 32 turns / 256 calls hard, with lower configured limits | total timeout, cancellation, or supersession | | 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 | @@ -65,6 +67,10 @@ shutdown. - World changes cross only `WorldMutator::apply`, which always receives the proposed call and an explicit `PolicyDecision`. This issue supplies no live mutation implementation. +- LLM traffic crosses one exact configured URL through `LlmClient`. Redirects + are refused, response bodies are bounded while streaming, bearer secrets are + redacted, and provider/model discovery does not exist. Proposed calls cross + `ToolExecutor` only after registered-name and schema validation. - Signals and console output belong to the binary. The reusable core relies on no terminal, Unix socket, Unix signal, separator, or fixed platform path. @@ -80,8 +86,14 @@ AgentService -> bounded Tokio channels/tasks -> injected GridBackend typed config/events/policy boundaries live-grid feature boundary | | +-----------------> libremetaverse-types +--> libremetaverse::GridClient + +avatar session -> bounded ToolLoop -> exact-endpoint LlmClient + | + +-> validated ToolExecutor boundary ``` 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. +The precise LLM compatibility and cancellation contract is documented in +[`grid-agent-llm.md`](grid-agent-llm.md). diff --git a/docs/grid-agent-llm.md b/docs/grid-agent-llm.md new file mode 100644 index 0000000..f37ab63 --- /dev/null +++ b/docs/grid-agent-llm.md @@ -0,0 +1,68 @@ +# Grid-agent LLM transport and tool loop + +The grid agent talks to one operator-supplied OpenAI-compatible chat-completion +endpoint. `llm.endpoint_url` is the complete request URL and `llm.api_key` is +the bearer credential. The client sends `POST` to that exact URL. It does not +append a path, select a provider, discover models, send a model field, or retry +against another service. Endpoint routing and model selection remain operator +responsibilities. + +## Compatibility envelope + +Requests contain only the normalized `messages` and `tools` members. Message +roles map to `system`, `user`, `assistant`, and `tool`. Content is an array of +`text` or `image_url` parts. Tool definitions use the standard function name, +description, and JSON-schema parameters envelope. Tool observations carry the +original `tool_call_id`. The client accepts the first response choice, optional +text, function tool calls, and optional token usage. Unknown response members +are ignored; missing required members, unsupported empty choices, malformed +JSON, duplicate call IDs, and oversized values return typed errors. + +The HTTP client uses Rustls and performs no automatic content decompression +because no compression feature is enabled. Redirect following is disabled, so +a bearer credential can never be forwarded to either a same-origin or +cross-origin redirect target. Connect, whole-request, response-idle, pool-idle, +and total elapsed timeouts are independently bounded. Prompt bytes, response +bytes, concurrent requests, retry count, and retry delay also have validated +hard ceilings. Only transport failures, timeouts, and HTTP 408, 425, 429, 500, +502, 503, or 504 are retryable. `Retry-After` seconds are honored only up to the +configured delay ceiling; otherwise bounded deterministic jitter is used. + +## Tool-loop safety + +`ToolLoop` validates every registered schema before use. A proposed call must +have a unique session call ID, a registered name, valid JSON arguments, and +arguments matching that registered schema before it reaches `ToolExecutor`. +Unknown names, malformed arguments, and schema mismatches become bounded tool +observations for the next model turn and never call the executor. Tool +executions are sequential, making the simultaneous execution ceiling one. + +Turn count, calls per turn and session, history messages, history bytes, +wall-clock time, and collected usage records are bounded. When history exceeds its +configured envelope, an injected deterministic summarizer may compact it. A +failed summarizer inserts a fixed bounded truncation marker and retains recent +context. An ambiguous mutating result terminates the loop immediately; it is +never retried or turned into another model request. + +Shutdown, operator cancellation, disconnect, session expiry, and avatar-session +replacement use cancellation plus `SessionGeneration`. Superseding a generation +cancels its in-flight HTTP request or executor and prevents a late result from +starting another action. Request and correlation IDs are deterministic and the +completion exposes token usage, latency, and attempt count. The final returned +message is the model-authored action summary. The wire mapping has no reasoning +or chain-of-thought field and does not retain or expose hidden reasoning. + +## Focused verification + +The `llm_transport` integration suite runs only against bounded loopback fake +endpoints. It covers exact URL and authorization behavior, secret redaction, +fragmented responses, images and tool schemas, malformed and oversized input, +redirect refusal, transient-only retry, cancellation and timeout races, +concurrency, complete tool round trips, invalid-call observations, endless +loops, duplicate IDs, ambiguous mutation, supersession, and safe history +compaction. + +```sh +cargo test --locked -p metacrate-grid-agent --test llm_transport +cargo clippy --locked -p metacrate-grid-agent --all-targets -- -D warnings +```