use async_trait::async_trait;
use futures_util::SinkExt as _;
use futures_util::StreamExt as _;
use serde_json::{Map, Value};
use crate::ports::{
Capability, Completion, CompletionBuilder, Prompt, TextDelta, TextModel, TextModelError,
ToolCall, ToolResult, ToolSpec, Turn,
};
use crate::stream::{BoxStream, pump};
#[async_trait]
pub trait ToolExecutor: Send + Sync {
async fn execute(&self, call: &ToolCall) -> Result<String, String>;
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct ToolBudget {
pub max_steps: u32,
pub max_total_tokens: Option<u64>,
}
impl ToolBudget {
#[must_use]
pub fn steps(max_steps: u32) -> Self {
Self {
max_steps,
max_total_tokens: None,
}
}
#[must_use]
pub fn max_total_tokens(mut self, max_total_tokens: u64) -> Self {
self.max_total_tokens = Some(max_total_tokens);
self
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StepUsage {
pub input_tokens: u64,
pub output_tokens: u64,
pub cached_input_tokens: Option<u64>,
pub tool_calls: usize,
}
#[derive(Debug, Clone, PartialEq)]
pub struct ToolLoopOutcome {
pub completion: Completion,
pub messages: Vec<Turn>,
pub steps: Vec<StepUsage>,
}
impl ToolLoopOutcome {
#[must_use]
pub fn input_tokens(&self) -> u64 {
self.steps.iter().map(|step| step.input_tokens).sum()
}
#[must_use]
pub fn output_tokens(&self) -> u64 {
self.steps.iter().map(|step| step.output_tokens).sum()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum ToolLoopError {
Model(TextModelError),
StepBudgetExhausted { steps: Vec<StepUsage> },
TokenBudgetExhausted { used: u64, steps: Vec<StepUsage> },
InvalidToolSchema { tool: String, reason: String },
}
impl std::fmt::Display for ToolLoopError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let scrub = crate::logging::scrub_text;
match self {
Self::Model(error) => write!(f, "the text model failed: {error}"),
Self::StepBudgetExhausted { steps } => write!(
f,
"the tool loop exhausted its step budget after {} model calls",
steps.len()
),
Self::TokenBudgetExhausted { used, steps } => write!(
f,
"the tool loop spent {used} tokens over {} model calls, past its token budget",
steps.len()
),
Self::InvalidToolSchema { tool, reason } => write!(
f,
"tool {tool:?} has an invalid argument schema: {}",
scrub(reason)
),
}
}
}
impl std::error::Error for ToolLoopError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Model(error) => Some(error),
_ => None,
}
}
}
pub async fn run_tool_loop(
model: &dyn TextModel,
mut prompt: Prompt,
executor: &dyn ToolExecutor,
budget: ToolBudget,
) -> Result<ToolLoopOutcome, ToolLoopError> {
preflight(model, &prompt)?;
let mut steps: Vec<StepUsage> = Vec::new();
let mut step_count: u32 = 0;
let mut used_tokens: u64 = 0;
loop {
if step_count >= budget.max_steps {
return Err(ToolLoopError::StepBudgetExhausted { steps });
}
let completion = model
.complete(&prompt)
.await
.map_err(ToolLoopError::Model)?;
step_count += 1;
match settle_step(
completion,
&mut prompt.messages,
&mut steps,
&mut used_tokens,
&budget,
)
.after
{
AfterStep::Final(outcome) => return Ok(outcome),
AfterStep::Exhausted(error) => return Err(error),
AfterStep::Calls(calls) => {
let mut results = Vec::with_capacity(calls.len());
for call in &calls {
results.push(execute_call(&prompt.tools, executor, call).await);
}
prompt.messages.push(Turn::tool_results(results));
}
}
}
}
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub enum ToolLoopEvent {
Delta {
step: usize,
delta: TextDelta,
},
ToolResult {
step: usize,
call: ToolCall,
result: ToolResult,
},
StepFinished {
step: usize,
usage: StepUsage,
},
Done(ToolLoopOutcome),
}
type EventSender = futures_channel::mpsc::Sender<Result<ToolLoopEvent, ToolLoopError>>;
pub fn run_tool_loop_stream<'a>(
model: &'a dyn TextModel,
prompt: Prompt,
executor: &'a dyn ToolExecutor,
budget: ToolBudget,
) -> BoxStream<'a, Result<ToolLoopEvent, ToolLoopError>> {
pump(move |tx| stream_loop(model, prompt, executor, budget, tx))
}
async fn stream_loop(
model: &dyn TextModel,
mut prompt: Prompt,
executor: &dyn ToolExecutor,
budget: ToolBudget,
mut tx: EventSender,
) {
if let Err(error) = preflight(model, &prompt) {
let _ = tx.send(Err(error)).await;
return;
}
let mut steps: Vec<StepUsage> = Vec::new();
let mut step_count: u32 = 0;
let mut used_tokens: u64 = 0;
loop {
if step_count >= budget.max_steps {
let _ = tx
.send(Err(ToolLoopError::StepBudgetExhausted { steps }))
.await;
return;
}
let step = usize::try_from(step_count).unwrap_or(usize::MAX);
step_count += 1;
let Some(completion) = stream_step(model, &prompt, step, &mut tx).await else {
return;
};
let settled = settle_step(
completion,
&mut prompt.messages,
&mut steps,
&mut used_tokens,
&budget,
);
if tx
.send(Ok(ToolLoopEvent::StepFinished {
step,
usage: settled.usage,
}))
.await
.is_err()
{
return;
}
match settled.after {
AfterStep::Final(outcome) => {
let _ = tx.send(Ok(ToolLoopEvent::Done(outcome))).await;
return;
}
AfterStep::Exhausted(error) => {
let _ = tx.send(Err(error)).await;
return;
}
AfterStep::Calls(calls) => {
let mut results = Vec::with_capacity(calls.len());
for call in &calls {
let result = execute_call(&prompt.tools, executor, call).await;
results.push(result.clone());
if tx
.send(Ok(ToolLoopEvent::ToolResult {
step,
call: call.clone(),
result,
}))
.await
.is_err()
{
return;
}
}
prompt.messages.push(Turn::tool_results(results));
}
}
}
}
async fn stream_step(
model: &dyn TextModel,
prompt: &Prompt,
step: usize,
tx: &mut EventSender,
) -> Option<Completion> {
let mut builder = CompletionBuilder::default();
let mut deltas = model.stream(prompt);
while let Some(delta) = deltas.next().await {
match delta {
Ok(delta) => {
builder.push(&delta);
if tx
.send(Ok(ToolLoopEvent::Delta { step, delta }))
.await
.is_err()
{
return None;
}
}
Err(error) => {
let _ = tx.send(Err(ToolLoopError::Model(error))).await;
return None;
}
}
}
Some(builder.finish())
}
enum AfterStep {
Final(ToolLoopOutcome),
Exhausted(ToolLoopError),
Calls(Vec<ToolCall>),
}
struct Settled {
usage: StepUsage,
after: AfterStep,
}
fn settle_step(
completion: Completion,
messages: &mut Vec<Turn>,
steps: &mut Vec<StepUsage>,
used_tokens: &mut u64,
budget: &ToolBudget,
) -> Settled {
let usage = step_usage(&completion);
steps.push(usage.clone());
*used_tokens = used_tokens.saturating_add(
completion
.input_tokens
.saturating_add(completion.output_tokens),
);
if completion.tool_calls.is_empty() {
return Settled {
usage,
after: AfterStep::Final(ToolLoopOutcome {
completion,
messages: std::mem::take(messages),
steps: std::mem::take(steps),
}),
};
}
if let Some(limit) = budget.max_total_tokens
&& *used_tokens > limit
{
return Settled {
usage,
after: AfterStep::Exhausted(ToolLoopError::TokenBudgetExhausted {
used: *used_tokens,
steps: std::mem::take(steps),
}),
};
}
let calls = completion.tool_calls;
messages.push(Turn::assistant_tool_calls(completion.text, calls.clone()));
Settled {
usage,
after: AfterStep::Calls(calls),
}
}
fn preflight(model: &dyn TextModel, prompt: &Prompt) -> Result<(), ToolLoopError> {
prompt.check_images().map_err(ToolLoopError::Model)?;
if !prompt.tools.is_empty() && !model.supports(prompt.tier, Capability::Tools) {
return Err(ToolLoopError::Model(TextModelError::Unsupported(
Capability::Tools,
)));
}
if prompt.has_images() && !model.supports(prompt.tier, Capability::Images) {
return Err(ToolLoopError::Model(TextModelError::Unsupported(
Capability::Images,
)));
}
for tool in &prompt.tools {
validate_tool_schema(&tool.name, &tool.parameters)?;
}
Ok(())
}
fn step_usage(completion: &Completion) -> StepUsage {
StepUsage {
input_tokens: completion.input_tokens,
output_tokens: completion.output_tokens,
cached_input_tokens: completion.cached_input_tokens,
tool_calls: completion.tool_calls.len(),
}
}
async fn execute_call(
tools: &[ToolSpec],
executor: &dyn ToolExecutor,
call: &ToolCall,
) -> ToolResult {
let Some(spec) = tools.iter().find(|tool| tool.name == call.name) else {
return ToolResult::error(
call.id.clone(),
format!("no tool named {:?} was offered", call.name),
);
};
if !call.arguments.is_object() {
return ToolResult::error(
call.id.clone(),
format!("arguments for tool {:?} must be a JSON object", call.name),
);
}
if let Err(violation) = validate_instance(&spec.parameters, &call.arguments) {
return ToolResult::error(
call.id.clone(),
format!("arguments do not match tool {:?}: {violation}", call.name),
);
}
match executor.execute(call).await {
Ok(content) => ToolResult::ok(call.id.clone(), content),
Err(message) => ToolResult::error(call.id.clone(), message),
}
}
const ANNOTATIONS: &[&str] = &[
"title",
"description",
"default",
"examples",
"$schema",
"$id",
"$comment",
"format",
];
const TYPE_NAMES: &[&str] = &[
"object", "array", "string", "number", "integer", "boolean", "null",
];
fn invalid_schema(tool: &str, reason: impl Into<String>) -> ToolLoopError {
ToolLoopError::InvalidToolSchema {
tool: tool.to_owned(),
reason: reason.into(),
}
}
fn validate_tool_schema(tool: &str, schema: &Value) -> Result<(), ToolLoopError> {
validate_schema(schema).map_err(|reason| invalid_schema(tool, reason))
}
pub(crate) fn validate_schema(schema: &Value) -> Result<(), String> {
let Some(object) = schema.as_object() else {
return Err("the schema is not a JSON object".to_owned());
};
for (keyword, value) in object {
match keyword.as_str() {
"type" => check_type_keyword(value)?,
"properties" => check_properties(value)?,
"required" => check_required(value)?,
"additionalProperties" => check_additional_properties(value)?,
"items" => validate_schema(value)?,
"enum" => {
if !value.is_array() {
return Err("`enum` must be an array".to_owned());
}
}
"const" => {}
"minimum" | "maximum" | "exclusiveMinimum" | "exclusiveMaximum" => {
if !value.is_number() {
return Err(format!("`{keyword}` must be a number"));
}
}
"minLength" | "maxLength" | "minItems" | "maxItems" => {
if value.as_u64().is_none() {
return Err(format!("`{keyword}` must be a non-negative integer"));
}
}
"anyOf" | "oneOf" | "allOf" => check_schema_list(keyword, value)?,
other if ANNOTATIONS.contains(&other) => {}
other => return Err(format!("the `{other}` keyword is not supported")),
}
}
Ok(())
}
fn check_type_keyword(value: &Value) -> Result<(), String> {
let names: Vec<&str> = match value {
Value::String(name) => vec![name.as_str()],
Value::Array(items) => {
if !items.iter().all(Value::is_string) {
return Err("`type` must name strings when given as an array".to_owned());
}
items.iter().filter_map(Value::as_str).collect()
}
_ => return Err("`type` must be a string or an array of strings".to_owned()),
};
if names.is_empty() || !names.iter().all(|name| TYPE_NAMES.contains(name)) {
return Err("`type` names an unknown JSON type".to_owned());
}
Ok(())
}
fn check_properties(value: &Value) -> Result<(), String> {
let Some(properties) = value.as_object() else {
return Err("`properties` must be an object".to_owned());
};
for subschema in properties.values() {
validate_schema(subschema)?;
}
Ok(())
}
fn check_required(value: &Value) -> Result<(), String> {
let Some(names) = value.as_array() else {
return Err("`required` must be an array".to_owned());
};
if !names.iter().all(Value::is_string) {
return Err("`required` must name strings".to_owned());
}
Ok(())
}
fn check_additional_properties(value: &Value) -> Result<(), String> {
match value {
Value::Bool(_) => Ok(()),
Value::Object(_) => validate_schema(value),
_ => Err("`additionalProperties` must be a boolean or a schema".to_owned()),
}
}
fn check_schema_list(keyword: &str, value: &Value) -> Result<(), String> {
let Some(schemas) = value.as_array() else {
return Err(format!("`{keyword}` must be an array of schemas"));
};
if schemas.is_empty() {
return Err(format!("`{keyword}` must not be empty"));
}
for subschema in schemas {
validate_schema(subschema)?;
}
Ok(())
}
pub(crate) fn validate_instance(schema: &Value, instance: &Value) -> Result<(), String> {
let Some(object) = schema.as_object() else {
return Err("the schema is not an object".to_owned());
};
if let Some(subschemas) = object.get("allOf").and_then(Value::as_array) {
for subschema in subschemas {
validate_instance(subschema, instance)?;
}
}
if let Some(subschemas) = object.get("anyOf").and_then(Value::as_array)
&& !subschemas
.iter()
.any(|subschema| validate_instance(subschema, instance).is_ok())
{
return Err("the value matches none of the `anyOf` schemas".to_owned());
}
if let Some(subschemas) = object.get("oneOf").and_then(Value::as_array) {
let matches = subschemas
.iter()
.filter(|subschema| validate_instance(subschema, instance).is_ok())
.count();
if matches != 1 {
return Err(format!(
"the value matches {matches} of the `oneOf` schemas, not exactly one"
));
}
}
if let Some(constant) = object.get("const")
&& constant != instance
{
return Err("the value is not the required `const`".to_owned());
}
if let Some(allowed) = object.get("enum").and_then(Value::as_array)
&& !allowed.contains(instance)
{
return Err("the value is none of the `enum` values".to_owned());
}
if let Some(type_keyword) = object.get("type")
&& !matches_type(type_keyword, instance)
{
return Err("the value is not of the declared `type`".to_owned());
}
check_number_bounds(object, instance)?;
check_string_bounds(object, instance)?;
check_array_bounds(object, instance)?;
check_object_shape(object, instance)
}
fn check_number_bounds(object: &Map<String, Value>, instance: &Value) -> Result<(), String> {
let Some(number) = instance.as_f64() else {
return Ok(());
};
if let Some(minimum) = object.get("minimum").and_then(Value::as_f64)
&& number < minimum
{
return Err(format!("the number is below the `minimum` {minimum}"));
}
if let Some(minimum) = object.get("exclusiveMinimum").and_then(Value::as_f64)
&& number <= minimum
{
return Err(format!(
"the number is not above the `exclusiveMinimum` {minimum}"
));
}
if let Some(maximum) = object.get("maximum").and_then(Value::as_f64)
&& number > maximum
{
return Err(format!("the number is above the `maximum` {maximum}"));
}
if let Some(maximum) = object.get("exclusiveMaximum").and_then(Value::as_f64)
&& number >= maximum
{
return Err(format!(
"the number is not below the `exclusiveMaximum` {maximum}"
));
}
Ok(())
}
fn check_string_bounds(object: &Map<String, Value>, instance: &Value) -> Result<(), String> {
let Some(text) = instance.as_str() else {
return Ok(());
};
let length = u64::try_from(text.chars().count()).unwrap_or(u64::MAX);
if let Some(minimum) = object.get("minLength").and_then(Value::as_u64)
&& length < minimum
{
return Err(format!("the string is shorter than `minLength` {minimum}"));
}
if let Some(maximum) = object.get("maxLength").and_then(Value::as_u64)
&& length > maximum
{
return Err(format!("the string is longer than `maxLength` {maximum}"));
}
Ok(())
}
fn check_array_bounds(object: &Map<String, Value>, instance: &Value) -> Result<(), String> {
let Some(items) = instance.as_array() else {
return Ok(());
};
let length = u64::try_from(items.len()).unwrap_or(u64::MAX);
if let Some(minimum) = object.get("minItems").and_then(Value::as_u64)
&& length < minimum
{
return Err(format!("the array has fewer than `minItems` {minimum}"));
}
if let Some(maximum) = object.get("maxItems").and_then(Value::as_u64)
&& length > maximum
{
return Err(format!("the array has more than `maxItems` {maximum}"));
}
if let Some(item_schema) = object.get("items") {
for item in items {
validate_instance(item_schema, item)
.map_err(|reason| format!("an array item is invalid: {reason}"))?;
}
}
Ok(())
}
fn check_object_shape(object: &Map<String, Value>, instance: &Value) -> Result<(), String> {
let Some(fields) = instance.as_object() else {
return Ok(());
};
if let Some(required) = object.get("required").and_then(Value::as_array) {
for name in required.iter().filter_map(Value::as_str) {
if !fields.contains_key(name) {
return Err(format!("the required property {name:?} is missing"));
}
}
}
let properties = object.get("properties").and_then(Value::as_object);
let additional = object.get("additionalProperties");
for (key, field) in fields {
if let Some(property_schema) = properties.and_then(|properties| properties.get(key)) {
validate_instance(property_schema, field)
.map_err(|reason| format!("the property {key:?} is invalid: {reason}"))?;
continue;
}
match additional {
Some(Value::Bool(false)) => {
return Err(format!(
"the property {key:?} is not allowed by `additionalProperties`"
));
}
Some(schema @ Value::Object(_)) => {
validate_instance(schema, field).map_err(|reason| {
format!("the additional property {key:?} is invalid: {reason}")
})?;
}
_ => {}
}
}
Ok(())
}
fn matches_type(type_keyword: &Value, instance: &Value) -> bool {
match type_keyword {
Value::String(name) => matches_one_type(name, instance),
Value::Array(names) => names.iter().any(|name| {
name.as_str()
.is_some_and(|name| matches_one_type(name, instance))
}),
_ => false,
}
}
fn matches_one_type(name: &str, instance: &Value) -> bool {
match name {
"object" => instance.is_object(),
"array" => instance.is_array(),
"string" => instance.is_string(),
"number" => instance.is_number(),
"integer" => instance
.as_f64()
.is_some_and(|number| number.fract() == 0.0),
"boolean" => instance.is_boolean(),
"null" => instance.is_null(),
_ => false,
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn a_known_schema_is_accepted_and_an_unknown_keyword_is_named() {
let good = json!({
"type": "object",
"description": "a note",
"properties": { "q": { "type": "string", "minLength": 1 } },
"required": ["q"],
"additionalProperties": false,
});
validate_tool_schema("lookup", &good).expect("known keywords are accepted");
let bad = json!({ "type": "object", "pattern": "^a" });
let error = validate_tool_schema("lookup", &bad).unwrap_err();
match error {
ToolLoopError::InvalidToolSchema { tool, reason } => {
assert_eq!(tool, "lookup");
assert!(reason.contains("pattern"), "{reason}");
}
other => panic!("expected an invalid-schema error, got {other:?}"),
}
}
#[test]
fn a_nested_unknown_keyword_is_reached() {
let schema = json!({
"type": "object",
"properties": { "q": { "$ref": "#/defs/thing" } },
});
let error = validate_tool_schema("lookup", &schema).unwrap_err();
assert!(error.to_string().contains("$ref"), "{error}");
}
#[test]
fn annotations_are_ignored_but_a_malformed_known_keyword_is_refused() {
let annotated = json!({ "type": "string", "format": "email", "title": "Q" });
validate_tool_schema("t", &annotated).expect("annotations are ignored");
let malformed = json!({ "type": "list-of-things" });
assert!(validate_tool_schema("t", &malformed).is_err());
}
#[test]
fn instance_validation_covers_types_bounds_and_objects() {
let schema = json!({
"type": "object",
"additionalProperties": false,
"required": ["name", "age"],
"properties": {
"name": { "type": "string", "minLength": 1, "maxLength": 5 },
"age": { "type": "integer", "minimum": 0, "maximum": 120 },
"tags": { "type": "array", "items": { "enum": ["a", "b"] } },
},
});
let good = json!({ "name": "al", "age": 30, "tags": ["a"] });
assert_eq!(validate_instance(&schema, &good), Ok(()));
let short = json!({ "name": "", "age": 30 });
assert!(validate_instance(&schema, &short).is_err());
let wrong_type = json!({ "name": "al", "age": "old" });
assert!(validate_instance(&schema, &wrong_type).is_err());
let below = json!({ "name": "al", "age": -1 });
assert!(validate_instance(&schema, &below).is_err());
let extra = json!({ "name": "al", "age": 30, "extra": true });
assert!(validate_instance(&schema, &extra).is_err());
let bad_item = json!({ "name": "al", "age": 30, "tags": ["c"] });
assert!(validate_instance(&schema, &bad_item).is_err());
let missing = json!({ "name": "al" });
assert!(validate_instance(&schema, &missing).is_err());
}
#[test]
fn combinators_are_enforced() {
let any_of = json!({ "anyOf": [{ "type": "string" }, { "type": "integer" }] });
assert_eq!(validate_instance(&any_of, &json!("x")), Ok(()));
assert!(validate_instance(&any_of, &json!(true)).is_err());
let one_of = json!({ "oneOf": [{ "type": "number" }, { "type": "integer" }] });
assert!(validate_instance(&one_of, &json!(3)).is_err());
assert_eq!(validate_instance(&one_of, &json!(3.5)), Ok(()));
let all_of = json!({
"allOf": [
{ "type": "integer" },
{ "minimum": 10 },
],
});
assert!(validate_instance(&all_of, &json!(3)).is_err());
assert_eq!(validate_instance(&all_of, &json!(12)), Ok(()));
}
#[test]
fn const_and_enum_are_enforced() {
let schema = json!({ "const": "fixed" });
assert_eq!(validate_instance(&schema, &json!("fixed")), Ok(()));
assert!(validate_instance(&schema, &json!("other")).is_err());
let schema = json!({ "enum": [1, 2] });
assert_eq!(validate_instance(&schema, &json!(2)), Ok(()));
assert!(validate_instance(&schema, &json!(3)).is_err());
}
}