feat: default tool calls to early dispatch

This commit is contained in:
2026-09-30 14:28:42 +09:00
parent 3d87c3860a
commit 44d6d740be
8 changed files with 100 additions and 41 deletions
+1 -1
View File
@@ -83,9 +83,9 @@ pub enum ToolRegistryError {
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum ToolCallDispatchMode {
/// Wait for the complete assistant response before starting any tools.
#[default]
AfterResponse,
/// Start calls at trustworthy provider per-call completion boundaries.
#[default]
OnToolCallComplete,
}
+10 -8
View File
@@ -20,8 +20,8 @@ use agen::tool::{
ToolResultDisposition,
};
use agen::{
Engine, EngineError, EngineRunExit, History, Item, RunInterruptionReason, ToolCallDispatchMode,
ToolExecutionPolicy,
Engine, EngineConfig, EngineError, EngineRunExit, History, Item, RunInterruptionReason,
ToolCallDispatchMode, ToolExecutionPolicy,
};
use async_trait::async_trait;
@@ -465,12 +465,13 @@ impl Tool for BarrierTool {
// =============================================================================
#[tokio::test]
async fn default_dispatch_waits_for_response_completion() {
async fn explicit_after_response_dispatch_waits_for_response_completion() {
let (client, tx) = ControlledStreamClient::new(ToolCallCompletionSupport::PerBlock);
let tool = BarrierTool::new("barrier_default");
let probe = tool.clone();
let mut engine = Engine::new(client);
engine.set_max_turns(Some(1));
engine.set_tool_call_dispatch_mode(ToolCallDispatchMode::AfterResponse);
engine.register_tool(tool.definition());
let (observed_tx, mut observed_rx) = tokio::sync::mpsc::unbounded_channel();
@@ -528,7 +529,6 @@ async fn turn_end_policy_incompatibility_visibly_falls_back_before_side_effects(
let tool = BarrierTool::new("turn_end_fallback");
let probe = tool.clone();
let mut engine = Engine::new(client);
engine.set_tool_call_dispatch_mode(ToolCallDispatchMode::OnToolCallComplete);
engine.set_interceptor(TurnEndPause);
engine.register_tool(tool.definition());
let (warning_tx, mut warning_rx) = tokio::sync::mpsc::unbounded_channel();
@@ -762,7 +762,11 @@ async fn blocking_pre_tool_policy_keeps_receiving_stream_and_services_cancel() {
}
#[tokio::test]
async fn per_block_dispatch_starts_siblings_without_waiting_for_stream_or_prior_result() {
async fn default_per_block_dispatch_starts_siblings_without_waiting_for_stream_or_prior_result() {
assert_eq!(
EngineConfig::default().tool_call_dispatch,
ToolCallDispatchMode::OnToolCallComplete
);
let (client, tx) = ControlledStreamClient::new(ToolCallCompletionSupport::PerBlock);
let first = BarrierTool::new("barrier_a");
let second = BarrierTool::new("barrier_b");
@@ -770,7 +774,6 @@ async fn per_block_dispatch_starts_siblings_without_waiting_for_stream_or_prior_
let second_probe = second.clone();
let mut engine = Engine::new(client);
engine.set_max_turns(Some(1));
engine.set_tool_call_dispatch_mode(ToolCallDispatchMode::OnToolCallComplete);
engine.register_tool(first.definition());
engine.register_tool(second.definition());
@@ -1641,13 +1644,12 @@ async fn clean_eof_without_provider_completion_after_early_start_stops_without_c
}
#[tokio::test]
async fn unsupported_provider_visibly_defers_opt_in_dispatch() {
async fn unsupported_provider_visibly_defers_default_dispatch() {
let (client, tx) = ControlledStreamClient::new(ToolCallCompletionSupport::ResponseComplete);
let tool = BarrierTool::new("barrier_fallback");
let probe = tool.clone();
let mut engine = Engine::new(client);
engine.set_max_turns(Some(1));
engine.set_tool_call_dispatch_mode(ToolCallDispatchMode::OnToolCallComplete);
engine.register_tool(tool.definition());
let warnings = Arc::new(Mutex::new(Vec::new()));
let warning_probe = warnings.clone();
@@ -80,33 +80,38 @@ async fn assert_response_complete_scheme_accepts_clean_eof<S: Scheme>(
ToolCallCompletionSupport::ResponseComplete
);
for mode in [
ToolCallDispatchMode::AfterResponse,
ToolCallDispatchMode::OnToolCallComplete,
for (label, mode) in [
("default on_tool_call_complete fallback", None),
(
"explicit after_response",
Some(ToolCallDispatchMode::AfterResponse),
),
] {
let client = SchemeReplayClient::new(scheme.clone(), vec![frames.clone()]);
let probe = client.clone();
let mut engine = Engine::new(client);
engine.set_tool_call_dispatch_mode(mode);
if let Some(mode) = mode {
engine.set_tool_call_dispatch_mode(mode);
}
let mut history = History::new();
let output = engine.run(&mut history, "hello").await;
assert!(
matches!(output.result, EngineRunExit::Finished),
"{mode:?} should accept the response-complete scheme's clean EOF, got {:?}",
"{label} should accept the response-complete scheme's clean EOF, got {:?}",
output.result
);
assert_eq!(
probe.stream_count(),
1,
"{mode:?} must not open a continuation request after the adapter's normal EOF"
"{label} must not open a continuation request after the adapter's normal EOF"
);
}
}
#[tokio::test]
async fn openai_chat_clean_eof_completes_in_default_and_opt_in_fallback_modes() {
async fn openai_chat_clean_eof_completes_in_default_and_explicit_after_response_modes() {
assert_response_complete_scheme_accepts_clean_eof(
OpenAIScheme::new(),
vec![
@@ -124,7 +129,7 @@ async fn openai_chat_clean_eof_completes_in_default_and_opt_in_fallback_modes()
}
#[tokio::test]
async fn gemini_clean_eof_completes_in_default_and_opt_in_fallback_modes() {
async fn gemini_clean_eof_completes_in_default_and_explicit_after_response_modes() {
assert_response_complete_scheme_accepts_clean_eof(
GeminiScheme::new(),
vec![