scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-17 23:52:34 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1
fabro/lib/components/fabro-agent/src/tool_execution.rs
Bryan Helmkamp 623e8e1c25
Speak the driver's exec vocabulary instead of mirroring it 
Fabro kept its own ExecResult, streaming request and result, output
capture stats, stdio process types, and an Error::Exec variant, each a
field-for-field copy of a sandbox-driver type with a translation layer
between them. Every command a tool, a stage, or a hook ran crossed that
layer twice.

The driver's types are now the ones fabro uses. SandboxExec applies
fabro's policy to an ExecSpec (stop grace, the run's working directory,
the explicit-env filter, the Bash helper's BASH_ENV blank winning over a
caller value) and returns the driver's ExecResult and
ExecStreamingResult as they are. Callers that stream build an ExecSpec
and ExecControls; the buffered exec_command keeps its signature.
ExecResultExt adds fabro's reading of a result: the event-facing
duration, the exit code only when the command exited on its own, the
redacted output tail, and the ExecFailure a non-zero exit becomes. The
three-way termination collapse the run events use lives in one function,
command_termination, called where events are built.

Error::Exec and the git-shaped stderr hint table are gone; a failed
command is the driver's ExecFailure, whose Display carries the label and
the classified metadata and never the raw output. OutputCaptureStats
moves to fabro-agent, whose tool output accounting it belongs to, and
converts from the driver's CaptureStats at the exec boundary. The stdio
process the ACP transport drives is the driver's own, so the cancel-token
bridge and StderrCollector go too.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-10 16:11:45 -06:00

1603 lines
52 KiB
Rust
Raw Blame History

