use std::{
collections::{BTreeMap, BTreeSet},
sync::Arc,
time::Duration,
};
use jsonschema::Validator;
use runtime_extension_api::{
AgentDefinition, CompletionMode, ExtensionContext, ObserverEvent, ToolContext, ToolInvocation,
ToolOutcome,
};
use runtime_kernel_api::{KernelEvent, KernelEventSink, KernelFailure};
use runtime_ports::{ExecutionSessions, ModelToolDefinition, OperationContext};
use runtime_types::ToolCallId;
use serde_json::Value;
use crate::facades::{PortInteractionFacade, PortSubagentFacade, PortWorkspaceFacade};
struct RegisteredTool {
tool: Arc<dyn runtime_extension_api::Tool>,
validator: Validator,
}
pub(crate) struct ToolRuntime {
tools: BTreeMap<String, RegisteredTool>,
definition: Arc<AgentDefinition>,
max_output_bytes: usize,
}
impl ToolRuntime {
pub(crate) fn new(
definition: Arc<AgentDefinition>,
max_output_bytes: usize,
granted_capabilities: &BTreeSet<String>,
) -> Result<Self, KernelFailure> {
let mut tools = BTreeMap::new();
for tool in &definition.tools {
let descriptor = tool.descriptor();
if descriptor.name.is_empty() || descriptor.name.len() > 128 {
return Err(KernelFailure::new(
"TOOL_NAME_INVALID",
"tool name must contain 1..=128 bytes",
));
}
if tools.contains_key(&descriptor.name) {
return Err(KernelFailure::new(
"TOOL_DUPLICATE",
format!("duplicate tool {}", descriptor.name),
));
}
let authorized = granted_capabilities.contains("*")
|| descriptor
.required_capabilities
.iter()
.all(|capability| granted_capabilities.contains(capability));
if !authorized {
continue;
}
let validator =
jsonschema::validator_for(&descriptor.input_schema).map_err(|error| {
KernelFailure::new(
"TOOL_SCHEMA_INVALID",
format!("{}: {error}", descriptor.name),
)
})?;
tools.insert(
descriptor.name,
RegisteredTool {
tool: tool.clone(),
validator,
},
);
}
if let CompletionMode::RequiredTool(name) = &definition.completion_mode
&& !tools.contains_key(name)
{
return Err(KernelFailure::new(
"COMPLETION_TOOL_UNAUTHORIZED",
format!("required completion tool {name} is unavailable under caller policy"),
));
}
Ok(Self {
tools,
definition,
max_output_bytes,
})
}
pub(crate) fn model_definitions(&self) -> Vec<ModelToolDefinition> {
self.tools
.values()
.map(|registered| {
let descriptor = registered.tool.descriptor();
ModelToolDefinition {
name: descriptor.name,
description: descriptor.description,
input_schema: descriptor.input_schema,
}
})
.collect()
}
pub(crate) async fn execute(
&self,
operation: &OperationContext,
sessions: &ExecutionSessions,
events: &Arc<dyn KernelEventSink>,
call_id: &str,
name: &str,
arguments: Value,
) -> Result<ToolOutcome, KernelFailure> {
let registered = self.tools.get(name).ok_or_else(|| {
KernelFailure::new(
"TOOL_NOT_FOUND",
format!("model requested unknown tool {name}"),
)
})?;
if let Err(error) = registered.validator.validate(&arguments) {
return Ok(ToolOutcome::failure(format!(
"invalid tool arguments: {error}"
)));
}
let call_id = ToolCallId::new(call_id)
.map_err(|error| KernelFailure::new("TOOL_CALL_ID_INVALID", error.to_string()))?;
let invocation = ToolInvocation {
execution_id: operation.execution_id.clone(),
call_id: call_id.clone(),
descriptor: registered.tool.descriptor(),
arguments: arguments.clone(),
};
for guard in &self.definition.guards {
let remaining = operation.remaining().map_err(port_error)?;
tokio::time::timeout(
remaining.min(Duration::from_secs(5)),
guard.authorize(&invocation),
)
.await
.map_err(|_| {
KernelFailure::new("GUARD_TIMEOUT", format!("guard timed out for {name}"))
})?
.map_err(|error| KernelFailure::new("TOOL_DENIED", error.to_string()))?;
}
events
.emit(KernelEvent::ToolStarted {
call_id: call_id.to_string(),
name: name.to_string(),
})
.await?;
let context = ToolContext {
extension: ExtensionContext {
execution_id: invocation.execution_id.clone(),
deadline: operation.deadline,
cancellation: operation.cancellation.clone(),
},
call_id: call_id.clone(),
workspace: Arc::new(PortWorkspaceFacade::new(
operation.clone(),
sessions.workspace.clone(),
)),
interaction: Arc::new(PortInteractionFacade::new(
operation.clone(),
call_id.to_string(),
sessions.interaction.clone(),
)),
subagent: Arc::new(PortSubagentFacade::new(
operation.clone(),
call_id.to_string(),
sessions.subagent.clone(),
)),
};
let remaining = operation.remaining().map_err(port_error)?;
let mut outcome =
tokio::time::timeout(remaining, registered.tool.invoke(context, arguments))
.await
.map_err(|_| KernelFailure::new("TOOL_TIMEOUT", format!("tool {name} timed out")))?
.unwrap_or_else(|error| ToolOutcome::failure(error.to_string()));
outcome.content = truncate_utf8(outcome.content, self.max_output_bytes);
events
.emit(KernelEvent::ToolCompleted {
call_id: call_id.to_string(),
name: name.to_string(),
failed: outcome.failed,
})
.await?;
for observer in &self.definition.observers {
let event = ObserverEvent::ToolCompleted {
execution_id: invocation.execution_id.clone(),
call_id: call_id.clone(),
name: name.to_string(),
failed: outcome.failed,
};
let _ = tokio::time::timeout(Duration::from_secs(2), observer.observe(event)).await;
}
Ok(outcome)
}
}
fn port_error(error: runtime_ports::PortFailure) -> KernelFailure {
KernelFailure {
code: error.code,
message: error.message,
retryable: error.retryable,
commit: error.commit,
}
}
fn truncate_utf8(value: String, max: usize) -> String {
if value.len() <= max {
return value;
}
let marker = format!(
"\n[tool output truncated; middle omitted; original bytes={}]\n",
value.len()
);
if max <= marker.len() {
let mut boundary = max;
while !value.is_char_boundary(boundary) {
boundary = boundary.saturating_sub(1);
}
return value[..boundary].to_string();
}
let available = max.saturating_sub(marker.len());
let mut head_end = available / 3;
while !value.is_char_boundary(head_end) {
head_end = head_end.saturating_sub(1);
}
let mut tail_start = value
.len()
.saturating_sub(available.saturating_sub(head_end));
while tail_start < value.len() && !value.is_char_boundary(tail_start) {
tail_start = tail_start.saturating_add(1);
}
format!("{}{marker}{}", &value[..head_end], &value[tail_start..])
}
#[cfg(test)]
mod tests {
use super::truncate_utf8;
#[test]
fn oversized_unicode_observation_preserves_head_and_tail_within_limit() {
let value = format!("HEAD-{}-TAIL", "数据".repeat(100));
let truncated = truncate_utf8(value, 100);
assert!(truncated.starts_with("HEAD"));
assert!(truncated.ends_with("TAIL"));
assert!(truncated.contains("middle omitted"));
assert!(truncated.len() <= 100);
assert!(std::str::from_utf8(truncated.as_bytes()).is_ok());
}
}