use std::collections::VecDeque;
use std::sync::Arc;
use tokio::sync::mpsc;
use crate::composition::CompositionExecutor;
use crate::tools::executor::execute_tool_call_with_context;
use crate::tools::{
convert_from_standard_result, plan_mode_allows_tool, AgenticToolResult,
ExecutingSupervisorObservation, ToolCall, ToolError, ToolExecutionContext,
ToolExecutionSessionFlags, ToolExecutor, ToolResult,
};
use crate::{AgentEvent, Message, PendingQuestionSource, Session};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ToolHandlingOutcome {
Continue,
AwaitingClarification,
WaitingForChildren,
}
pub fn is_waiting_for_children_control(result: &ToolResult) -> bool {
if result.display_preference.as_deref() == Some("runtime_control:waiting_for_children") {
return true;
}
result.result.trim_start().starts_with('{')
&& serde_json::from_str::<serde_json::Value>(&result.result)
.ok()
.and_then(|value| value.get("runtime_control").cloned())
.and_then(|control| control.as_str().map(str::to_string))
.is_some_and(|control| control == "waiting_for_children")
}
pub const MAX_SUB_ACTIONS: usize = 64;
pub fn parse_tool_args(arguments: &str) -> std::result::Result<serde_json::Value, ToolError> {
let args_raw = arguments.trim();
if args_raw.is_empty() {
return Ok(serde_json::json!({}));
}
serde_json::from_str(args_raw)
.map_err(|error| ToolError::InvalidArguments(format!("Invalid JSON arguments: {error}")))
}
pub fn tool_start_arguments_for_display(
tool_name: &str,
args: &serde_json::Value,
) -> serde_json::Value {
let browser = tool_name
.trim()
.rsplit("::")
.next()
.is_some_and(|name| name.eq_ignore_ascii_case("browser"));
if browser
&& args
.get("action")
.and_then(serde_json::Value::as_str)
.is_some_and(|action| action.eq_ignore_ascii_case("download"))
{
let mut display = serde_json::json!({"action":"download"});
if let Some(epoch) = args
.get("expected_epoch")
.and_then(serde_json::Value::as_u64)
{
display["expected_epoch"] = serde_json::json!(epoch);
}
display
} else if browser
&& (args.get("action").and_then(serde_json::Value::as_str) == Some("set_file_input")
|| args.get("data_base64").is_some())
{
serde_json::json!({"action":"set_file_input","file":"[redacted]"})
} else {
args.clone()
}
}
fn tool_start_name_for_display(tool_name: &str) -> String {
if tool_name
.trim()
.rsplit("::")
.next()
.is_some_and(|name| name.eq_ignore_ascii_case("browser"))
{
"browser".to_string()
} else {
tool_name.to_string()
}
}
fn trim_end_whitespace_in_place(value: &mut String) {
let trimmed_len = value.trim_end_matches(char::is_whitespace).len();
value.truncate(trimmed_len);
}
fn strip_trailing_commas_in_place(value: &mut String) {
loop {
trim_end_whitespace_in_place(value);
if value.ends_with(',') {
value.pop();
continue;
}
break;
}
}
fn preview_for_log(value: &str, max_chars: usize) -> String {
let mut iter = value.chars();
let mut preview = String::new();
for _ in 0..max_chars {
match iter.next() {
Some(ch) => preview.push(ch),
None => break,
}
}
if iter.next().is_some() {
preview.push_str("...");
}
preview.replace('\n', "\\n").replace('\r', "\\r")
}
fn attempt_repair_truncated_json(arguments: &str) -> Option<String> {
let args_raw = arguments.trim();
if args_raw.is_empty() {
return None;
}
if !args_raw.starts_with('{') && !args_raw.starts_with('[') {
return None;
}
let mut stack: Vec<char> = Vec::new();
let mut in_string = false;
let mut escaped = false;
for ch in args_raw.chars() {
if in_string {
if escaped {
escaped = false;
continue;
}
match ch {
'\\' => escaped = true,
'"' => in_string = false,
_ => {}
}
continue;
}
match ch {
'"' => in_string = true,
'{' => stack.push('}'),
'[' => stack.push(']'),
'}' | ']' => {
if stack.last().copied() == Some(ch) {
stack.pop();
} else {
return None;
}
}
_ => {}
}
}
if !in_string && stack.is_empty() {
return None;
}
let mut repaired = args_raw.to_string();
if in_string {
repaired.push('"');
}
while let Some(closing) = stack.pop() {
strip_trailing_commas_in_place(&mut repaired);
repaired.push(closing);
}
strip_trailing_commas_in_place(&mut repaired);
Some(repaired)
}
pub fn parse_tool_args_best_effort(arguments: &str) -> (serde_json::Value, Option<String>) {
let args_raw = arguments.trim();
if args_raw.is_empty() {
return (serde_json::json!({}), None);
}
match serde_json::from_str::<serde_json::Value>(args_raw) {
Ok(parsed) => (parsed, None),
Err(primary_error) => {
if let Some(repaired_json) = attempt_repair_truncated_json(args_raw) {
match serde_json::from_str::<serde_json::Value>(&repaired_json) {
Ok(parsed) => {
let warning = format!(
"Invalid JSON arguments recovered via auto-repair: original_error={}, repaired_preview=\"{}\"",
primary_error,
preview_for_log(&repaired_json, 180)
);
return (parsed, Some(warning));
}
Err(repair_error) => {
let warning = format!(
"Invalid JSON arguments: {} (auto-repair failed: {}); falling back to empty object",
primary_error, repair_error
);
return (serde_json::json!({}), Some(warning));
}
}
}
let warning = format!(
"Invalid JSON arguments: {}; falling back to empty object",
primary_error
);
(serde_json::json!({}), Some(warning))
}
}
}
pub fn try_parse_agentic_result(result: &ToolResult) -> Option<AgenticToolResult> {
if result.result.trim_start().starts_with('{') {
if let Ok(parsed) = serde_json::from_str::<AgenticToolResult>(&result.result) {
return Some(parsed);
}
}
match result.display_preference.as_deref() {
Some("clarification") | Some("actions_needed") => {
Some(convert_from_standard_result(result.clone()))
}
_ => None,
}
}
pub fn tool_result_requires_suspension(result: &ToolResult) -> bool {
result.success
&& (is_waiting_for_children_control(result)
|| matches!(
try_parse_agentic_result(result),
Some(AgenticToolResult::NeedClarification { .. })
))
}
pub async fn handle_tool_result_with_agentic_support(
result: &ToolResult,
tool_call: &ToolCall,
event_tx: &mpsc::Sender<AgentEvent>,
session: &mut Session,
tools: &dyn ToolExecutor,
session_flags: ToolExecutionSessionFlags,
composition_executor: Option<Arc<CompositionExecutor>>,
) -> ToolHandlingOutcome {
handle_tool_result_with_agentic_support_and_persistence(
result,
tool_call,
event_tx,
session,
tools,
session_flags,
composition_executor,
None,
)
.await
}
async fn refuse_suspension_result(
session: &mut Session,
tool_call: &ToolCall,
event_tx: &mpsc::Sender<AgentEvent>,
) {
let error = "suspension_disabled: this execution cannot wait for a user or child".to_string();
session.add_message(Message::tool_result_with_status(
tool_call.id.clone(),
error.clone(),
false,
));
let _ = event_tx
.send(AgentEvent::ToolError {
tool_call_id: tool_call.id.clone(),
error,
})
.await;
}
#[allow(clippy::too_many_arguments)]
pub async fn handle_tool_result_with_agentic_support_and_persistence(
result: &ToolResult,
tool_call: &ToolCall,
event_tx: &mpsc::Sender<AgentEvent>,
session: &mut Session,
tools: &dyn ToolExecutor,
session_flags: ToolExecutionSessionFlags,
composition_executor: Option<Arc<CompositionExecutor>>,
clarification_persistence: Option<&Arc<dyn bamboo_domain::RuntimeSessionPersistence>>,
) -> ToolHandlingOutcome {
if !crate::tools::suspension_allowed() && tool_result_requires_suspension(result) {
refuse_suspension_result(session, tool_call, event_tx).await;
return ToolHandlingOutcome::Continue;
}
let should_wait_for_children = is_waiting_for_children_control(result)
&& (result.success || crate::tools::suspension_allowed());
if should_wait_for_children {
session.metadata.insert(
"runtime.suspend_reason".to_string(),
"waiting_for_children".to_string(),
);
}
let parsed = if !crate::tools::suspension_allowed() && !result.success {
None
} else {
try_parse_agentic_result(result)
};
let Some(agentic_result) = parsed else {
let message = if result.images.is_empty() {
Message::tool_result_with_status(
tool_call.id.clone(),
result.result.clone(),
result.success,
)
} else {
Message::tool_result_with_images(
tool_call.id.clone(),
result.result.clone(),
result.success,
result.images.clone(),
)
};
session.add_message(message);
return if should_wait_for_children {
ToolHandlingOutcome::WaitingForChildren
} else {
ToolHandlingOutcome::Continue
};
};
match agentic_result {
AgenticToolResult::Success { result } => {
session.add_message(Message::tool_result(tool_call.id.clone(), result));
if should_wait_for_children {
ToolHandlingOutcome::WaitingForChildren
} else {
ToolHandlingOutcome::Continue
}
}
AgenticToolResult::Error { error } => {
let _ = event_tx
.send(AgentEvent::ToolError {
tool_call_id: tool_call.id.clone(),
error: error.clone(),
})
.await;
session.add_message(Message::tool_result_with_status(
tool_call.id.clone(),
format!("Error: {error}"),
false,
));
ToolHandlingOutcome::Continue
}
AgenticToolResult::NeedClarification { question, options } => {
persist_and_publish_agentic_clarification(
event_tx,
session,
tool_call,
question,
options,
clarification_persistence,
)
.await;
ToolHandlingOutcome::AwaitingClarification
}
AgenticToolResult::NeedMoreActions { actions, reason } => {
session.add_message(Message::tool_result(
tool_call.id.clone(),
format!(
"Need more actions: {reason} ({} actions pending)",
actions.len()
),
));
execute_sub_actions_with_persistence(
&actions,
event_tx,
session,
tools,
session_flags,
composition_executor,
clarification_persistence,
)
.await
}
}
}
async fn persist_and_publish_agentic_clarification(
event_tx: &mpsc::Sender<AgentEvent>,
session: &mut Session,
tool_call: &ToolCall,
question: String,
options: Option<Vec<String>>,
clarification_persistence: Option<&Arc<dyn bamboo_domain::RuntimeSessionPersistence>>,
) -> bool {
persist_agentic_clarification(session, tool_call, question.clone(), options.clone());
if let Some(persistence) = clarification_persistence {
if let Err(error) = persistence.save_runtime_session(session).await {
tracing::error!(
session_id = %session.id,
%error,
"Failed to persist agentic clarification; suppressing NeedClarification"
);
let _ = event_tx
.send(AgentEvent::Error {
message: format!(
"Clarification could not be saved; retry after session storage recovers: {error}"
),
})
.await;
return false;
}
}
send_clarification_request(
event_tx,
question,
options,
Some(tool_call.id.clone()),
Some(tool_call.function.name.clone()),
)
.await;
true
}
fn persist_agentic_clarification(
session: &mut Session,
tool_call: &ToolCall,
question: String,
options: Option<Vec<String>>,
) {
let normalized_options = options.unwrap_or_default();
session.set_pending_question_with_source(
tool_call.id.clone(),
tool_call.function.name.clone(),
question.clone(),
normalized_options,
true,
PendingQuestionSource::AgenticClarification,
);
session.metadata.insert(
"runtime.suspend_reason".to_string(),
"awaiting_clarification".to_string(),
);
session.add_message(Message::tool_result(
tool_call.id.clone(),
format!("Clarification needed: {question}"),
));
}
pub async fn send_clarification_request(
event_tx: &mpsc::Sender<AgentEvent>,
question: String,
options: Option<Vec<String>>,
tool_call_id: Option<String>,
tool_name: Option<String>,
) {
let _ = event_tx
.send(AgentEvent::NeedClarification {
question,
options,
tool_call_id,
tool_name,
allow_custom: true,
source: Some(PendingQuestionSource::AgenticClarification),
})
.await;
}
pub async fn execute_sub_actions(
actions: &[ToolCall],
event_tx: &mpsc::Sender<AgentEvent>,
session: &mut Session,
tools: &dyn ToolExecutor,
session_flags: ToolExecutionSessionFlags,
composition_executor: Option<Arc<CompositionExecutor>>,
) -> ToolHandlingOutcome {
execute_sub_actions_with_persistence(
actions,
event_tx,
session,
tools,
session_flags,
composition_executor,
None,
)
.await
}
async fn execute_sub_actions_with_persistence(
actions: &[ToolCall],
event_tx: &mpsc::Sender<AgentEvent>,
session: &mut Session,
tools: &dyn ToolExecutor,
session_flags: ToolExecutionSessionFlags,
composition_executor: Option<Arc<CompositionExecutor>>,
clarification_persistence: Option<&Arc<dyn bamboo_domain::RuntimeSessionPersistence>>,
) -> ToolHandlingOutcome {
let executing_supervisor =
ExecutingSupervisorObservation::capture_from_executing_session(session);
let mut pending: VecDeque<ToolCall> = actions.iter().cloned().collect();
let mut processed = 0usize;
let available_tools = tools.list_tools();
while let Some(action) = pending.pop_front() {
if crate::tools::suspension_running_violated() {
break;
}
if processed >= MAX_SUB_ACTIONS {
let error = format!("Reached max sub-action limit ({MAX_SUB_ACTIONS})");
let _ = event_tx
.send(AgentEvent::ToolError {
tool_call_id: action.id.clone(),
error: error.clone(),
})
.await;
session.add_message(Message::tool_result_with_status(
action.id.clone(),
error,
false,
));
return ToolHandlingOutcome::Continue;
}
processed += 1;
let execution_name = if session.model_orchestration_only_enabled() {
let selected =
bamboo_domain::resolve_tool_reference_name(&action.function.name, |name| {
available_tools
.iter()
.any(|schema| schema.function.name == name)
});
if selected
.as_deref()
.is_none_or(|name| !session.allows_model_tool_execution(name))
{
let error = if executing_supervisor.is_some() {
format!("supervisor_delegation_required: {} is outside Supervisor orchestration authority; delegate to an authorized worker or report the recoverable blocker", action.function.name)
} else {
format!(
"Tool '{}' is outside orchestration-only Root authority",
action.function.name
)
};
let _ = event_tx
.send(AgentEvent::ToolError {
tool_call_id: action.id.clone(),
error: error.clone(),
})
.await;
session.add_message(Message::tool_result_with_status(
action.id.clone(),
error,
false,
));
continue;
}
selected
} else {
None
};
if session_flags.plan_read_only && !plan_mode_allows_tool(&action.function.name) {
let error = format!(
"Plan mode: {} operation blocked",
action.function.name.trim()
);
let display_error = format!(
"Plan mode: {} operation blocked",
tool_start_name_for_display(&action.function.name)
);
let _ = event_tx
.send(AgentEvent::ToolError {
tool_call_id: action.id.clone(),
error: display_error,
})
.await;
session.add_message(Message::tool_result_with_status(
action.id.clone(),
error,
false,
));
continue;
}
let args =
parse_tool_args(&action.function.arguments).unwrap_or_else(|_| serde_json::json!({}));
let _ = event_tx
.send(AgentEvent::ToolStart {
tool_call_id: action.id.clone(),
tool_name: tool_start_name_for_display(&action.function.name),
arguments: tool_start_arguments_for_display(&action.function.name, &args),
})
.await;
let tool_ctx = ToolExecutionContext::for_dispatch(
&session.id,
&session.root_session_id,
&action.id,
event_tx,
available_tools.as_slice(),
session_flags,
false,
None,
None,
)
.with_executing_supervisor(executing_supervisor);
match super::context::with_root_actor_tool_writer(None, async {
if let Some(execution_name) = execution_name.as_deref() {
tools
.execute_exact_with_context_outcome(&action, execution_name, tool_ctx)
.await
.map(crate::tools::ToolOutcome::into_tool_result)
} else if crate::tools::capture_suspension_restriction().is_some() {
crate::tools::executor::execute_tool_call_with_context_outcome(
&action,
tools,
composition_executor.clone(),
tool_ctx,
)
.await
.map(crate::tools::ToolOutcome::into_tool_result)
} else {
execute_tool_call_with_context(
&action,
tools,
composition_executor.clone(),
tool_ctx,
)
.await
}
})
.await
{
Ok(result) => {
if crate::tools::suspension_running_violated() {
let error =
"suspension_disabled: detached Running outcome refused; stop unconfirmed"
.to_string();
session.add_message(Message::tool_result_with_status(
action.id.clone(),
error.clone(),
false,
));
let _ = event_tx
.send(AgentEvent::ToolError {
tool_call_id: action.id.clone(),
error,
})
.await;
break;
}
if !crate::tools::suspension_allowed() && tool_result_requires_suspension(&result) {
refuse_suspension_result(session, &action, event_tx).await;
continue;
}
let _ = event_tx
.send(AgentEvent::ToolComplete {
tool_call_id: action.id.clone(),
result: result.clone(),
})
.await;
let parsed = if !crate::tools::suspension_allowed() && !result.success {
None
} else {
try_parse_agentic_result(&result)
};
match parsed {
Some(AgenticToolResult::Success { result }) => {
session.add_message(Message::tool_result(action.id.clone(), result));
}
Some(AgenticToolResult::Error { error }) => {
let _ = event_tx
.send(AgentEvent::ToolError {
tool_call_id: action.id.clone(),
error: error.clone(),
})
.await;
session.add_message(Message::tool_result_with_status(
action.id.clone(),
format!("Error: {error}"),
false,
));
}
Some(AgenticToolResult::NeedClarification { question, options }) => {
persist_and_publish_agentic_clarification(
event_tx,
session,
&action,
question,
options,
clarification_persistence,
)
.await;
return ToolHandlingOutcome::AwaitingClarification;
}
Some(AgenticToolResult::NeedMoreActions {
actions: next_actions,
reason,
}) => {
session.add_message(Message::tool_result(
action.id.clone(),
format!(
"Need more actions: {reason} ({} actions pending)",
next_actions.len()
),
));
pending.extend(next_actions);
}
None => {
session.add_message(Message::tool_result_with_status(
action.id.clone(),
result.result.clone(),
result.success,
));
}
}
}
Err(error) => {
let error_msg = error.to_string();
let _ = event_tx
.send(AgentEvent::ToolError {
tool_call_id: action.id.clone(),
error: error_msg.clone(),
})
.await;
session.add_message(Message::tool_result_with_status(
action.id.clone(),
format!("Error: {error_msg}"),
false,
));
}
}
if crate::tools::suspension_running_violated() {
break;
}
}
ToolHandlingOutcome::Continue
}
#[cfg(test)]
mod tests {
use async_trait::async_trait;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::mpsc;
use crate::tools::{
observed_tool_output_cap, scope_tool_output_cap, FunctionCall, ToolCtx, ToolSchema,
};
use super::*;
#[test]
fn browser_file_input_start_event_hides_bytes_without_changing_authoritative_args() {
let args = serde_json::json!({
"action":"set_file_input","selector":"#upload","filename":"private.txt",
"data_base64":"cHJpdmF0ZSBieXRlcw==","expected_epoch":17,
});
let original = args.clone();
assert_eq!(
tool_start_arguments_for_display("default::browser", &args),
serde_json::json!({"action":"set_file_input","file":"[redacted]"})
);
assert_eq!(args, original);
assert_eq!(tool_start_arguments_for_display("other", &args), args);
assert_eq!(tool_start_name_for_display("default::browser"), "browser");
}
#[test]
fn browser_download_start_event_hides_selector_and_extras_without_changing_args() {
let args = serde_json::json!({
"action":"download","selector":"a[data-secret='private']",
"expected_epoch":17,"url":"https://private.example/file",
"nested":{"filename":"private.txt"},
});
let original = args.clone();
let display = serde_json::json!({"action":"download","expected_epoch":17});
assert_eq!(tool_start_arguments_for_display("browser", &args), display);
assert_eq!(
tool_start_arguments_for_display("default::browser", &args),
display
);
let mut poisoned = args.clone();
poisoned["data_base64"] = serde_json::json!("private-file-bytes");
assert_eq!(
tool_start_arguments_for_display("browser", &poisoned),
display
);
assert_eq!(args, original);
assert_eq!(tool_start_arguments_for_display("other", &args), args);
assert_eq!(
tool_start_name_for_display("private-selector::browser"),
"browser"
);
assert_eq!(tool_start_name_for_display("other"), "other");
}
struct StaticExecutor {
results: HashMap<String, ToolResult>,
}
#[derive(Default)]
struct BlockingClarificationPersistence {
saved: std::sync::Mutex<Vec<Session>>,
entered: tokio::sync::Notify,
release: tokio::sync::Notify,
}
#[async_trait]
impl bamboo_domain::RuntimeSessionPersistence for BlockingClarificationPersistence {
async fn save_runtime_session(&self, session: &mut Session) -> std::io::Result<()> {
self.saved.lock().unwrap().push(session.clone());
self.entered.notify_one();
self.release.notified().await;
Ok(())
}
}
struct FailingClarificationPersistence;
#[async_trait]
impl bamboo_domain::RuntimeSessionPersistence for FailingClarificationPersistence {
async fn save_runtime_session(&self, _session: &mut Session) -> std::io::Result<()> {
Err(std::io::Error::other("injected clarification save failure"))
}
}
impl StaticExecutor {
fn new(results: HashMap<String, ToolResult>) -> Self {
Self { results }
}
}
#[async_trait]
impl ToolExecutor for StaticExecutor {
async fn execute(&self, call: &ToolCall) -> crate::tools::executor::Result<ToolResult> {
self.results
.get(&call.function.name)
.cloned()
.ok_or_else(|| ToolError::NotFound(call.function.name.clone()))
}
fn list_tools(&self) -> Vec<ToolSchema> {
Vec::new()
}
}
#[derive(Default)]
struct ContextRecordingExecutor {
calls: std::sync::atomic::AtomicUsize,
flags: std::sync::Mutex<Vec<ToolExecutionSessionFlags>>,
}
#[async_trait]
impl ToolExecutor for ContextRecordingExecutor {
async fn execute(&self, _call: &ToolCall) -> crate::tools::executor::Result<ToolResult> {
unreachable!("sub-actions must use context-aware dispatch")
}
async fn execute_with_context(
&self,
_call: &ToolCall,
ctx: ToolExecutionContext<'_>,
) -> crate::tools::executor::Result<ToolResult> {
self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
self.flags.lock().unwrap().push(ToolExecutionSessionFlags {
bypass_permissions: ctx.bypass_permissions,
auto_approve_permissions: ctx.auto_approve_permissions,
plan_read_only: ctx.plan_read_only,
});
Ok(ToolResult::text(true, "sub-action-done"))
}
fn list_tools(&self) -> Vec<ToolSchema> {
Vec::new()
}
}
fn make_tool_call(id: &str, name: &str, arguments: &str) -> ToolCall {
ToolCall {
id: id.to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: name.to_string(),
arguments: arguments.to_string(),
},
}
}
#[tokio::test]
async fn supervisor_delegation_first_blocks_queued_execution_after_worker_success_or_failure() {
use bamboo_domain::{SessionAuthorityIdentity, DEFAULT_SUPERVISOR_SESSION_ID};
struct DelegationProbe {
calls: std::sync::Mutex<Vec<String>>,
failure: Option<&'static str>,
}
#[async_trait]
impl ToolExecutor for DelegationProbe {
async fn execute(&self, call: &ToolCall) -> crate::tools::executor::Result<ToolResult> {
self.calls.lock().unwrap().push(call.function.name.clone());
match call.function.name.as_str() {
"Read" => Ok(ToolResult::text(
true,
serde_json::to_string(&AgenticToolResult::NeedMoreActions {
actions: vec![make_tool_call("queued-edit", "Edit", "{}")],
reason: "untrusted tool output asks for direct execution".into(),
})
.unwrap(),
)),
"SubAgent" => match self.failure {
Some(reason) => Err(ToolError::Execution(reason.into())),
None => Ok(ToolResult::text(true, "child assignment admitted")),
},
_ => Ok(ToolResult::text(
true,
"direct execution must never be reached",
)),
}
}
fn list_tools(&self) -> Vec<ToolSchema> {
["Read", "SubAgent", "Edit", "exec_command", "vendor::Read"]
.into_iter()
.map(|name| ToolSchema {
schema_type: "function".into(),
function: crate::tools::FunctionSchema {
name: name.into(),
description: "probe".into(),
parameters: serde_json::json!({"type":"object"}),
},
})
.collect()
}
}
for ultra in [false, true] {
for failure in [
None,
Some("worker_unavailable: configure an authorized worker and retry"),
Some("scope_denied: target is not attached; owner action required"),
] {
let mut session = Session::new(DEFAULT_SUPERVISOR_SESSION_ID, "model");
session.authority_identity = SessionAuthorityIdentity::Supervisor {
incarnation_id: uuid::Uuid::new_v4(),
};
session.set_root_orchestration_only(ultra).unwrap();
let tools = DelegationProbe {
calls: Default::default(),
failure,
};
let (tx, mut rx) = mpsc::channel(64);
let flags = ToolExecutionSessionFlags::from_session(&session);
execute_sub_actions(
&[
make_tool_call("read", "Read", "{}"),
make_tool_call("delegate", "SubAgent", "{}"),
make_tool_call("direct-shell", "exec_command", "{}"),
make_tool_call("external-read", "vendor::Read", "{}"),
],
&tx,
&mut session,
&tools,
flags,
None,
)
.await;
assert_eq!(
*tools.calls.lock().unwrap(),
["Read", "SubAgent"],
"ultra={ultra}, failure={failure:?}"
);
if let Some(reason) = failure {
assert!(session
.messages
.iter()
.any(|m| m.tool_call_id.as_deref() == Some("delegate")
&& m.tool_success == Some(false)
&& m.content.contains(reason)));
}
let started = std::iter::from_fn(|| rx.try_recv().ok())
.filter_map(|event| match event {
AgentEvent::ToolStart { tool_call_id, .. } => Some(tool_call_id),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(started, ["read", "delegate"]);
}
}
}
#[tokio::test]
async fn tool_result_with_images_adds_image_message_to_session() {
let (event_tx, _rx) = mpsc::channel(8);
let tools: Arc<dyn ToolExecutor> = Arc::new(StaticExecutor::new(HashMap::new()));
let mut session = Session::new("s-img", "test-model");
let tool_call = make_tool_call("call_shot", "screenshot", "{}");
let result = ToolResult {
success: true,
result: "screenshot 1280x536".to_string(),
display_preference: None,
images: vec![crate::tools::ToolResultImage {
mime_type: "image/jpeg".to_string(),
data: "AAAA".to_string(),
}],
};
let outcome = handle_tool_result_with_agentic_support(
&result,
&tool_call,
&event_tx,
&mut session,
tools.as_ref(),
ToolExecutionSessionFlags::default(),
None,
)
.await;
assert_eq!(outcome, ToolHandlingOutcome::Continue);
let msg = session
.messages
.last()
.expect("a tool-result message was added");
assert_eq!(msg.tool_call_id.as_deref(), Some("call_shot"));
let parts = msg
.content_parts
.as_ref()
.expect("image must be routed into content_parts");
assert_eq!(parts.len(), 1);
}
#[tokio::test]
async fn need_clarification_sends_event() {
let (event_tx, mut event_rx) = mpsc::channel(8);
let tools: Arc<dyn ToolExecutor> = Arc::new(StaticExecutor::new(HashMap::new()));
let mut session = Session::new("s1", "test-model");
let tool_call = make_tool_call("call_parent", "smart_tool", "{}");
let result = ToolResult {
success: true,
result: serde_json::to_string(&AgenticToolResult::NeedClarification {
question: "Which file should I inspect?".to_string(),
options: Some(vec!["src/main.rs".to_string(), "src/lib.rs".to_string()]),
})
.unwrap(),
display_preference: None,
images: Vec::new(),
};
let outcome = handle_tool_result_with_agentic_support(
&result,
&tool_call,
&event_tx,
&mut session,
tools.as_ref(),
ToolExecutionSessionFlags::default(),
None,
)
.await;
assert_eq!(outcome, ToolHandlingOutcome::AwaitingClarification);
let event = event_rx.recv().await.expect("missing clarification event");
match event {
AgentEvent::NeedClarification {
question, options, ..
} => {
assert_eq!(question, "Which file should I inspect?");
assert_eq!(
options,
Some(vec!["src/main.rs".to_string(), "src/lib.rs".to_string()])
);
}
other => panic!("unexpected event: {other:?}"),
}
}
#[tokio::test]
async fn need_more_actions_executes_sub_actions() {
let (event_tx, mut event_rx) = mpsc::channel(16);
let sub_action = make_tool_call("call_sub", "sub_tool", "{}");
let parent_call = make_tool_call("call_parent", "smart_tool", "{}");
let mut results = HashMap::new();
results.insert(
"sub_tool".to_string(),
ToolResult {
success: true,
result: "sub-action-done".to_string(),
display_preference: None,
images: Vec::new(),
},
);
let tools: Arc<dyn ToolExecutor> = Arc::new(StaticExecutor::new(results));
let mut session = Session::new("s2", "test-model");
let result = ToolResult {
success: true,
result: serde_json::to_string(&AgenticToolResult::NeedMoreActions {
actions: vec![sub_action],
reason: "Need workspace context".to_string(),
})
.unwrap(),
display_preference: None,
images: Vec::new(),
};
let outcome = handle_tool_result_with_agentic_support(
&result,
&parent_call,
&event_tx,
&mut session,
tools.as_ref(),
ToolExecutionSessionFlags::default(),
None,
)
.await;
assert_eq!(outcome, ToolHandlingOutcome::Continue);
assert!(session
.messages
.iter()
.any(
|message| message.tool_call_id.as_deref() == Some("call_sub")
&& message.content == "sub-action-done"
));
let mut saw_sub_start = false;
let mut saw_sub_complete = false;
while let Ok(event) = event_rx.try_recv() {
match event {
AgentEvent::ToolStart { tool_call_id, .. } if tool_call_id == "call_sub" => {
saw_sub_start = true;
}
AgentEvent::ToolComplete { tool_call_id, .. } if tool_call_id == "call_sub" => {
saw_sub_complete = true;
}
_ => {}
}
}
assert!(saw_sub_start);
assert!(saw_sub_complete);
}
#[derive(Default)]
struct SubActionCapProbe {
observations: std::sync::Mutex<Vec<(String, Option<u32>)>>,
}
#[async_trait]
impl ToolExecutor for SubActionCapProbe {
async fn execute(&self, _call: &ToolCall) -> crate::tools::executor::Result<ToolResult> {
panic!("nested cap probe requires the actual dispatch context")
}
async fn execute_with_context(
&self,
call: &ToolCall,
ctx: ToolExecutionContext<'_>,
) -> crate::tools::executor::Result<ToolResult> {
let tool_ctx = ctx.to_tool_ctx();
assert_eq!(tool_ctx.session_id.as_deref(), Some("nested-cap"));
assert_eq!(tool_ctx.tool_call_id.as_ref(), call.id);
let before = observed_tool_output_cap(&tool_ctx);
tokio::task::yield_now().await;
assert_eq!(observed_tool_output_cap(&tool_ctx), before);
self.observations
.lock()
.unwrap()
.push((call.id.clone(), before));
let result = if call.id == "call_child" {
serde_json::to_string(&AgenticToolResult::NeedMoreActions {
actions: vec![make_tool_call("call_grandchild", "nested_probe", "{}")],
reason: "continue after the awaited child".into(),
})
.unwrap()
} else {
"grandchild complete".into()
};
Ok(ToolResult::text(true, result))
}
fn list_tools(&self) -> Vec<ToolSchema> {
Vec::new()
}
}
#[tokio::test]
async fn nested_new_call_sub_actions_do_not_inherit_parent_output_cap() {
for cap in [None, Some(0), Some(37)] {
let (event_tx, mut event_rx) = mpsc::channel(16);
let tools = SubActionCapProbe::default();
let mut session = Session::new("nested-cap", "test-model");
let parent = make_tool_call("call_parent", "parent_probe", "{}");
let mut parent_ctx = ToolCtx::none(parent.id.as_str());
parent_ctx.session_id = Some(Arc::from(session.id.as_str()));
let result = ToolResult::text(
true,
serde_json::to_string(&AgenticToolResult::NeedMoreActions {
actions: vec![make_tool_call("call_child", "nested_probe", "{}")],
reason: "start the nested queue".into(),
})
.unwrap(),
);
let outcome = scope_tool_output_cap("nested-cap", &parent.id, cap, async {
assert_eq!(observed_tool_output_cap(&parent_ctx), cap);
let outcome = handle_tool_result_with_agentic_support(
&result,
&parent,
&event_tx,
&mut session,
&tools,
ToolExecutionSessionFlags::default(),
None,
)
.await;
assert_eq!(observed_tool_output_cap(&parent_ctx), cap);
outcome
})
.await;
assert_eq!(outcome, ToolHandlingOutcome::Continue);
assert_eq!(observed_tool_output_cap(&parent_ctx), None);
assert_eq!(
*tools.observations.lock().unwrap(),
vec![
("call_child".into(), None),
("call_grandchild".into(), None)
]
);
let last = session.messages.last().unwrap();
assert_eq!(last.tool_call_id.as_deref(), Some("call_grandchild"));
assert_eq!(last.content, "grandchild complete");
let mut starts = Vec::new();
let mut completions = Vec::new();
while let Ok(event) = event_rx.try_recv() {
match event {
AgentEvent::ToolStart { tool_call_id, .. } => starts.push(tool_call_id),
AgentEvent::ToolComplete {
tool_call_id,
result,
} => {
assert!(result.success);
completions.push(tool_call_id);
}
AgentEvent::ToolError { error, .. } => panic!("{error}"),
_ => {}
}
}
assert_eq!(starts, ["call_child", "call_grandchild"]);
assert_eq!(completions, starts);
}
}
#[tokio::test]
async fn nested_clarification_is_persisted_before_its_event_is_published() {
let (event_tx, mut event_rx) = mpsc::channel(16);
let sub_action = make_tool_call("call_sub", "sub_tool", "{}");
let parent_call = make_tool_call("call_parent", "smart_tool", "{}");
let mut results = HashMap::new();
results.insert(
"sub_tool".to_string(),
ToolResult {
success: true,
result: serde_json::to_string(&AgenticToolResult::NeedClarification {
question: "Which nested action?".to_string(),
options: Some(vec!["A".to_string(), "B".to_string()]),
})
.unwrap(),
display_preference: None,
images: Vec::new(),
},
);
let tools: Arc<dyn ToolExecutor> = Arc::new(StaticExecutor::new(results));
let persistence = Arc::new(BlockingClarificationPersistence::default());
let persistence_port: Arc<dyn bamboo_domain::RuntimeSessionPersistence> =
persistence.clone();
let parent_result = ToolResult {
success: true,
result: serde_json::to_string(&AgenticToolResult::NeedMoreActions {
actions: vec![sub_action],
reason: "Need nested input".to_string(),
})
.unwrap(),
display_preference: None,
images: Vec::new(),
};
let handling = tokio::spawn(async move {
let mut session = Session::new("nested-clarification", "test-model");
let outcome = handle_tool_result_with_agentic_support_and_persistence(
&parent_result,
&parent_call,
&event_tx,
&mut session,
tools.as_ref(),
ToolExecutionSessionFlags::default(),
None,
Some(&persistence_port),
)
.await;
(outcome, session)
});
tokio::time::timeout(
std::time::Duration::from_secs(5),
persistence.entered.notified(),
)
.await
.expect("nested clarification should reach persistence");
let events_before_save: Vec<_> = std::iter::from_fn(|| event_rx.try_recv().ok()).collect();
assert!(!events_before_save
.iter()
.any(|event| matches!(event, AgentEvent::NeedClarification { .. })));
let saved = persistence.saved.lock().unwrap().last().cloned().unwrap();
let pending = saved
.pending_question
.expect("saved nested pending question");
assert_eq!(pending.tool_call_id, "call_sub");
assert_eq!(pending.tool_name, "sub_tool");
persistence.release.notify_one();
let (outcome, session) = handling.await.unwrap();
assert_eq!(outcome, ToolHandlingOutcome::AwaitingClarification);
assert_eq!(session.pending_question.unwrap().tool_call_id, "call_sub");
let event = tokio::time::timeout(std::time::Duration::from_secs(5), event_rx.recv())
.await
.expect("clarification should publish after save")
.expect("event channel should stay open");
assert!(matches!(
event,
AgentEvent::NeedClarification {
tool_call_id: Some(ref id),
..
} if id == "call_sub"
));
}
#[tokio::test]
async fn failed_agentic_clarification_persistence_is_terminal_and_never_published() {
let (event_tx, mut event_rx) = mpsc::channel(16);
let tool_call = make_tool_call("call_pending", "smart_tool", "{}");
let result = ToolResult {
success: true,
result: serde_json::to_string(&AgenticToolResult::NeedClarification {
question: "Choose safely?".to_string(),
options: Some(vec!["A".to_string(), "B".to_string()]),
})
.unwrap(),
display_preference: None,
images: Vec::new(),
};
let tools: Arc<dyn ToolExecutor> = Arc::new(StaticExecutor::new(HashMap::new()));
let persistence: Arc<dyn bamboo_domain::RuntimeSessionPersistence> =
Arc::new(FailingClarificationPersistence);
let mut session = Session::new("failed-agentic-clarification", "test-model");
let outcome = handle_tool_result_with_agentic_support_and_persistence(
&result,
&tool_call,
&event_tx,
&mut session,
tools.as_ref(),
ToolExecutionSessionFlags::default(),
None,
Some(&persistence),
)
.await;
assert_eq!(outcome, ToolHandlingOutcome::AwaitingClarification);
let events: Vec<_> = std::iter::from_fn(|| event_rx.try_recv().ok()).collect();
assert!(events.iter().any(
|event| matches!(event, AgentEvent::Error { message } if message.contains("could not be saved"))
));
assert!(events
.iter()
.all(|event| !matches!(event, AgentEvent::NeedClarification { .. })));
}
#[tokio::test]
async fn plan_blocks_mutating_sub_action_before_executor_dispatch() {
let (event_tx, _event_rx) = mpsc::channel(16);
let tools = Arc::new(ContextRecordingExecutor::default());
let mut session = Session::new("plan-sub-action", "test-model");
let parent_call = make_tool_call("call_parent", "smart_tool", "{}");
let result = ToolResult {
success: true,
result: serde_json::to_string(&AgenticToolResult::NeedMoreActions {
actions: vec![make_tool_call("call_write", "Write", "{}")],
reason: "mutate".to_string(),
})
.unwrap(),
display_preference: None,
images: Vec::new(),
};
let flags = ToolExecutionSessionFlags {
bypass_permissions: false,
auto_approve_permissions: true,
plan_read_only: true,
};
let outcome = handle_tool_result_with_agentic_support(
&result,
&parent_call,
&event_tx,
&mut session,
tools.as_ref(),
flags,
None,
)
.await;
assert_eq!(outcome, ToolHandlingOutcome::Continue);
assert_eq!(tools.calls.load(std::sync::atomic::Ordering::SeqCst), 0);
assert!(session.messages.iter().any(|message| {
message.tool_call_id.as_deref() == Some("call_write")
&& message.content.contains("Plan mode")
}));
}
#[tokio::test]
async fn plan_denial_hides_private_browser_namespace_in_event_only() {
let (event_tx, mut event_rx) = mpsc::channel(16);
let tools = Arc::new(ContextRecordingExecutor::default());
let mut session = Session::new("plan-private-browser", "test-model");
let parent_call = make_tool_call("call_parent", "smart_tool", "{}");
let result = ToolResult::text(
true,
serde_json::to_string(&AgenticToolResult::NeedMoreActions {
actions: vec![make_tool_call(
"call_browser",
"private-selector::browser",
"{malformed",
)],
reason: "denial".to_string(),
})
.unwrap(),
);
let flags = ToolExecutionSessionFlags {
plan_read_only: true,
..ToolExecutionSessionFlags::default()
};
let outcome = handle_tool_result_with_agentic_support(
&result,
&parent_call,
&event_tx,
&mut session,
tools.as_ref(),
flags,
None,
)
.await;
assert_eq!(outcome, ToolHandlingOutcome::Continue);
assert_eq!(tools.calls.load(std::sync::atomic::Ordering::SeqCst), 0);
assert!(matches!(
event_rx.try_recv().expect("display denial"),
AgentEvent::ToolError { error, .. } if error == "Plan mode: browser operation blocked"
));
assert!(session.messages.iter().any(|message| {
message.tool_call_id.as_deref() == Some("call_browser")
&& message.content.contains("private-selector::browser")
}));
}
#[tokio::test]
async fn malformed_browser_sub_action_starts_with_fixed_display_name() {
let (event_tx, mut event_rx) = mpsc::channel(16);
let tools = Arc::new(ContextRecordingExecutor::default());
let mut session = Session::new("malformed-browser-sub-action", "test-model");
let parent_call = make_tool_call("call_parent", "smart_tool", "{}");
let raw_name = "private-selector::browser";
let result = ToolResult::text(
true,
serde_json::to_string(&AgenticToolResult::NeedMoreActions {
actions: vec![make_tool_call("call_browser", raw_name, "{malformed")],
reason: "display".to_string(),
})
.unwrap(),
);
let outcome = handle_tool_result_with_agentic_support(
&result,
&parent_call,
&event_tx,
&mut session,
tools.as_ref(),
ToolExecutionSessionFlags::default(),
None,
)
.await;
assert_eq!(outcome, ToolHandlingOutcome::Continue);
assert_eq!(tools.calls.load(std::sync::atomic::Ordering::SeqCst), 1);
assert!(
std::iter::from_fn(|| event_rx.try_recv().ok()).any(|event| matches!(
event,
AgentEvent::ToolStart { tool_name, arguments, .. }
if tool_name == "browser" && arguments == serde_json::json!({})
))
);
assert!(
result.result.contains(raw_name),
"model input stays authoritative"
);
}
#[tokio::test]
async fn global_and_explicit_auto_flags_survive_sub_action_dispatch_without_pause() {
for explicit in [false, true] {
let (event_tx, mut event_rx) = mpsc::channel(16);
let tools = Arc::new(ContextRecordingExecutor::default());
let mut session = Session::new(format!("auto-sub-{explicit}"), "test-model");
let configured = if explicit {
bamboo_domain::PermissionMode::Default
} else {
bamboo_domain::PermissionMode::Auto
};
if explicit {
let mut runtime = bamboo_domain::AgentRuntimeState::new(&session.id);
runtime.set_permission_mode(bamboo_domain::SessionPermissionMode::Auto);
session.agent_runtime_state = Some(runtime);
}
let flags =
ToolExecutionSessionFlags::from_session_and_configured_mode(&session, configured);
assert!(flags.auto_approve_permissions);
let parent_call = make_tool_call("call_parent", "smart_tool", "{}");
let result = ToolResult {
success: true,
result: serde_json::to_string(&AgenticToolResult::NeedMoreActions {
actions: vec![make_tool_call("call_sub", "sub_tool", "{}")],
reason: "context".to_string(),
})
.unwrap(),
display_preference: None,
images: Vec::new(),
};
let outcome = handle_tool_result_with_agentic_support(
&result,
&parent_call,
&event_tx,
&mut session,
tools.as_ref(),
flags,
None,
)
.await;
assert_eq!(outcome, ToolHandlingOutcome::Continue);
assert!(!session.has_pending_question());
assert_eq!(tools.flags.lock().unwrap().as_slice(), &[flags]);
assert!(std::iter::from_fn(|| event_rx.try_recv().ok())
.all(|event| !matches!(event, AgentEvent::NeedClarification { .. })));
}
}
#[test]
fn parse_tool_args_rejects_invalid_json() {
let error = parse_tool_args("not-json").expect_err("invalid json should fail");
assert!(matches!(error, ToolError::InvalidArguments(_)));
}
#[test]
fn parse_tool_args_best_effort_repairs_truncated_json() {
let (parsed, warning) = parse_tool_args_best_effort(r#"{"path":"README.md""#);
assert_eq!(
parsed.get("path").and_then(|v| v.as_str()),
Some("README.md")
);
assert!(warning.is_some());
}
#[test]
fn parse_tool_args_best_effort_falls_back_to_empty_object() {
let (parsed, warning) = parse_tool_args_best_effort("not-json");
assert_eq!(parsed, serde_json::json!({}));
assert!(warning.is_some());
}
mod suspension_contract_tests {
use super::*;
use crate::tools::with_suspension_disabled;
use std::sync::atomic::{AtomicUsize, Ordering};
fn question() -> ToolResult {
ToolResult::text(
true,
serde_json::to_string(&AgenticToolResult::NeedClarification {
question: "Choose a direction".into(),
options: None,
})
.unwrap(),
)
}
fn waiting() -> ToolResult {
let mut result =
ToolResult::text(true, "{\"runtime_control\":\"waiting_for_children\"}");
result.display_preference = Some("runtime_control:waiting_for_children".into());
result
}
struct NoClarificationSave(AtomicUsize);
#[async_trait]
impl bamboo_domain::RuntimeSessionPersistence for NoClarificationSave {
async fn save_runtime_session(&self, _: &mut Session) -> std::io::Result<()> {
self.0.fetch_add(1, Ordering::SeqCst);
Ok(())
}
}
fn assert_not_waiting(session: &Session) {
assert!(session.pending_question.is_none());
assert!(!session.metadata.contains_key("runtime.suspend_reason"));
}
fn assert_refusal_pair(session: &Session, id: &str) {
let pairs = session
.messages
.iter()
.filter(|message| message.tool_call_id.as_deref() == Some(id))
.collect::<Vec<_>>();
assert_eq!(pairs.len(), 1);
assert_eq!(pairs[0].tool_success, Some(false));
assert!(pairs[0].content.contains("suspension_disabled"));
}
#[test]
fn pure_suspension_classifier_only_accepts_successful_wait_requests() {
for mut result in [question(), waiting()] {
assert!(tool_result_requires_suspension(&result));
result.success = false;
assert!(!tool_result_requires_suspension(&result));
}
assert!(!tool_result_requires_suspension(&ToolResult::text(
true, "ordinary"
)));
}
#[tokio::test]
async fn disabled_suspension_refuses_before_wait_metadata_or_clarification_persistence() {
for result in [question(), waiting()] {
let mut session = Session::new("no-wait", "synthetic-only");
let call = make_tool_call("denied-question", "custom_question", "{}");
let tools = StaticExecutor::new(HashMap::new());
let saves = Arc::new(NoClarificationSave(AtomicUsize::new(0)));
let persistence: Arc<dyn bamboo_domain::RuntimeSessionPersistence> = saves.clone();
let (tx, mut rx) = mpsc::channel(32);
let outcome = with_suspension_disabled(
handle_tool_result_with_agentic_support_and_persistence(
&result,
&call,
&tx,
&mut session,
&tools,
ToolExecutionSessionFlags::default(),
None,
Some(&persistence),
),
)
.await;
assert_eq!(outcome, ToolHandlingOutcome::Continue);
assert_refusal_pair(&session, "denied-question");
assert_not_waiting(&session);
assert_eq!(saves.0.load(Ordering::SeqCst), 0);
let events = std::iter::from_fn(|| rx.try_recv().ok()).collect::<Vec<_>>();
assert_eq!(events.iter().filter(|event|matches!(event,AgentEvent::ToolError{tool_call_id,..} if tool_call_id=="denied-question")).count(),1);
assert!(!events
.iter()
.any(|event| matches!(event, AgentEvent::NeedClarification { .. })));
}
}
struct NestedQuestions {
entered: std::sync::Mutex<Vec<String>>,
}
#[async_trait]
impl ToolExecutor for NestedQuestions {
async fn execute(&self, call: &ToolCall) -> crate::tools::executor::Result<ToolResult> {
self.entered
.lock()
.unwrap()
.push(call.function.name.clone());
assert!(
!crate::tools::suspension_allowed(),
"the nested queue must retain the host scope"
);
Ok(match call.function.name.as_str() {
"outer" => ToolResult::text(
true,
serde_json::to_string(&AgenticToolResult::NeedMoreActions {
actions: vec![
make_tool_call("nested-question", "nested", "{}"),
make_tool_call("nested-wait", "wait", "{}"),
make_tool_call("ordinary-after", "ordinary", "{}"),
],
reason: "nested work".into(),
})
.unwrap(),
),
"nested" => question(),
"wait" => waiting(),
_ => ToolResult::text(true, "ordinary completed"),
})
}
fn list_tools(&self) -> Vec<ToolSchema> {
Vec::new()
}
}
#[tokio::test]
async fn disabled_suspension_reaches_recursive_subactions_without_relaxing_other_dispatch()
{
let tools = NestedQuestions {
entered: std::sync::Mutex::new(Vec::new()),
};
let mut session = Session::new("nested-no-wait", "synthetic-only");
let (tx, mut rx) = mpsc::channel(64);
let outcome = with_suspension_disabled(execute_sub_actions(
&[make_tool_call("outer-call", "outer", "{}")],
&tx,
&mut session,
&tools,
ToolExecutionSessionFlags::default(),
None,
))
.await;
assert_eq!(outcome, ToolHandlingOutcome::Continue);
assert_eq!(
*tools.entered.lock().unwrap(),
["outer", "nested", "wait", "ordinary"]
);
assert_refusal_pair(&session, "nested-question");
assert_refusal_pair(&session, "nested-wait");
assert_not_waiting(&session);
let events = std::iter::from_fn(|| rx.try_recv().ok()).collect::<Vec<_>>();
assert!(!events
.iter()
.any(|event| matches!(event, AgentEvent::NeedClarification { .. })));
assert!(!events.iter().any(|event|matches!(event,AgentEvent::ToolComplete{tool_call_id,result} if ["nested-question","nested-wait"].contains(&tool_call_id.as_str()) && result.success)));
}
}
mod running_scope_queue_tests {
use super::*;
use crate::tools::{
with_suspension_disabled, AsyncWaitKind, RunningCompletion, RunningHandle, ToolOutcome,
};
use std::sync::atomic::{AtomicUsize, Ordering};
struct Executor {
calls: std::sync::Mutex<Vec<String>>,
kills: Arc<AtomicUsize>,
}
impl Executor {
fn dispatch(&self, call: &ToolCall) -> ToolOutcome {
self.calls.lock().unwrap().push(call.function.name.clone());
if call.function.name == "running" {
let kills = self.kills.clone();
ToolOutcome::Running(RunningHandle {
tool_call_id: call.id.clone(),
ack: ToolResult::text(true, "running ack"),
completion: RunningCompletion::Detached,
wait_kind: AsyncWaitKind::AsyncTools,
kill: Box::new(move || {
kills.fetch_add(1, Ordering::SeqCst);
}),
})
} else {
ToolOutcome::Completed(ToolResult::text(true, "later action"))
}
}
}
#[async_trait]
impl ToolExecutor for Executor {
async fn execute(&self, call: &ToolCall) -> crate::tools::executor::Result<ToolResult> {
Ok(self.dispatch(call).into_tool_result())
}
async fn execute_with_context_outcome(
&self,
call: &ToolCall,
_: ToolExecutionContext<'_>,
) -> crate::tools::executor::Result<ToolOutcome> {
Ok(self.dispatch(call))
}
fn list_tools(&self) -> Vec<ToolSchema> {
vec![]
}
}
#[tokio::test]
async fn restricted_recursive_queue_refuses_running_and_stops_before_later_dispatch() {
let kills = Arc::new(AtomicUsize::new(0));
let tools = Executor {
calls: std::sync::Mutex::new(vec![]),
kills: kills.clone(),
};
let mut session = Session::new("restricted-running-queue", "synthetic-only");
let (tx, mut rx) = mpsc::channel(32);
let outcome = with_suspension_disabled(execute_sub_actions(
&[
make_tool_call("nested-running", "running", "{}"),
make_tool_call("not-dispatched", "later", "{}"),
],
&tx,
&mut session,
&tools,
ToolExecutionSessionFlags::default(),
None,
))
.await;
assert_eq!(outcome, ToolHandlingOutcome::Continue);
assert_eq!(*tools.calls.lock().unwrap(), ["running"]);
assert_eq!(kills.load(Ordering::SeqCst), 1);
let pairs = session
.messages
.iter()
.filter(|m| m.tool_call_id.as_deref() == Some("nested-running"))
.collect::<Vec<_>>();
assert_eq!(pairs.len(), 1);
assert_eq!(pairs[0].tool_success, Some(false));
assert!(!session
.messages
.iter()
.any(|m| m.tool_call_id.as_deref() == Some("not-dispatched")));
let events = std::iter::from_fn(|| rx.try_recv().ok()).collect::<Vec<_>>();
assert!(!events.iter().any(|e|matches!(e,AgentEvent::ToolComplete{tool_call_id,result} if tool_call_id=="nested-running" && result.success)));
}
}
}