use std::borrow::Cow;
use std::sync::Arc;
use fabro_types::{tool_call_arguments, tool_result_from_json};
use futures::future;
use lithos_llm::types::{ContentPart, ToolCall, ToolDefinitionKind, ToolInput, ToolResult};
use tokio_util::sync::CancellationToken;
use tracing::debug;
use crate::config::{SessionOptions, ToolHookCallback, ToolHookDecision};
use crate::event::{Emitter, SessionBoundEmitter};
use crate::question_tools::{self, AgentToolRuntime, is_question_tool};
use crate::sandbox::RunSandbox;
use crate::session::ToolEnvProvider;
use crate::tool_registry::{AgentEventEmitter, RegisteredTool, ToolContext, ToolRegistry};
use crate::truncation::{
MAX_RETAINED_TOOL_OUTPUT_BYTES, OutputCaptureStats, preview_tool_output, serialized_json_bytes,
truncate_tool_output,
};
use crate::types::AgentEvent;
/// Execute tool calls, choosing parallel or sequential based on `parallel`
/// flag.
#[allow(
clippy::too_many_arguments,
reason = "Tool dispatch needs the shared runtime handles and call list together."
)]
pub async fn execute_tool_calls(
tool_calls: &[ToolCall],
parallel: bool,
registry: &ToolRegistry,
env: Arc<RunSandbox>,
tool_hooks: Option<&Arc<dyn ToolHookCallback>>,
cancel_token: &CancellationToken,
config: &SessionOptions,
emitter: &Emitter,
session_id: &str,
root_session_id: &str,
tool_env_provider: Option<&Arc<dyn ToolEnvProvider>>,
agent_tool_runtime: &AgentToolRuntime,
) -> Vec<ToolResult> {
if tool_calls.iter().any(|tc| is_question_tool(&tc.name)) {
return execute_question_tool_round(
tool_calls,
registry,
env,
tool_hooks,
cancel_token,
config,
emitter,
session_id,
root_session_id,
tool_env_provider,
agent_tool_runtime,
)
.await;
}
if parallel && tool_calls.len() > 1 {
execute_tool_calls_parallel(
tool_calls,
registry,
env,
tool_hooks,
cancel_token,
config,
emitter,
session_id,
root_session_id,
tool_env_provider,
agent_tool_runtime,
)
.await
} else {
execute_tool_calls_sequential(
tool_calls,
registry,
env,
tool_hooks,
cancel_token,
config,
emitter,
session_id,
root_session_id,
tool_env_provider,
agent_tool_runtime,
)
.await
}
}
#[allow(
clippy::too_many_arguments,
reason = "Sequential execution threads the runtime handles through each tool call."
)]
async fn execute_tool_calls_sequential(
tool_calls: &[ToolCall],
registry: &ToolRegistry,
env: Arc<RunSandbox>,
tool_hooks: Option<&Arc<dyn ToolHookCallback>>,
cancel_token: &CancellationToken,
config: &SessionOptions,
emitter: &Emitter,
session_id: &str,
root_session_id: &str,
tool_env_provider: Option<&Arc<dyn ToolEnvProvider>>,
agent_tool_runtime: &AgentToolRuntime,
) -> Vec<ToolResult> {
let mut results = Vec::new();
for tc in tool_calls {
if cancel_token.is_cancelled() {
results.push(error_result(&tc.id, "Cancelled"));
continue;
}
let result = execute_and_emit_one_tool_with_runtime(
tc,
registry,
env.clone(),
tool_hooks,
cancel_token.child_token(),
config,
emitter,
session_id,
root_session_id,
tool_env_provider,
agent_tool_runtime,
)
.await;
results.push(result);
}
results
}
#[allow(
clippy::too_many_arguments,
reason = "Parallel execution threads the runtime handles into each spawned tool task."
)]
async fn execute_tool_calls_parallel(
tool_calls: &[ToolCall],
registry: &ToolRegistry,
env: Arc<RunSandbox>,
tool_hooks: Option<&Arc<dyn ToolHookCallback>>,
cancel_token: &CancellationToken,
config: &SessionOptions,
emitter: &Emitter,
session_id: &str,
root_session_id: &str,
tool_env_provider: Option<&Arc<dyn ToolEnvProvider>>,
agent_tool_runtime: &AgentToolRuntime,
) -> Vec<ToolResult> {
let tool_env_provider = tool_env_provider.cloned();
let agent_tool_runtime = agent_tool_runtime.clone();
let futures: Vec<_> = tool_calls
.iter()
.map(|tc| {
let emitter = emitter.clone();
let env = env.clone();
let config = config.clone();
let cancel_token = cancel_token.clone();
let tc = tc.clone();
let session_id = session_id.to_owned();
let root_session_id = root_session_id.to_owned();
let tool_hooks = tool_hooks.cloned();
let tool_env_provider = tool_env_provider.clone();
let agent_tool_runtime = agent_tool_runtime.clone();
let access_denial = config.tool_access_denial_reason(&tc.name);
// Look up the tool before spawning since ToolRegistry is not Send.
let registered_tool = if access_denial.is_none() {
registry.get(&tc.name).cloned()
} else {
None
};
async move {
execute_and_emit_one_tool_with_lookup(
&tc,
registered_tool.as_ref(),
access_denial,
env,
tool_hooks.as_ref(),
cancel_token.child_token(),
&config,
&emitter,
&session_id,
&root_session_id,
tool_env_provider.as_ref(),
&agent_tool_runtime,
)
.await
}
})
.collect();
future::join_all(futures).await
}
#[allow(
clippy::too_many_arguments,
reason = "Question-tool round handling needs the same execution context as normal tool dispatch."
)]
async fn execute_question_tool_round(
tool_calls: &[ToolCall],
registry: &ToolRegistry,
env: Arc<RunSandbox>,
tool_hooks: Option<&Arc<dyn ToolHookCallback>>,
cancel_token: &CancellationToken,
config: &SessionOptions,
emitter: &Emitter,
session_id: &str,
root_session_id: &str,
tool_env_provider: Option<&Arc<dyn ToolEnvProvider>>,
agent_tool_runtime: &AgentToolRuntime,
) -> Vec<ToolResult> {
let first_question_index = tool_calls
.iter()
.position(|tc| is_question_tool(&tc.name))
.expect("question-tool round should contain a question tool");
let mut results = Vec::with_capacity(tool_calls.len());
for (index, tc) in tool_calls.iter().enumerate() {
if cancel_token.is_cancelled() {
results.push(error_result(&tc.id, "Cancelled"));
continue;
}
if index == first_question_index {
results.push(
execute_and_emit_one_tool_with_runtime(
tc,
registry,
env.clone(),
tool_hooks,
cancel_token.child_token(),
config,
emitter,
session_id,
root_session_id,
tool_env_provider,
agent_tool_runtime,
)
.await,
);
} else if is_question_tool(&tc.name) {
results.push(error_tool_result_with_events(
tc,
emitter,
session_id,
config,
"Only one human-question tool call may be used in a tool round. Combine all questions into a single questions[] batch and call the question tool once.",
));
} else {
results.push(error_tool_result_with_events(
tc,
emitter,
session_id,
config,
"This tool call was not executed because human-question tools must run alone in a tool round. Retry non-question tools in a later round after the user answers.",
));
}
}
results
}
fn error_tool_result_with_events(
tc: &ToolCall,
emitter: &Emitter,
session_id: &str,
config: &SessionOptions,
message: &str,
) -> ToolResult {
emit_tool_call_started(emitter, session_id, tc);
finish_error_result(tc, emitter, session_id, config, message)
}
/// Bound, emit, and truncate an error result for a tool call whose
/// started event was already emitted.
fn finish_error_result(
tc: &ToolCall,
emitter: &Emitter,
session_id: &str,
config: &SessionOptions,
message: &str,
) -> ToolResult {
let retained = retain_tool_result(error_result(&tc.id, message), None);
emit_tool_call_result(
emitter,
session_id,
tc,
&retained.result,
retained.output_stats,
);
truncate_tool_result(&retained.result, &tc.name, config)
}
/// A tool result carrying one error message.
fn error_result(tool_call_id: &str, message: impl Into<String>) -> ToolResult {
tool_result_from_json(
tool_call_id,
serde_json::Value::String(message.into()),
true,
)
}
/// A successful tool result carrying one output value.
fn success_result(tool_call_id: &str, output: serde_json::Value) -> ToolResult {
tool_result_from_json(tool_call_id, output, false)
}
/// The single JSON value a tool result carries: a string for text output.
fn result_output(result: &ToolResult) -> serde_json::Value {
fabro_types::tool_result_to_json(result)
}
fn emit_tool_call_started(emitter: &Emitter, session_id: &str, tc: &ToolCall) {
emitter.emit(session_id.to_owned(), AgentEvent::ToolCallStarted {
tool_name: tc.name.clone(),
tool_call_id: tc.id.clone(),
arguments: tool_call_arguments(tc),
});
}
fn emit_tool_call_result(
emitter: &Emitter,
session_id: &str,
tc: &ToolCall,
result: &ToolResult,
output_stats: OutputCaptureStats,
) {
let output = result_output(result);
emitter.emit(session_id.to_owned(), AgentEvent::ToolCallOutputDelta {
delta: output.to_string(),
});
emitter.emit(session_id.to_owned(), AgentEvent::ToolCallCompleted {
tool_name: tc.name.clone(),
tool_call_id: tc.id.clone(),
output,
is_error: result.is_error,
output_bytes_observed: output_stats.observed_bytes,
output_bytes_retained: output_stats.retained_bytes,
output_bytes_omitted: output_stats.omitted_bytes,
});
}
/// Execute a single tool call with event emission and output truncation.
#[allow(
clippy::too_many_arguments,
reason = "Single-tool execution needs the tool, runtime handles, and emission context."
)]
pub async fn execute_and_emit_one_tool(
tc: &ToolCall,
registry: &ToolRegistry,
env: Arc<RunSandbox>,
tool_hooks: Option<&Arc<dyn ToolHookCallback>>,
cancel_token: CancellationToken,
config: &SessionOptions,
emitter: &Emitter,
session_id: &str,
root_session_id: &str,
tool_env_provider: Option<&Arc<dyn ToolEnvProvider>>,
) -> ToolResult {
execute_and_emit_one_tool_with_runtime(
tc,
registry,
env,
tool_hooks,
cancel_token,
config,
emitter,
session_id,
root_session_id,
tool_env_provider,
&AgentToolRuntime::default(),
)
.await
}
#[allow(
clippy::too_many_arguments,
reason = "Single-tool execution needs the tool, runtime handles, and emission context."
)]
async fn execute_and_emit_one_tool_with_runtime(
tc: &ToolCall,
registry: &ToolRegistry,
env: Arc<RunSandbox>,
tool_hooks: Option<&Arc<dyn ToolHookCallback>>,
cancel_token: CancellationToken,
config: &SessionOptions,
emitter: &Emitter,
session_id: &str,
root_session_id: &str,
tool_env_provider: Option<&Arc<dyn ToolEnvProvider>>,
agent_tool_runtime: &AgentToolRuntime,
) -> ToolResult {
let access_denial = config.tool_access_denial_reason(&tc.name);
let registered_tool = if access_denial.is_none() {
registry.get(&tc.name)
} else {
None
};
execute_and_emit_one_tool_with_lookup(
tc,
registered_tool,
access_denial,
env,
tool_hooks,
cancel_token,
config,
emitter,
session_id,
root_session_id,
tool_env_provider,
agent_tool_runtime,
)
.await
}
/// Execute a single tool call with event emission, using a pre-looked-up tool
/// reference.
#[allow(
clippy::too_many_arguments,
reason = "The looked-up execution path still needs the tool, runtime handles, and emission context."
)]
async fn execute_and_emit_one_tool_with_lookup(
tc: &ToolCall,
registered_tool: Option<&RegisteredTool>,
access_denial: Option<String>,
env: Arc<RunSandbox>,
tool_hooks: Option<&Arc<dyn ToolHookCallback>>,
cancel_token: CancellationToken,
config: &SessionOptions,
emitter: &Emitter,
session_id: &str,
root_session_id: &str,
tool_env_provider: Option<&Arc<dyn ToolEnvProvider>>,
agent_tool_runtime: &AgentToolRuntime,
) -> ToolResult {
emit_tool_call_started(emitter, session_id, tc);
if let Some(reason) = access_denial {
return finish_error_result(tc, emitter, session_id, config, &reason);
}
// Pre-tool-use hook
if let Some(hooks) = tool_hooks {
debug!(tool = %tc.name, hook_event = "pre_tool_use", "Calling tool hook");
let start = std::time::Instant::now();
let decision = hooks.pre_tool_use(&tc.name, &tool_call_arguments(tc)).await;
let elapsed = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
debug!(tool = %tc.name, hook_event = "pre_tool_use", ?decision, duration_ms = elapsed, "Tool hook complete");
if let ToolHookDecision::Block { reason } = decision {
return finish_error_result(tc, emitter, session_id, config, &reason);
}
}
let executed = execute_one_tool(
tc,
registered_tool,
env,
cancel_token,
emitter,
session_id,
root_session_id,
tool_env_provider,
agent_tool_runtime,
)
.await;
let retained = retain_tool_result(executed.result, executed.output_stats);
let result = retained.result;
emit_tool_call_result(emitter, session_id, tc, &result, retained.output_stats);
// Post-tool-use hooks
if let Some(hooks) = tool_hooks {
let output = result_output(&result);
let fallback;
let content_str = if let Some(s) = output.as_str() {
s
} else {
fallback = output.to_string();
&fallback
};
if result.is_error {
debug!(tool = %tc.name, hook_event = "post_tool_use_failure", "Calling tool hook");
hooks
.post_tool_use_failure(&tc.name, &tc.id, content_str)
.await;
debug!(tool = %tc.name, hook_event = "post_tool_use_failure", "Tool hook complete");
} else {
debug!(tool = %tc.name, hook_event = "post_tool_use", "Calling tool hook");
hooks.post_tool_use(&tc.name, &tc.id, content_str).await;
debug!(tool = %tc.name, hook_event = "post_tool_use", "Tool hook complete");
}
}
truncate_tool_result(&result, &tc.name, config)
}
struct RetainedToolResult {
result: ToolResult,
output_stats: OutputCaptureStats,
}
/// Bound model-native tool output before it reaches hooks, events, or history.
fn retain_tool_result(
mut result: ToolResult,
previous_stats: Option<OutputCaptureStats>,
) -> RetainedToolResult {
let output_stats = match result.content.as_mut_slice() {
[ContentPart::Text { text: output }] => {
let previously_omitted = previous_stats.map_or(0, |stats| stats.omitted_bytes);
let previewed =
preview_tool_output(output, MAX_RETAINED_TOOL_OUTPUT_BYTES, previously_omitted);
let stats = previewed.stats;
if let Cow::Owned(previewed_output) = previewed.output {
*output = previewed_output;
}
stats
}
_ => OutputCaptureStats::complete(serialized_json_bytes(&result_output(&result))),
};
RetainedToolResult {
result,
output_stats,
}
}
struct ExecutedToolResult {
result: ToolResult,
output_stats: Option<OutputCaptureStats>,
}
/// Execute a single tool call: argument validation and execution.
#[allow(
clippy::too_many_arguments,
reason = "Single-tool execution threads session identity plus runtime handles to populate ToolContext."
)]
async fn execute_one_tool(
tc: &ToolCall,
registered_tool: Option<&RegisteredTool>,
env: Arc<RunSandbox>,
cancel_token: CancellationToken,
emitter: &Emitter,
session_id: &str,
root_session_id: &str,
tool_env_provider: Option<&Arc<dyn ToolEnvProvider>>,
agent_tool_runtime: &AgentToolRuntime,
) -> ExecutedToolResult {
match registered_tool {
Some(tool) => {
let arguments = match &tc.input {
ToolInput::Function(arguments) => match arguments.json() {
Ok(value) => value.clone(),
Err(err) => {
return ExecutedToolResult {
result: error_result(
&tc.id,
format!("Tool arguments are not valid JSON: {err}"),
),
output_stats: None,
};
}
},
_ => tool_call_arguments(tc),
};
if matches!(tc.input, ToolInput::Function(_)) {
if let ToolDefinitionKind::Function { input_schema } = &tool.definition.kind {
if let Err(validation_error) = validate_tool_args(input_schema, &arguments) {
return ExecutedToolResult {
result: error_result(&tc.id, validation_error),
output_stats: None,
};
}
}
}
let session_emitter = Arc::new(SessionBoundEmitter::new(
emitter.clone(),
session_id.to_owned(),
Some(tc.id.clone()),
));
let agent_event_emitter: Option<Arc<dyn AgentEventEmitter>> =
Some(session_emitter.clone());
let ctx = ToolContext {
env,
cancel: cancel_token,
tool_env_provider: tool_env_provider.cloned(),
session_id: Some(session_id.to_owned()),
root_session_id: Some(root_session_id.to_owned()),
tool_call_id: Some(tc.id.clone()),
agent_event_emitter,
};
let execution = (tool.executor)(arguments, ctx);
let result = match question_tools::scope_agent_tool_runtime(
agent_tool_runtime.clone(),
execution,
)
.await
{
Ok(output) => success_result(&tc.id, serde_json::Value::String(output)),
Err(err) => error_result(&tc.id, err),
};
ExecutedToolResult {
result,
output_stats: session_emitter.take_tool_output_stats(),
}
}
None => ExecutedToolResult {
result: error_result(&tc.id, format!("Unknown tool: {}", tc.name)),
output_stats: None,
},
}
}
/// Truncate tool output for history storage while preserving identity fields.
fn truncate_tool_result(
result: &ToolResult,
tool_name: &str,
config: &SessionOptions,
) -> ToolResult {
let content = match result.content.as_slice() {
[ContentPart::Text { text }] => {
vec![ContentPart::Text {
text: truncate_tool_output(text, tool_name, config),
}]
}
other => other.to_vec(),
};
ToolResult {
tool_call_id: result.tool_call_id.clone(),
name: result.name.clone(),
content,
is_error: result.is_error,
}
}
pub fn validate_tool_args(
schema: &serde_json::Value,
args: &serde_json::Value,
) -> Result<(), String> {
// Skip validation for empty/trivial schemas
if schema.is_null() {
return Ok(());
}
if let Some(obj) = schema.as_object() {
if obj.is_empty() {
return Ok(());
}
}
let validator =
jsonschema::validator_for(schema).map_err(|e| format!("Invalid tool schema: {e}"))?;
let errors: Vec<String> = validator.iter_errors(args).map(|e| e.to_string()).collect();
if errors.is_empty() {
Ok(())
} else {
Err(format!(
"Tool argument validation failed: {}",
errors.join("; ")
))
}
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use async_trait::async_trait;
use fabro_sandbox::Termination;
use fabro_sandbox::test_support::exec_result;
use fabro_types::run_event::{AgentToolCompletedProps, MAX_RUN_EVENT_BODY_BYTES};
use fabro_types::{AgentProfileKind, tool_result_to_json};
use lithos_llm::types::{ToolCall, ToolDefinition};
use tokio::sync::broadcast;
use super::*;
use crate::config::{
ToolAccess, ToolAccessPolicy, ToolExposureMode, ToolHookCallback, ToolHookDecision,
};
use crate::event::Emitter;
use crate::local_sandbox;
use crate::question_tools::{
AgentQuestion, AgentQuestionAnswer, AgentQuestionAnswerStatus, AgentQuestionRuntime,
AgentToolRuntime, register_question_tools,
};
use crate::test_support::MockSandbox;
use crate::tool_registry::{RegisteredTool, ToolContext, ToolRegistry, ToolSource};
use crate::tools::make_shell_tool;
use crate::truncation::MAX_SERIALIZED_TOOL_OUTPUT_BYTES;
use crate::types::SessionEvent;
struct NamedPolicy {
decisions: HashMap<String, ToolAccess>,
}
impl NamedPolicy {
fn new(decisions: impl IntoIterator<Item = (&'static str, ToolAccess)>) -> Self {
Self {
decisions: decisions
.into_iter()
.map(|(name, access)| (name.to_string(), access))
.collect(),
}
}
}
impl ToolAccessPolicy for NamedPolicy {
fn access_for_tool(&self, tool_name: &str) -> ToolAccess {
self.decisions
.get(tool_name)
.copied()
.unwrap_or(ToolAccess::Denied)
}
}
fn make_echo_tool() -> RegisteredTool {
RegisteredTool {
definition: ToolDefinition::function(
"echo",
"Echo input",
serde_json::json!({
"type": "object",
"properties": {
"text": {"type": "string"}
},
"required": ["text"]
}),
),
executor: Arc::new(|args: serde_json::Value, _ctx: ToolContext| {
Box::pin(async move {
let text = args["text"].as_str().unwrap_or("").to_string();
Ok(format!("echo: {text}"))
})
}),
source: ToolSource::Native,
}
}
fn make_fail_tool() -> RegisteredTool {
RegisteredTool {
definition: ToolDefinition::function(
"fail_tool",
"Always fails",
serde_json::json!({}),
),
executor: Arc::new(|_args: serde_json::Value, _ctx: ToolContext| {
Box::pin(async move { Err("tool failed".to_string()) })
}),
source: ToolSource::Native,
}
}
fn make_tool_call(name: &str, id: &str, args: serde_json::Value) -> ToolCall {
ToolCall::function(id, name, args)
}
struct StubQuestionRuntime;
#[async_trait]
impl AgentQuestionRuntime for StubQuestionRuntime {
async fn ask_questions(
&self,
_tool_call_id: &str,
questions: Vec<AgentQuestion>,
_cancel_token: CancellationToken,
) -> Result<Vec<AgentQuestionAnswer>, String> {
Ok(questions
.into_iter()
.map(|question| AgentQuestionAnswer {
original_id: question.original_id,
original_question: question.original_question,
answers: vec!["Ship".to_string()],
status: AgentQuestionAnswerStatus::Answered,
})
.collect())
}
}
#[tokio::test]
async fn question_tool_round_rejects_non_question_peers_and_preserves_order() {
let mut registry = ToolRegistry::new();
register_question_tools(AgentProfileKind::OpenAi, &mut registry);
registry.register(make_echo_tool());
let tool_calls = vec![
make_tool_call(
"request_user_input",
"call_question",
serde_json::json!({
"questions": [{
"id": "q1",
"header": "Decision",
"question": "Ship it?",
"options": [{ "label": "Ship" }]
}]
}),
),
make_tool_call("echo", "call_echo", serde_json::json!({"text": "hello"})),
];
let runtime = AgentToolRuntime::with_question_runtime(Arc::new(StubQuestionRuntime));
let results = execute_tool_calls(
&tool_calls,
true,
&registry,
Arc::new(
local_sandbox(std::env::current_dir().unwrap())
.await
.unwrap(),
),
None,
&CancellationToken::new(),
&SessionOptions::default(),
&Emitter::new(),
"root",
"root",
None,
&runtime,
)
.await;
assert_eq!(results.len(), 2);
assert_eq!(results[0].tool_call_id, "call_question");
assert!(!results[0].is_error);
assert_eq!(results[1].tool_call_id, "call_echo");
assert!(results[1].is_error);
assert!(
tool_result_to_json(&results[1])
.as_str()
.unwrap()
.contains("human-question tools must run alone")
);
}
#[tokio::test]
async fn multiple_question_tool_calls_execute_only_first() {
let mut registry = ToolRegistry::new();
register_question_tools(AgentProfileKind::OpenAi, &mut registry);
let question_args = serde_json::json!({
"questions": [{
"id": "q1",
"header": "Decision",
"question": "Ship it?",
"options": [{ "label": "Ship" }]
}]
});
let tool_calls = vec![
make_tool_call("request_user_input", "call_first", question_args.clone()),
make_tool_call("request_user_input", "call_second", question_args),
];
let runtime = AgentToolRuntime::with_question_runtime(Arc::new(StubQuestionRuntime));
let results = execute_tool_calls(
&tool_calls,
true,
&registry,
Arc::new(
local_sandbox(std::env::current_dir().unwrap())
.await
.unwrap(),
),
None,
&CancellationToken::new(),
&SessionOptions::default(),
&Emitter::new(),
"root",
"root",
None,
&runtime,
)
.await;
assert!(!results[0].is_error);
assert!(results[1].is_error);
assert!(
tool_result_to_json(&results[1])
.as_str()
.unwrap()
.contains("Combine all questions into a single questions[] batch")
);
}
struct MockHookCallback {
pre_decision: ToolHookDecision,
post_calls: Arc<Mutex<Vec<(String, String, String)>>>,
post_failure_calls: Arc<Mutex<Vec<(String, String, String)>>>,
}
impl MockHookCallback {
fn new(decision: ToolHookDecision) -> Self {
Self {
pre_decision: decision,
post_calls: Arc::new(Mutex::new(Vec::new())),
post_failure_calls: Arc::new(Mutex::new(Vec::new())),
}
}
}
#[async_trait::async_trait]
impl ToolHookCallback for MockHookCallback {
async fn pre_tool_use(
&self,
_tool_name: &str,
_tool_input: &serde_json::Value,
) -> ToolHookDecision {
self.pre_decision.clone()
}
async fn post_tool_use(&self, tool_name: &str, tool_call_id: &str, tool_output: &str) {
self.post_calls.lock().unwrap().push((
tool_name.to_string(),
tool_call_id.to_string(),
tool_output.to_string(),
));
}
async fn post_tool_use_failure(&self, tool_name: &str, tool_call_id: &str, error: &str) {
self.post_failure_calls.lock().unwrap().push((
tool_name.to_string(),
tool_call_id.to_string(),
error.to_string(),
));
}
}
async fn make_sandbox() -> Arc<RunSandbox> {
Arc::new(
local_sandbox(std::env::current_dir().unwrap())
.await
.unwrap(),
)
}
#[tokio::test]
async fn pre_tool_use_hook_blocks_execution() {
let mut registry = ToolRegistry::new();
registry.register(make_echo_tool());
let hooks: Arc<dyn ToolHookCallback> =
Arc::new(MockHookCallback::new(ToolHookDecision::Block {
reason: "blocked by hook".to_string(),
}));
let tc = make_tool_call("echo", "call_1", serde_json::json!({"text": "hello"}));
let emitter = Emitter::new();
let config = SessionOptions::default();
let result = execute_and_emit_one_tool(
&tc,
&registry,
make_sandbox().await,
Some(&hooks),
CancellationToken::new(),
&config,
&emitter,
"test-session",
"test-session",
None,
)
.await;
assert!(result.is_error);
let content = tool_result_to_json(&result);
assert!(content.as_str().unwrap().contains("blocked by hook"));
}
#[tokio::test]
async fn pre_tool_use_hook_proceeds() {
let mut registry = ToolRegistry::new();
registry.register(make_echo_tool());
let hooks: Arc<dyn ToolHookCallback> =
Arc::new(MockHookCallback::new(ToolHookDecision::Proceed));
let tc = make_tool_call("echo", "call_1", serde_json::json!({"text": "hello"}));
let emitter = Emitter::new();
let config = SessionOptions::default();
let result = execute_and_emit_one_tool(
&tc,
&registry,
make_sandbox().await,
Some(&hooks),
CancellationToken::new(),
&config,
&emitter,
"test-session",
"test-session",
None,
)
.await;
assert!(!result.is_error);
let content = tool_result_to_json(&result).to_string();
assert!(content.contains("echo: hello"));
}
#[tokio::test]
async fn tool_output_is_bounded_before_events_and_history() {
let mut registry = ToolRegistry::new();
registry.register(make_echo_tool());
let text = "x".repeat(MAX_RETAINED_TOOL_OUTPUT_BYTES + 100);
let tc = make_tool_call("echo", "call_large", serde_json::json!({"text": text}));
let emitter = Emitter::new();
let mut receiver = emitter.subscribe();
let result = execute_and_emit_one_tool(
&tc,
&registry,
make_sandbox().await,
None,
CancellationToken::new(),
&SessionOptions::default(),
&emitter,
"test-session",
"test-session",
None,
)
.await;
let result_output = tool_result_to_json(&result);
let result_output = result_output.as_str().expect("string tool output");
assert!(result_output.len() <= MAX_RETAINED_TOOL_OUTPUT_BYTES);
assert!(result_output.starts_with("Warning: truncated output"));
assert!(result_output.contains("bytes omitted"));
assert!(result_output.contains("tokens truncated"));
assert!(!result_output.contains("re-run"));
let completed = loop {
let event = receiver.try_recv().expect("tool completion event");
if let AgentEvent::ToolCallCompleted {
output,
is_error,
output_bytes_observed,
output_bytes_retained,
output_bytes_omitted,
..
} = event.event
{
break (
output,
is_error,
output_bytes_observed,
output_bytes_retained,
output_bytes_omitted,
);
}
};
let event_output = completed.0.as_str().expect("string event output");
assert_eq!(event_output, result_output);
assert!(!completed.1, "truncation must not make the tool an error");
assert_eq!(
completed.2,
MAX_RETAINED_TOOL_OUTPUT_BYTES + 100 + "echo: ".len()
);
assert!(completed.3 < MAX_RETAINED_TOOL_OUTPUT_BYTES);
assert_eq!(completed.4, completed.2 - completed.3);
assert!(event_output.contains(&format!("... {} bytes omitted ...", completed.4)));
}
#[tokio::test]
async fn serialized_tool_output_and_full_event_stay_within_reserved_budgets() {
let mut registry = ToolRegistry::new();
registry.register(make_echo_tool());
let text = format!(
"HEAD{}TAIL",
"\0".repeat(MAX_RETAINED_TOOL_OUTPUT_BYTES - "echo: HEADTAIL".len())
);
let tc = make_tool_call("echo", "call_escaped", serde_json::json!({"text": text}));
let emitter = Emitter::new();
let mut receiver = emitter.subscribe();
let result = execute_and_emit_one_tool(
&tc,
&registry,
make_sandbox().await,
None,
CancellationToken::new(),
&SessionOptions::default(),
&emitter,
"test-session",
"test-session",
None,
)
.await;
assert!(!result.is_error);
let completed = loop {
let event = receiver.try_recv().expect("tool completion event");
if let AgentEvent::ToolCallCompleted {
tool_name,
tool_call_id,
output,
is_error,
output_bytes_observed,
output_bytes_retained,
output_bytes_omitted,
} = event.event
{
break (
tool_name,
tool_call_id,
output,
is_error,
output_bytes_observed,
output_bytes_retained,
output_bytes_omitted,
);
}
};
let serialized_output_bytes = serde_json::to_vec(&completed.2)
.expect("tool output serializes")
.len();
assert!(serialized_output_bytes <= MAX_SERIALIZED_TOOL_OUTPUT_BYTES);
let run_id = fabro_types::RunId::new();
let run_event = fabro_types::RunEvent {
id: "evt-escaped-output".to_string(),
ts: chrono::Utc::now(),
run_id,
node_id: None,
node_label: None,
stage_id: None,
parallel_group_id: None,
parallel_branch_id: None,
session_id: Some("test-session".to_string()),
parent_session_id: None,
tool_call_id: Some(completed.1.clone()),
actor: None,
body: fabro_types::EventBody::AgentToolCompleted(AgentToolCompletedProps {
tool_name: completed.0,
tool_call_id: completed.1,
output: completed.2,
is_error: completed.3,
visit: 1,
output_bytes_observed: Some(completed.4 as u64),
output_bytes_retained: Some(completed.5 as u64),
output_bytes_omitted: Some(completed.6 as u64),
tool_result: None,
turn_id: None,
}),
};
let serialized_event_bytes = serde_json::to_vec(&run_event)
.expect("run event serializes")
.len();
// Leave at least 1 MiB of envelope headroom under the server's
// run-event body limit.
let event_body_budget = MAX_RUN_EVENT_BODY_BYTES - 1024 * 1024;
assert!(
serialized_event_bytes < event_body_budget,
"serialized event was {serialized_event_bytes} bytes"
);
}
#[tokio::test]
async fn post_tool_use_hook_fires_on_success() {
let mut registry = ToolRegistry::new();
registry.register(make_echo_tool());
let mock = Arc::new(MockHookCallback::new(ToolHookDecision::Proceed));
let hooks: Arc<dyn ToolHookCallback> = mock.clone();
let tc = make_tool_call("echo", "call_1", serde_json::json!({"text": "hello"}));
let emitter = Emitter::new();
let config = SessionOptions::default();
execute_and_emit_one_tool(
&tc,
&registry,
make_sandbox().await,
Some(&hooks),
CancellationToken::new(),
&config,
&emitter,
"test-session",
"test-session",
None,
)
.await;
let calls = mock.post_calls.lock().unwrap();
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].0, "echo");
assert_eq!(calls[0].1, "call_1");
assert!(calls[0].2.contains("echo: hello"));
let failure_calls = mock.post_failure_calls.lock().unwrap();
assert!(failure_calls.is_empty());
}
#[tokio::test]
async fn post_tool_use_failure_hook_fires_on_error() {
let mut registry = ToolRegistry::new();
registry.register(make_fail_tool());
let mock = Arc::new(MockHookCallback::new(ToolHookDecision::Proceed));
let hooks: Arc<dyn ToolHookCallback> = mock.clone();
let tc = make_tool_call("fail_tool", "call_1", serde_json::json!({}));
let emitter = Emitter::new();
let config = SessionOptions::default();
execute_and_emit_one_tool(
&tc,
&registry,
make_sandbox().await,
Some(&hooks),
CancellationToken::new(),
&config,
&emitter,
"test-session",
"test-session",
None,
)
.await;
let failure_calls = mock.post_failure_calls.lock().unwrap();
assert_eq!(failure_calls.len(), 1);
assert_eq!(failure_calls[0].0, "fail_tool");
assert_eq!(failure_calls[0].1, "call_1");
assert!(failure_calls[0].2.contains("tool failed"));
let calls = mock.post_calls.lock().unwrap();
assert!(calls.is_empty());
}
#[tokio::test]
async fn no_hooks_skips_all_callbacks() {
let mut registry = ToolRegistry::new();
registry.register(make_echo_tool());
let tc = make_tool_call("echo", "call_1", serde_json::json!({"text": "hello"}));
let emitter = Emitter::new();
let config = SessionOptions::default();
let result = execute_and_emit_one_tool(
&tc,
&registry,
make_sandbox().await,
None,
CancellationToken::new(),
&config,
&emitter,
"test-session",
"test-session",
None,
)
.await;
assert!(!result.is_error);
let content = tool_result_to_json(&result).to_string();
assert!(content.contains("echo: hello"));
}
#[tokio::test]
async fn denied_policy_tool_is_blocked_before_executor_lookup() {
let executions = Arc::new(Mutex::new(0usize));
let mut registry = ToolRegistry::new();
let executions_for_tool = Arc::clone(&executions);
registry.register(RegisteredTool {
definition: ToolDefinition::function(
"write_file",
"Writes a file",
serde_json::json!({"type": "object"}),
),
executor: Arc::new(move |_args: serde_json::Value, _ctx: ToolContext| {
let executions = Arc::clone(&executions_for_tool);
Box::pin(async move {
*executions.lock().unwrap() += 1;
Ok("wrote".to_string())
})
}),
source: ToolSource::Native,
});
let config = SessionOptions {
tool_access_policy: Some(Arc::new(NamedPolicy::new([(
"write_file",
ToolAccess::Denied,
)]))),
tool_exposure_mode: ToolExposureMode::IncludeRequiresApproval,
..SessionOptions::default()
};
let tc = make_tool_call("write_file", "call_1", serde_json::json!({}));
let result = execute_and_emit_one_tool(
&tc,
&registry,
make_sandbox().await,
None,
CancellationToken::new(),
&config,
&Emitter::new(),
"test-session",
"test-session",
None,
)
.await;
assert!(result.is_error);
assert!(
tool_result_to_json(&result)
.as_str()
.unwrap_or_default()
.contains("denied by tool access policy")
);
assert_eq!(*executions.lock().unwrap(), 0);
}
#[tokio::test]
async fn approval_required_tool_hidden_by_exposure_mode_is_blocked() {
let executions = Arc::new(Mutex::new(0usize));
let mut registry = ToolRegistry::new();
let executions_for_tool = Arc::clone(&executions);
registry.register(RegisteredTool {
definition: ToolDefinition::function(
"shell",
"Runs a command",
serde_json::json!({"type": "object"}),
),
executor: Arc::new(move |_args: serde_json::Value, _ctx: ToolContext| {
let executions = Arc::clone(&executions_for_tool);
Box::pin(async move {
*executions.lock().unwrap() += 1;
Ok("ran".to_string())
})
}),
source: ToolSource::Native,
});
let config = SessionOptions {
tool_access_policy: Some(Arc::new(NamedPolicy::new([(
"shell",
ToolAccess::RequiresApproval,
)]))),
tool_exposure_mode: ToolExposureMode::AutoApprovedOnly,
..SessionOptions::default()
};
let tc = make_tool_call("shell", "call_1", serde_json::json!({}));
let result = execute_and_emit_one_tool(
&tc,
&registry,
make_sandbox().await,
None,
CancellationToken::new(),
&config,
&Emitter::new(),
"test-session",
"test-session",
None,
)
.await;
assert!(result.is_error);
assert!(
tool_result_to_json(&result)
.as_str()
.unwrap_or_default()
.contains("requires approval")
);
assert_eq!(*executions.lock().unwrap(), 0);
}
fn shell_sandbox(result: fabro_sandbox::ExecResult) -> Arc<RunSandbox> {
MockSandbox {
exec_result: result,
..Default::default()
}
.sandbox()
}
fn exited(exit_code: i32) -> fabro_sandbox::ExecResult {
exec_result("out", "err", Some(exit_code), Termination::Exited, 12)
}
fn cancelled() -> fabro_sandbox::ExecResult {
exec_result("out", "", None, Termination::Cancelled, 12)
}
async fn run_shell_tool(
exec_result: fabro_sandbox::ExecResult,
hooks: Option<&Arc<dyn ToolHookCallback>>,
emitter: &Emitter,
) -> ToolResult {
let mut registry = ToolRegistry::new();
registry.register(make_shell_tool());
let tc = make_tool_call(
"shell",
"call_1",
serde_json::json!({"command": "make test"}),
);
execute_and_emit_one_tool(
&tc,
&registry,
shell_sandbox(exec_result),
hooks,
CancellationToken::new(),
&SessionOptions::default(),
emitter,
"test-session",
"test-session",
None,
)
.await
}
fn drain(receiver: &mut broadcast::Receiver<SessionEvent>) -> Vec<SessionEvent> {
let mut events = Vec::new();
while let Ok(event) = receiver.try_recv() {
events.push(event);
}
events
}
#[tokio::test]
async fn shell_nonzero_exit_becomes_an_error_tool_result() {
let emitter = Emitter::new();
let result = run_shell_tool(exited(7), None, &emitter).await;
assert!(result.is_error);
assert!(
tool_result_to_json(&result)
.as_str()
.unwrap()
.contains("Exit code: 7"),
"got: {}",
tool_result_to_json(&result)
);
}
#[tokio::test]
async fn shell_exit_zero_remains_a_successful_tool_result() {
let emitter = Emitter::new();
let result = run_shell_tool(exited(0), None, &emitter).await;
assert!(!result.is_error);
}
#[tokio::test]
async fn shell_events_record_process_and_rendered_output_byte_counts() {
let output_len = MAX_RETAINED_TOOL_OUTPUT_BYTES + 1_000;
let emitter = Emitter::new();
let mut receiver = emitter.subscribe();
let result = run_shell_tool(
exec_result(
&("x".repeat(output_len)),
"",
Some(0),
Termination::Exited,
12,
),
None,
&emitter,
)
.await;
assert!(!result.is_error);
let events = drain(&mut receiver);
let process = events
.iter()
.find_map(|event| match &event.event {
AgentEvent::ToolProcessCompleted {
output_bytes_observed,
output_bytes_retained,
output_bytes_omitted,
..
} => Some((
*output_bytes_observed,
*output_bytes_retained,
*output_bytes_omitted,
)),
_ => None,
})
.expect("process event");
assert_eq!(process, (output_len, MAX_RETAINED_TOOL_OUTPUT_BYTES, 1_000));
let completed = events
.iter()
.find_map(|event| match &event.event {
AgentEvent::ToolCallCompleted {
output,
is_error,
output_bytes_observed,
output_bytes_retained,
output_bytes_omitted,
..
} => Some((
output.as_str().expect("string event output"),
*is_error,
*output_bytes_observed,
*output_bytes_retained,
*output_bytes_omitted,
)),
_ => None,
})
.expect("tool completion event");
assert!(completed.0.starts_with("Warning: truncated output"));
assert!(!completed.1, "truncation must not make the tool an error");
assert!(completed.2 > output_len);
assert!(completed.3 < MAX_RETAINED_TOOL_OUTPUT_BYTES);
assert_eq!(completed.4, completed.2 - completed.3);
assert!(
completed
.0
.contains(&format!("... {} bytes omitted ...", completed.4))
);
}
#[tokio::test]
async fn shell_failure_emits_started_then_process_then_completed() {
let emitter = Emitter::new();
let mut receiver = emitter.subscribe();
run_shell_tool(exited(7), None, &emitter).await;
let events = drain(&mut receiver);
let names: Vec<&str> = events
.iter()
.filter_map(|event| match &event.event {
AgentEvent::ToolCallStarted { .. } => Some("started"),
AgentEvent::ToolProcessCompleted { .. } => Some("process"),
AgentEvent::ToolCallCompleted { .. } => Some("completed"),
_ => None,
})
.collect();
assert_eq!(names, vec!["started", "process", "completed"]);
for event in &events {
assert_eq!(event.session_id, "test-session");
}
let process = events
.iter()
.find(|event| matches!(event.event, AgentEvent::ToolProcessCompleted { .. }))
.expect("process event");
assert_eq!(process.tool_call_id.as_deref(), Some("call_1"));
match &process.event {
AgentEvent::ToolProcessCompleted {
exit_code,
termination,
..
} => {
assert_eq!(*exit_code, Some(7));
assert_eq!(*termination, fabro_types::CommandTermination::Exited);
}
other => panic!("expected a process event, got {other:?}"),
}
let completed = events
.iter()
.find_map(|event| match &event.event {
AgentEvent::ToolCallCompleted {
tool_call_id,
is_error,
..
} => Some((tool_call_id.clone(), *is_error)),
_ => None,
})
.expect("tool completed event");
assert_eq!(completed, ("call_1".to_string(), true));
}
#[tokio::test]
async fn shell_failure_runs_only_the_failure_hook() {
for exec_result in [exited(7), cancelled()] {
let mock = Arc::new(MockHookCallback::new(ToolHookDecision::Proceed));
let hooks: Arc<dyn ToolHookCallback> = mock.clone();
run_shell_tool(exec_result, Some(&hooks), &Emitter::new()).await;
assert_eq!(mock.post_failure_calls.lock().unwrap().len(), 1);
assert!(mock.post_calls.lock().unwrap().is_empty());
}
}
#[tokio::test]
async fn shell_success_runs_only_the_success_hook() {
let mock = Arc::new(MockHookCallback::new(ToolHookDecision::Proceed));
let hooks: Arc<dyn ToolHookCallback> = mock.clone();
run_shell_tool(exited(0), Some(&hooks), &Emitter::new()).await;
assert_eq!(mock.post_calls.lock().unwrap().len(), 1);
assert!(mock.post_failure_calls.lock().unwrap().is_empty());
}
#[test]
fn truncation_preserves_tool_call_id_and_error_state() {
let result = tool_result_from_json(
"call_1",
serde_json::Value::String("x".repeat(60_000)),
true,
);
let truncated = truncate_tool_result(&result, "shell", &SessionOptions::default());
assert_eq!(truncated.tool_call_id, "call_1");
assert!(truncated.is_error);
assert!(tool_result_to_json(&truncated).as_str().unwrap().len() < 60_000);
}
}
Reference in a new issue View git blame Copy permalink