use crate::{
BuiltinProvider, ContentBlock, Role, Runtime, TokenUsage,
agent::AgentEvent,
error::RuntimeError,
provider::{ContentBlockDelta, ContentBlockStart, ProviderEvent},
runtime::{CancellationToken, EarlyEnd, RunOptions},
};
use super::support::{
ScriptedProvider, StaticTool, StopTrippingTool, StreamScript, model_info, ok_stream,
};
fn usage(input_tokens: u64, output_tokens: u64) -> TokenUsage {
TokenUsage {
input_tokens: Some(input_tokens),
output_tokens: Some(output_tokens),
..Default::default()
}
}
fn tool_use_stream_with_usage(
model: &str,
id: &str,
name: &str,
input_json: &str,
usage: TokenUsage,
) -> StreamScript {
ok_stream(vec![
ProviderEvent::MessageStarted {
id: format!("msg-{id}"),
model: model.to_string(),
role: Role::Assistant,
},
ProviderEvent::ContentBlockStarted {
index: 0,
kind: ContentBlockStart::ToolUse {
id: id.to_string(),
name: name.to_string(),
},
},
ProviderEvent::ContentBlockDelta {
index: 0,
delta: ContentBlockDelta::ToolUseInputJson(input_json.to_string()),
},
ProviderEvent::ContentBlockStopped { index: 0 },
ProviderEvent::MessageDelta {
stop_reason: None,
usage: Some(usage),
},
ProviderEvent::MessageStopped,
])
}
fn text_stream_with_usage(model: &str, text: &str, usage: TokenUsage) -> StreamScript {
ok_stream(vec![
ProviderEvent::MessageStarted {
id: format!("msg-{text}"),
model: model.to_string(),
role: Role::Assistant,
},
ProviderEvent::ContentBlockStarted {
index: 0,
kind: ContentBlockStart::Text,
},
ProviderEvent::ContentBlockDelta {
index: 0,
delta: ContentBlockDelta::Text(text.to_string()),
},
ProviderEvent::ContentBlockStopped { index: 0 },
ProviderEvent::MessageDelta {
stop_reason: None,
usage: Some(usage),
},
ProviderEvent::MessageStopped,
])
}
#[tokio::test]
async fn token_budget_stops_gracefully_after_the_round_that_crosses_it() {
let model = model_info("model", BuiltinProvider::Anthropic);
let provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![
tool_use_stream_with_usage(
&model.id,
"call-1",
"probe_tool",
r#"{"value":"hi"}"#,
usage(60, 40),
),
text_stream_with_usage(&model.id, "must not run", usage(1, 1)),
],
);
let provider_handle = provider.clone();
let runtime = Runtime::empty_builder()
.with_provider_instance(provider)
.with_tool(StaticTool::success("probe_tool", "ok"))
.build()
.expect("build runtime");
let mut agent = runtime.spawn("agent", model).expect("spawn agent");
let result = agent
.run(
vec![ContentBlock::text("go")],
RunOptions {
token_budget: Some(100),
..Default::default()
},
)
.await;
assert!(matches!(result, Err(RuntimeError::EmptyAssistantResponse)));
assert_eq!(
agent.history().len(),
3,
"the round that crossed the budget stays committed, not rolled back"
);
assert_eq!(
provider_handle.recorded_requests().await.len(),
1,
"the budget halted the run before a second model request"
);
}
#[tokio::test]
async fn absent_token_budget_ignores_reported_usage() {
let model = model_info("model", BuiltinProvider::Anthropic);
let provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![
tool_use_stream_with_usage(
&model.id,
"call-1",
"probe_tool",
r#"{"value":"hi"}"#,
usage(10_000, 10_000),
),
text_stream_with_usage(&model.id, "done", usage(10_000, 10_000)),
],
);
let provider_handle = provider.clone();
let runtime = Runtime::empty_builder()
.with_provider_instance(provider)
.with_tool(StaticTool::success("probe_tool", "ok"))
.build()
.expect("build runtime");
let mut agent = runtime.spawn("agent", model).expect("spawn agent");
let message = agent
.run(vec![ContentBlock::text("go")], RunOptions::default())
.await
.expect("run completes normally despite large reported usage");
assert_eq!(message.text(), "done");
assert_eq!(provider_handle.recorded_requests().await.len(), 2);
assert_eq!(agent.history().len(), 4);
}
#[tokio::test]
async fn a_crossed_budget_reports_why_the_turn_ended() {
let model = model_info("model", BuiltinProvider::Anthropic);
let provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![
tool_use_stream_with_usage(
&model.id,
"call-1",
"probe_tool",
r#"{"value":"hi"}"#,
usage(60, 40),
),
text_stream_with_usage(&model.id, "must not run", usage(1, 1)),
],
);
let provider_handle = provider.clone();
let runtime = Runtime::empty_builder()
.with_provider_instance(provider)
.with_tool(StaticTool::success("probe_tool", "ok"))
.build()
.expect("build runtime");
let mut agent = runtime.spawn("agent", model).expect("spawn agent");
let options = RunOptions {
token_budget: Some(100),
..Default::default()
};
let result = agent
.run(vec![ContentBlock::text("go")], options.clone())
.await;
assert_eq!(
options.ended_early(),
Some(EarlyEnd::TokenBudget),
"the run must report the bound that ended it, not leave it to be inferred"
);
assert!(matches!(result, Err(RuntimeError::EmptyAssistantResponse)));
assert_eq!(
agent.history().len(),
3,
"the round that crossed the budget stays committed, not rolled back"
);
assert_eq!(provider_handle.recorded_requests().await.len(), 1);
}
#[tokio::test]
async fn a_requested_stop_is_not_reported_as_a_crossed_budget() {
let model = model_info("model", BuiltinProvider::Anthropic);
let stop = CancellationToken::default();
let provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![
tool_use_stream_with_usage(
&model.id,
"call-1",
"stop_probe",
r#"{"value":"enough"}"#,
usage(10, 10),
),
text_stream_with_usage(&model.id, "must not run", usage(1, 1)),
],
);
let provider_handle = provider.clone();
let runtime = Runtime::empty_builder()
.with_provider_instance(provider)
.with_tool(StopTrippingTool::new("stop_probe", stop.clone()))
.build()
.expect("build runtime");
let mut agent = runtime.spawn("agent", model).expect("spawn agent");
let options = RunOptions {
stop: Some(stop),
token_budget: Some(10_000),
..Default::default()
};
let result = agent
.run(vec![ContentBlock::text("go")], options.clone())
.await;
assert_eq!(options.ended_early(), Some(EarlyEnd::StopRequested));
assert!(matches!(result, Err(RuntimeError::EmptyAssistantResponse)));
assert_eq!(
options.reported_tokens(),
20,
"the bound was never near: a caller recomputing it would have found nothing"
);
assert_eq!(provider_handle.recorded_requests().await.len(), 1);
}
#[tokio::test]
async fn a_stop_and_a_crossed_budget_together_report_the_stop() {
let model = model_info("model", BuiltinProvider::Anthropic);
let stop = CancellationToken::default();
let provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![
tool_use_stream_with_usage(
&model.id,
"call-1",
"stop_probe",
r#"{"value":"enough"}"#,
usage(60, 60),
),
text_stream_with_usage(&model.id, "must not run", usage(1, 1)),
],
);
let runtime = Runtime::empty_builder()
.with_provider_instance(provider)
.with_tool(StopTrippingTool::new("stop_probe", stop.clone()))
.build()
.expect("build runtime");
let mut agent = runtime.spawn("agent", model).expect("spawn agent");
let options = RunOptions {
stop: Some(stop),
token_budget: Some(100),
..Default::default()
};
let _ = agent
.run(vec![ContentBlock::text("go")], options.clone())
.await;
assert!(
options.reported_tokens() >= 100,
"the budget really is crossed, so the precedence is what decides the report"
);
assert_eq!(options.ended_early(), Some(EarlyEnd::StopRequested));
}
#[tokio::test]
async fn a_turn_that_runs_to_completion_reports_nothing() {
let model = model_info("model", BuiltinProvider::Anthropic);
let provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![text_stream_with_usage(&model.id, "done", usage(10, 10))],
);
let runtime = Runtime::empty_builder()
.with_provider_instance(provider)
.build()
.expect("build runtime");
let mut agent = runtime.spawn("agent", model).expect("spawn agent");
let options = RunOptions {
stop: Some(CancellationToken::default()),
token_budget: Some(10_000),
..Default::default()
};
let message = agent
.run(vec![ContentBlock::text("go")], options.clone())
.await
.expect("the run completes under both bounds");
assert_eq!(message.text(), "done");
assert_eq!(options.ended_early(), None);
}
#[tokio::test]
async fn a_turn_that_ends_on_the_budget_and_still_answers_reports_why() {
let model = model_info("model", BuiltinProvider::Anthropic);
let provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![
text_stream_with_usage(&model.id, "answered first", usage(60, 40)),
text_stream_with_usage(&model.id, "must not run", usage(1, 1)),
],
);
let provider_handle = provider.clone();
let runtime = Runtime::empty_builder()
.with_provider_instance(provider)
.build()
.expect("build runtime");
let mut agent = runtime.spawn("agent", model).expect("spawn agent");
let steering = agent.steering_handle();
steering.steer(vec![ContentBlock::text("and then this")]);
let options = RunOptions {
token_budget: Some(100),
..Default::default()
};
let message = agent
.run(vec![ContentBlock::text("go")], options.clone())
.await
.expect("a turn that ends on the budget after a committed message succeeds");
assert_eq!(message.text(), "answered first");
assert_eq!(
options.ended_early(),
Some(EarlyEnd::TokenBudget),
"a successful turn is exactly where an unreported bound is invisible"
);
assert!(
steering.has_pending(),
"the steer no request could carry is kept, not consumed"
);
assert_eq!(provider_handle.recorded_requests().await.len(), 1);
}
#[tokio::test]
async fn child_run_shares_cancellation_with_parent() {
let model = model_info("model", BuiltinProvider::Anthropic);
let provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![text_stream_with_usage(
&model.id,
"should not complete",
usage(1, 1),
)],
);
let runtime = Runtime::empty_builder()
.with_provider_instance(provider)
.build()
.expect("build runtime");
let mut child_agent = runtime.spawn("child", model).expect("spawn child agent");
let cancellation = CancellationToken::default();
let parent_options = RunOptions {
cancellation: Some(cancellation.clone()),
..Default::default()
};
let child_options = parent_options.child();
cancellation.cancel();
let error = child_agent
.run(vec![ContentBlock::text("go")], child_options)
.await
.expect_err("a cancelled parent token must stop the derived child run");
assert!(matches!(error, RuntimeError::Cancelled));
}
#[tokio::test]
async fn child_usage_counts_toward_shared_token_budget() {
let model = model_info("model", BuiltinProvider::Anthropic);
let parent_provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![text_stream_with_usage(
&model.id,
"parent done",
usage(40, 20),
)],
);
let parent_runtime = Runtime::empty_builder()
.with_provider_instance(parent_provider)
.build()
.expect("build runtime");
let mut parent_agent = parent_runtime
.spawn("parent", model.clone())
.expect("spawn parent");
let parent_options = RunOptions {
token_budget: Some(100),
..Default::default()
};
parent_agent
.run(vec![ContentBlock::text("go")], parent_options.clone())
.await
.expect("parent run completes under budget");
assert_eq!(
parent_options.reported_tokens(),
60,
"parent alone stays under the shared bound"
);
let child_options = parent_options.child();
assert_eq!(
child_options.reported_tokens(),
60,
"the derived child starts from the parent's already-reported usage"
);
let child_provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![
tool_use_stream_with_usage(
&model.id,
"call-1",
"probe_tool",
r#"{"value":"hi"}"#,
usage(30, 20),
),
text_stream_with_usage(&model.id, "must not run", usage(1, 1)),
],
);
let child_provider_handle = child_provider.clone();
let child_runtime = Runtime::empty_builder()
.with_provider_instance(child_provider)
.with_tool(StaticTool::success("probe_tool", "ok"))
.build()
.expect("build runtime");
let mut child_agent = child_runtime.spawn("child", model).expect("spawn child");
let result = child_agent
.run(vec![ContentBlock::text("go")], child_options)
.await;
assert!(
matches!(result, Err(RuntimeError::EmptyAssistantResponse)),
"the child stops gracefully once the combined parent+child usage crosses the bound"
);
assert_eq!(
child_provider_handle.recorded_requests().await.len(),
1,
"the shared bound halted the child before its second round"
);
}
fn collect_events(receiver: &mut tokio::sync::broadcast::Receiver<AgentEvent>) -> Vec<AgentEvent> {
let mut events = Vec::new();
while let Ok(event) = receiver.try_recv() {
events.push(event);
}
events
}
fn reported_usage_totals(events: &[AgentEvent]) -> Vec<u64> {
events
.iter()
.filter_map(|event| match event {
AgentEvent::UsageReport {
input_tokens,
output_tokens,
..
} => Some(input_tokens + output_tokens),
_ => None,
})
.collect()
}
#[tokio::test]
async fn delegated_subagent_usage_counts_against_the_parent_token_budget() {
let model = model_info("model", BuiltinProvider::Anthropic);
let provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![
tool_use_stream_with_usage(
&model.id,
"parent-task",
"task",
r#"{"prompt":"delegate"}"#,
usage(40, 20),
),
text_stream_with_usage(&model.id, "child summary", usage(30, 20)),
text_stream_with_usage(&model.id, "parent done", usage(1, 1)),
],
);
let provider_handle = provider.clone();
let runtime = Runtime::builder()
.with_provider_instance(provider)
.build()
.expect("build runtime");
let mut agent = runtime.spawn("agent", model).expect("spawn agent");
let options = RunOptions {
token_budget: Some(100),
..Default::default()
};
let result = agent
.run(vec![ContentBlock::text("delegate that")], options.clone())
.await;
assert_eq!(
options.reported_tokens(),
110,
"the delegated run must report into the parent's accounting handle, not a fresh one"
);
assert!(
matches!(result, Err(RuntimeError::EmptyAssistantResponse)),
"the parent stops gracefully at the boundary after delegated spend crossed the bound, \
reporting the same 'stopped before a final answer' outcome any tripped bound does"
);
assert_eq!(
provider_handle.recorded_requests().await.len(),
2,
"one parent round and one delegated round: the parent never got a second round"
);
}
#[tokio::test]
async fn parent_cancellation_reaches_the_delegated_subagent() {
let model = model_info("model", BuiltinProvider::Anthropic);
let cancellation = CancellationToken::default();
let provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![
tool_use_stream_with_usage(
&model.id,
"parent-task",
"task",
r#"{"prompt":"delegate"}"#,
usage(1, 1),
),
tool_use_stream_with_usage(
&model.id,
"child-tool",
"cancel_probe",
r#"{"value":"trip it"}"#,
usage(1, 1),
),
text_stream_with_usage(&model.id, "child must not continue", usage(1, 1)),
],
);
let provider_handle = provider.clone();
let runtime = Runtime::builder()
.with_provider_instance(provider)
.with_tool(StopTrippingTool::new("cancel_probe", cancellation.clone()))
.build()
.expect("build runtime");
let mut agent = runtime.spawn("agent", model).expect("spawn agent");
let error = agent
.run(
vec![ContentBlock::text("delegate that")],
RunOptions {
cancellation: Some(cancellation),
..Default::default()
},
)
.await
.expect_err("a cancelled run must fail rather than finish");
assert!(matches!(error, RuntimeError::Cancelled));
assert_eq!(
provider_handle.recorded_requests().await.len(),
2,
"the child stopped at its own round boundary; without the shared token it would \
have run a second round before the parent ever saw the cancellation"
);
}
#[tokio::test]
async fn delegated_usage_reports_reach_the_parent_event_stream() {
let model = model_info("model", BuiltinProvider::Anthropic);
let provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![
tool_use_stream_with_usage(
&model.id,
"parent-task",
"task",
r#"{"prompt":"delegate"}"#,
usage(40, 20),
),
text_stream_with_usage(&model.id, "child summary", usage(30, 20)),
text_stream_with_usage(&model.id, "parent done", usage(5, 5)),
],
);
let runtime = Runtime::builder()
.with_provider_instance(provider)
.build()
.expect("build runtime");
let mut agent = runtime.spawn("agent", model).expect("spawn agent");
let mut events = agent.subscribe_events();
let options = RunOptions::default();
let message = agent
.run(vec![ContentBlock::text("delegate that")], options.clone())
.await
.expect("the run completes with no bound set");
assert_eq!(message.text(), "parent done");
let totals = reported_usage_totals(&collect_events(&mut events));
assert_eq!(
totals,
vec![60, 50, 10],
"the parent's stream carries the delegated round's usage between its own two rounds"
);
assert_eq!(
totals.iter().sum::<u64>(),
options.reported_tokens(),
"what an observer sums from the stream matches what the budget is checked against"
);
}
#[tokio::test]
async fn delegating_with_the_budget_already_spent_fails_the_delegation() {
let model = model_info("model", BuiltinProvider::Anthropic);
let provider = ScriptedProvider::new(
BuiltinProvider::Anthropic,
vec![model.clone()],
vec![
tool_use_stream_with_usage(
&model.id,
"parent-task",
"task",
r#"{"prompt":"delegate"}"#,
usage(60, 60),
),
text_stream_with_usage(&model.id, "must not run", usage(1, 1)),
],
);
let provider_handle = provider.clone();
let runtime = Runtime::builder()
.with_provider_instance(provider)
.build()
.expect("build runtime");
let mut agent = runtime.spawn("agent", model).expect("spawn agent");
let result = agent
.run(
vec![ContentBlock::text("delegate that")],
RunOptions {
token_budget: Some(100),
..Default::default()
},
)
.await;
assert!(matches!(result, Err(RuntimeError::EmptyAssistantResponse)));
assert_eq!(
provider_handle.recorded_requests().await.len(),
1,
"the delegated run stopped at its first boundary without a model request"
);
let subagents = agent.watch_snapshot().borrow().subagents.clone();
assert_eq!(subagents.len(), 1);
assert!(
matches!(
&subagents[0].status,
crate::agent::SpawnedAgentStatus::Failed(message)
if message == "run completed without a final assistant message"
),
"the exhausted delegation is recorded as failed, not finished: {:?}",
subagents[0].status
);
}