use super::*;
use crate::agent::recovery::is_dangling_tool_intent;
#[test]
fn no_tool_assistant_complete_precedes_chunk_flush_and_output_persistence() {
let temp = tempfile::TempDir::new().unwrap();
let provider = ScriptedProvider::new(vec![text_done("final")]);
let session = crate::sessions::SessionManager::new(temp.path().join("sessions"))
.create()
.unwrap();
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let mut sink = SessionOrderProbeSink::new(&session);
agent
.run_print_with_tools_streaming_output(
&provider,
"say final",
None,
Some(&session),
temp.path(),
Some(&mut sink),
)
.unwrap();
assert_eq!(sink.events_at_assistant_complete, vec!["user_input"]);
assert_event_types(
&session,
&["user_input", "assistant_chunk", "assistant_output"],
);
}
#[test]
fn assistant_complete_sink_error_still_flushes_and_persists_replay() {
let temp = tempfile::TempDir::new().unwrap();
let provider = ScriptedProvider::new(vec![text_done("persist me")]);
let session = crate::sessions::SessionManager::new(temp.path().join("sessions"))
.create()
.unwrap();
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let mut sink = FailingOutputSink::new(SinkFailurePoint::AssistantComplete);
let error = agent
.run_print_with_tools_streaming_output(
&provider,
"say final",
None,
Some(&session),
temp.path(),
Some(&mut sink),
)
.unwrap_err()
.to_string();
assert_eq!(error, "sink assistant complete broke");
let events = session.read_events().unwrap();
assert_assistant_chunk(&session, "persist me");
let assistant_output = events
.iter()
.find(|event| event.event_type == "assistant_output")
.expect("assistant output persisted after sink failure");
assert_eq!(assistant_output.payload["text"], "persist me");
}
#[test]
fn request_token_projection_cache_projects_only_appended_turn_items() {
let base_items = vec![ProviderConversationItem::Message(ChatMessage::user(
"hello",
))];
let first_turn_item = ProviderConversationItem::Message(ChatMessage::assistant("reading"));
let second_turn_item = ProviderConversationItem::ToolResult(ProviderToolResult {
call_id: "call_1".to_string(),
tool_name: "read".to_string(),
success: true,
output: "file text".to_string(),
});
let mut turn_items = vec![first_turn_item.clone()];
let mut cache = RequestTokenProjectionCache::new("local-ai", "model", &base_items);
let initial = cache.initial_projection();
let first = cache.project_turn_items(&turn_items);
let first_again = cache.project_turn_items(&turn_items);
turn_items.push(second_turn_item.clone());
let second = cache.project_turn_items(&turn_items);
let full_first = ProviderRequest::from_conversation_without_tools(
"model",
base_items
.iter()
.cloned()
.chain(std::iter::once(first_turn_item))
.collect(),
);
let full_second = ProviderRequest::from_conversation_without_tools(
"model",
base_items.into_iter().chain(turn_items).collect(),
);
assert_eq!(initial.tokens, 6);
assert_eq!(
first,
project_provider_request_input_tokens("local-ai", &full_first)
);
assert_eq!(first_again, first);
assert_eq!(
second,
project_provider_request_input_tokens("local-ai", &full_second)
);
}
#[test]
fn request_token_projection_cache_includes_static_and_dynamic_tool_schemas_once() {
let base_items = vec![ProviderConversationItem::Message(ChatMessage::user(
"hello",
))];
let dynamic_tools = vec![json!({
"type": "function",
"name": "dynamic_probe",
"description": "dynamic schema projection probe",
"parameters": {"type": "object", "properties": {"value": {"type": "string"}}}
})];
let initial_request = ProviderRequest::from_conversation("model", base_items.clone())
.with_dynamic_tool_definitions(dynamic_tools.clone());
let mut cache = RequestTokenProjectionCache::new_for_request("local-ai", &initial_request);
let turn_items = vec![ProviderConversationItem::Message(ChatMessage::assistant(
"done",
))];
let full_request = ProviderRequest::from_conversation(
"model",
base_items.into_iter().chain(turn_items.clone()).collect(),
)
.with_dynamic_tool_definitions(dynamic_tools);
let tool_definitions = initial_request.tool_definitions_json();
let tool_names = tool_definitions
.as_array()
.expect("tool definitions")
.iter()
.filter_map(|definition| definition["name"].as_str())
.collect::<Vec<_>>();
assert!(tool_names.contains(&"read"));
assert!(tool_names.contains(&"subagents"));
assert!(tool_names.contains(&"dynamic_probe"));
assert_eq!(
cache.initial_projection(),
project_provider_request_input_tokens("local-ai", &initial_request)
);
assert_eq!(
cache.project_turn_items(&turn_items),
project_provider_request_input_tokens("local-ai", &full_request)
);
}
#[test]
fn deterministic_projection_processes_each_committed_item_once() {
let base_items = vec![ProviderConversationItem::Message(ChatMessage::user("base"))];
let mut cache = RequestTokenProjectionCache::new("local-ai", "model", &base_items);
let mut turn_items = Vec::new();
for index in 0..1_000 {
turn_items.push(ProviderConversationItem::ToolResult(ProviderToolResult {
call_id: format!("call_{index}"),
tool_name: "read".to_string(),
success: true,
output: "x".to_string(),
}));
let incremental = cache.project_deterministic_turn_items(&turn_items, &[]);
if matches!(index, 99 | 499 | 999) {
let full = ProviderRequest::from_conversation_without_tools(
"model",
base_items
.iter()
.cloned()
.chain(turn_items.iter().cloned())
.collect(),
);
assert_eq!(
incremental,
project_provider_request_input_tokens("local-ai", &full)
);
}
}
assert_eq!(
cache.deterministic_projection_metrics(),
DeterministicProjectionMetrics {
delta_projections: 1_000,
projected_items: 1_000,
}
);
}
#[test]
fn temporary_steering_projection_is_non_mutating_and_not_double_counted() {
let base_items = vec![ProviderConversationItem::Message(ChatMessage::user("base"))];
let committed = ProviderConversationItem::Message(ChatMessage::assistant("working"));
let steering = ProviderConversationItem::Message(ChatMessage::user("steer now"));
let mut cache = RequestTokenProjectionCache::new("local-ai", "model", &base_items);
let mut turn_items = vec![committed];
let with_temporary =
cache.project_deterministic_turn_items(&turn_items, std::slice::from_ref(&steering));
let without_temporary = cache.project_deterministic_turn_items(&turn_items, &[]);
turn_items.push(steering);
let after_injection = cache.project_deterministic_turn_items(&turn_items, &[]);
assert!(with_temporary.tokens > without_temporary.tokens);
assert_eq!(after_injection, with_temporary);
assert_eq!(
cache.deterministic_projection_metrics(),
DeterministicProjectionMetrics {
delta_projections: 2,
projected_items: 2,
}
);
}
#[test]
fn deterministic_projection_stays_separate_from_anthropic_calibration() {
let base_items = vec![ProviderConversationItem::Message(ChatMessage::user("base"))];
let mut turn_items = vec![ProviderConversationItem::Message(ChatMessage::assistant(
"first",
))];
let mut cache = RequestTokenProjectionCache::new(
crate::providers::ANTHROPIC_PROVIDER,
"claude-test",
&base_items,
);
cache.project_turn_items(&turn_items);
cache.calibrate_anthropic_input_tokens(turn_items.len(), 500);
turn_items.push(ProviderConversationItem::Message(ChatMessage::assistant(
"second",
)));
let calibrated = cache.project_turn_items(&turn_items);
let deterministic = cache.project_deterministic_turn_items(&turn_items, &[]);
let full = ProviderRequest::from_conversation_without_tools(
"claude-test",
base_items.into_iter().chain(turn_items).collect(),
);
assert_eq!(calibrated.source, ContextUsageSource::LastProviderUsage);
assert!(calibrated.tokens >= 500);
assert_eq!(
deterministic,
project_provider_request_input_tokens(crate::providers::ANTHROPIC_PROVIDER, &full)
);
}
#[test]
fn deterministic_projection_resets_after_shrink_or_explicit_replacement() {
let base_items = vec![ProviderConversationItem::Message(ChatMessage::user("base"))];
let first = ProviderConversationItem::Message(ChatMessage::assistant("first"));
let second = ProviderConversationItem::Message(ChatMessage::assistant("second"));
let replacement = ProviderConversationItem::Message(ChatMessage::assistant("replacement"));
let mut cache = RequestTokenProjectionCache::new("local-ai", "model", &base_items);
cache.project_deterministic_turn_items(&[first.clone(), second], &[]);
let after_shrink = cache.project_deterministic_turn_items(std::slice::from_ref(&first), &[]);
let shrink_request = ProviderRequest::from_conversation_without_tools(
"model",
base_items
.iter()
.cloned()
.chain(std::iter::once(first))
.collect(),
);
assert_eq!(
after_shrink,
project_provider_request_input_tokens("local-ai", &shrink_request)
);
cache.invalidate_deterministic_turn_items();
let after_replacement =
cache.project_deterministic_turn_items(std::slice::from_ref(&replacement), &[]);
let replacement_request = ProviderRequest::from_conversation_without_tools(
"model",
base_items
.into_iter()
.chain(std::iter::once(replacement))
.collect(),
);
assert_eq!(
after_replacement,
project_provider_request_input_tokens("local-ai", &replacement_request)
);
}
#[test]
fn dangling_tool_detector_uses_only_advertised_request_names() {
let temp = tempfile::TempDir::new().unwrap();
let tools = ToolRuntime::new(temp.path()).unwrap();
tools.set_tool_enabled("read", false).unwrap();
let advertised = advertised_tool_names_for_request(Some(&tools));
assert!(!advertised.contains("read"));
assert!(!is_dangling_tool_intent(
"I'll invoke read now.",
&advertised
));
assert!(is_dangling_tool_intent(
"I'll invoke tool X now.",
&advertised
));
tools.set_tool_enabled("subagents", false).unwrap();
let advertised = advertised_tool_names_for_request(Some(&tools));
assert!(!advertised.contains("subagents"));
assert!(!is_dangling_tool_intent(
"I'll invoke subagents now.",
&advertised
));
tools.set_tool_enabled("subagents", true).unwrap();
let depth_ineligible_tools = tools
.clone_for_cwd_with_subagent_depth(temp.path(), tools.subagents_max_depth())
.unwrap();
let advertised = advertised_tool_names_for_request(Some(&depth_ineligible_tools));
assert!(!advertised.contains("subagents"));
assert!(!is_dangling_tool_intent(
"I'll invoke subagents now.",
&advertised
));
let defaults = advertised_tool_names_for_request(None);
for alias in ["shell", "ffgrep", "fffind"] {
assert!(
!defaults.contains(alias),
"dispatch alias advertised: {alias}"
);
assert!(!is_dangling_tool_intent(
&format!("I'll invoke {alias} now."),
&defaults
));
}
}
#[test]
fn advertised_names_and_settings_match_owned_and_shared_requests() {
fn request_tool_names(request: &ProviderRequest) -> HashSet<String> {
request
.tool_definitions_json()
.as_array()
.expect("tool definitions")
.iter()
.filter_map(|definition| definition.get("name").and_then(serde_json::Value::as_str))
.map(str::to_string)
.collect()
}
let temp = tempfile::TempDir::new().unwrap();
let mut agent = AgentSession::new("model", &[], &SkillDiscovery::default());
agent.thinking_level = crate::thinking::ThinkingLevel::High;
agent.text_verbosity = Some(crate::config::TextVerbosity::High);
agent.send_default_reasoning_summary = false;
let conversation = vec![ProviderConversationItem::Message(ChatMessage::user(
"prompt",
))];
let assert_schema_matches_configuration = |tools: Option<&ToolRuntime>| {
let configuration = tool_request_configuration(tools);
let timeout = Some(Duration::from_secs(17));
let cache_key = Some("request-parity");
let owned =
agent.request_from_conversation(conversation.clone(), timeout, cache_key, tools);
let shared = agent.request_from_shared_conversation(
Arc::from(conversation.clone()),
&[],
timeout,
cache_key,
&configuration,
);
for request in [&owned, &shared] {
assert_eq!(request.thinking_level, agent.thinking_level);
assert_eq!(request.text_verbosity(), agent.text_verbosity);
assert!(!request.send_default_reasoning_summary());
assert_eq!(request.semantic_progress_timeout(), timeout);
assert_eq!(request.prompt_cache_key(), cache_key);
}
assert_eq!(
owned.tool_definitions_json(),
shared.tool_definitions_json()
);
assert_eq!(
request_tool_names(&owned),
configuration.advertised_tool_names
);
assert_eq!(
request_tool_names(&shared),
configuration.advertised_tool_names
);
for alias in ["shell", "ffgrep", "fffind"] {
assert!(!configuration.advertised_tool_names.contains(alias));
}
(owned, shared, configuration)
};
let tools = ToolRuntime::new(temp.path()).unwrap();
let (_, _, configuration) = assert_schema_matches_configuration(Some(&tools));
assert!(configuration.advertised_tool_names.contains("subagents"));
tools.set_tool_enabled("read", false).unwrap();
let (_, _, configuration) = assert_schema_matches_configuration(Some(&tools));
assert!(!configuration.advertised_tool_names.contains("read"));
tools.set_tool_enabled("subagents", false).unwrap();
let (_, _, configuration) = assert_schema_matches_configuration(Some(&tools));
assert!(!configuration.advertised_tool_names.contains("subagents"));
tools.set_tool_enabled("subagents", true).unwrap();
let depth_ineligible_tools = tools
.clone_for_cwd_with_subagent_depth(temp.path(), tools.subagents_max_depth())
.unwrap();
let (_, _, configuration) = assert_schema_matches_configuration(Some(&depth_ineligible_tools));
assert!(!configuration.advertised_tool_names.contains("subagents"));
}
#[test]
fn request_token_projection_cache_uses_openai_codex_tokenizer_source() {
let base_items = vec![ProviderConversationItem::Message(ChatMessage::user(
"お誕生日おめでとう",
))];
let turn_items = vec![ProviderConversationItem::Message(ChatMessage::assistant(
"<|endoftext|>",
))];
let mut cache = RequestTokenProjectionCache::new(
crate::providers::OPENAI_CODEX_PROVIDER,
"gpt-4o",
&base_items,
);
let initial = cache.initial_projection();
let projected = cache.project_turn_items(&turn_items);
assert_eq!(initial.source, ContextUsageSource::TokenizerEstimate);
assert_eq!(projected.source, ContextUsageSource::TokenizerEstimate);
assert!(projected.tokens > initial.tokens);
}
#[test]
fn provider_reasoning_tokens_reach_output_usage_event() {
let provider = ScriptedProvider::new(vec![vec![
ProviderEvent::Usage(Usage {
input: 42,
output: 1,
cache_read: 0,
cache_write: 0,
total: 43,
reasoning_tokens: Some(9),
}),
done(),
]]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default()).with_context_budget(
ContextBudget {
enabled: true,
max_tokens: 16_000,
reserve_tokens: 0,
keep_recent_tokens: 0,
..ContextBudget::default()
},
);
let mut sink = CapturingOutputSink::default();
let output = agent
.run_print_with_tools_streaming_output(
&provider,
"say hi",
None,
None,
Path::new("."),
Some(&mut sink),
)
.unwrap();
assert_eq!(output.usage.unwrap().reasoning_tokens, Some(9));
assert!(sink.outputs.iter().any(|event| matches!(
event,
OutputEvent::ContextUsage {
current_tokens: 42,
max_tokens: 16_000,
reasoning_tokens: Some(9),
..
}
)));
}
#[test]
fn text_only_run_does_not_continue() {
let provider = ScriptedProvider::new(vec![text_done("hello")]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let output = agent
.run_print_with_tools(&provider, "say hi", None, None, Path::new("."))
.unwrap();
assert_eq!(output.text, "hello");
assert_eq!(provider.requests().len(), 1);
}
#[test]
fn text_only_run_streams_deltas_and_preserves_output() {
let provider = ScriptedProvider::new(vec![vec![text("hel"), text("lo"), done()]]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let mut streamed = Vec::new();
let mut sink = |delta: &str| {
streamed.push(delta.to_string());
Ok(())
};
let output = agent
.run_print_with_tools_streaming(
&provider,
"say hi",
None,
None,
Path::new("."),
Some(&mut sink),
)
.unwrap();
assert_eq!(streamed, vec!["hel", "lo"]);
assert_eq!(output.text, "hello");
assert_eq!(provider.requests().len(), 1);
}
#[test]
fn streaming_sink_error_returns_after_provider_callback_without_final_output() {
let temp = tempfile::TempDir::new().unwrap();
std::fs::write(temp.path().join("file.txt"), "hello").unwrap();
let provider = ScriptedProvider::new(vec![vec![
text("ok"),
text("fail"),
read_call("call_after_error"),
text("after"),
done(),
]]);
let manager = crate::sessions::SessionManager::new(temp.path().join("sessions"));
let session = manager.create().unwrap();
let tools = ToolRuntime::new(temp.path()).unwrap();
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let mut streamed = Vec::new();
let mut sink = |delta: &str| {
streamed.push(delta.to_string());
if delta == "fail" {
anyhow::bail!("sink broke");
}
Ok(())
};
let error = agent
.run_print_with_tools_streaming(
&provider,
"say hi",
Some(&tools),
Some(&session),
temp.path(),
Some(&mut sink),
)
.unwrap_err()
.to_string();
assert_eq!(error, "sink broke");
assert_eq!(streamed, vec!["ok", "fail"]);
let events = session.read_events().unwrap();
assert_eq!(
events
.iter()
.map(|event| event.event_type.as_str())
.collect::<Vec<_>>(),
vec!["user_input", "assistant_chunk", "turn_status"]
);
assert_eq!(assistant_chunk_text(&events), "ok");
let turn_status = events
.iter()
.find(|event| event.event_type == "turn_status")
.unwrap();
assert_eq!(turn_status.payload["status"], "failed");
assert_eq!(turn_status.payload["assistant_text"], "ok");
assert_eq!(turn_status.payload["error_summary"], "sink broke");
}
#[test]
fn checkpoint_context_failure_records_one_failed_terminal_status_after_user_input() {
let temp = tempfile::TempDir::new().unwrap();
let manager = crate::sessions::SessionManager::new(temp.path().join("sessions"));
let session = manager.create().unwrap();
let tools = ToolRuntime::new_with_full_settings_and_mcp(
temp.path(),
crate::config::McPaths::from_root(temp.path().join("mc")),
crate::config::Settings::default(),
None,
)
.unwrap();
tools.poison_checkpoint_context_lock();
let provider = ScriptedProvider::new(vec![text_done("must not run")]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let error = agent
.run_print_with_tools_streaming_output(
&provider,
"checkpoint failure",
Some(&tools),
Some(&session),
temp.path(),
None,
)
.unwrap_err()
.to_string();
assert_eq!(error, "checkpoint context lock poisoned");
assert!(provider.requests().is_empty());
let events = session.read_events().unwrap();
assert_eq!(
events
.iter()
.map(|event| event.event_type.as_str())
.collect::<Vec<_>>(),
vec!["user_input", "turn_status"]
);
assert_eq!(events[1].payload["status"], "failed");
}
#[test]
fn streaming_preserves_usage_accounting() {
let provider = ScriptedProvider::new(vec![vec![
text("hello"),
ProviderEvent::Usage(Usage {
input: 1,
output: 2,
cache_read: 3,
cache_write: 4,
total: 10,
reasoning_tokens: None,
}),
done(),
]]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let mut streamed = String::new();
let mut sink = |delta: &str| {
streamed.push_str(delta);
Ok(())
};
let output = agent
.run_print_with_tools_streaming(
&provider,
"say hi",
None,
None,
Path::new("."),
Some(&mut sink),
)
.unwrap();
assert_eq!(streamed, "hello");
assert_eq!(output.usage.unwrap().total, 10);
assert_eq!(output.total_tokens, Some(10));
}
#[test]
fn streaming_output_only_usage_waits_for_known_input_before_publication() {
for input in [None, Some(0), Some(12)] {
let mut events = vec![ProviderEvent::UsageObserved(
crate::providers::UsageObservation {
usage: Usage {
output: 7,
..Usage::default()
},
presence: crate::providers::UsagePresence {
output: true,
..Default::default()
},
},
)];
if let Some(input) = input {
events.push(ProviderEvent::UsageObserved(
crate::providers::UsageObservation {
usage: Usage {
input,
..Usage::default()
},
presence: crate::providers::UsagePresence {
input: true,
..Default::default()
},
},
));
}
events.push(done());
let provider = ScriptedProvider::new(vec![events]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let mut sink = CapturingOutputSink::default();
let output = agent
.run_print_with_tools_streaming_output(
&provider,
"count",
None,
None,
Path::new("."),
Some(&mut sink),
)
.unwrap();
assert_eq!(output.usage.unwrap().output, 7);
let provider_context: Vec<_> = sink
.outputs
.iter()
.filter_map(|event| match event {
OutputEvent::ContextUsage {
current_tokens,
source:
crate::output::ContextUsageSource::ProviderPartial
| crate::output::ContextUsageSource::ProviderExact,
..
} => Some(*current_tokens),
_ => None,
})
.collect();
let snapshots: Vec<_> = sink
.outputs
.iter()
.filter_map(|event| match event {
OutputEvent::UsageSnapshot {
usage, final_usage, ..
} => Some((usage, *final_usage)),
_ => None,
})
.collect();
if let Some(input) = input {
assert_eq!(provider_context, vec![input as usize, input as usize]);
assert_eq!(snapshots.len(), 2);
assert!(!snapshots[0].1);
assert!(snapshots[1].1);
for (usage, _) in snapshots {
assert_eq!(usage.effective_input, input);
assert_eq!(usage.output, 7);
}
} else {
assert!(provider_context.is_empty());
assert!(snapshots.is_empty());
}
}
}
#[test]
fn streamed_requests_without_complete_usage_never_publish_a_final_snapshot() {
for presence in [None, Some(false), Some(true)] {
let mut events = Vec::new();
if let Some(output_known) = presence {
events.push(ProviderEvent::UsageObserved(
crate::providers::UsageObservation {
usage: Usage {
input: 12,
output: 7,
..Default::default()
},
presence: crate::providers::UsagePresence {
input: true,
output: output_known,
..Default::default()
},
},
));
}
events.push(done());
let provider = ScriptedProvider::new(vec![events]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let mut sink = CapturingOutputSink::default();
agent
.run_print_with_tools_streaming_output(
&provider,
"count",
None,
None,
Path::new("."),
Some(&mut sink),
)
.unwrap();
assert!(
sink.outputs
.iter()
.any(|event| matches!(event, OutputEvent::ContextUsage { .. }))
);
assert_eq!(
sink.outputs.iter().any(|event| matches!(
event,
OutputEvent::UsageSnapshot {
final_usage: true,
..
}
)),
presence == Some(true),
"presence {presence:?}"
);
}
}
#[test]
fn streaming_finalizes_split_anthropic_usage_with_cache_tokens() {
let provider = ScriptedProvider::new(vec![vec![
ProviderEvent::UsageObserved(crate::providers::UsageObservation {
usage: Usage {
input: 10,
cache_read: 20,
..Usage::default()
},
presence: crate::providers::UsagePresence {
input: true,
cache_read: true,
..Default::default()
},
}),
ProviderEvent::UsageObserved(crate::providers::UsageObservation {
usage: Usage {
output: 7,
cache_write: 3,
..Usage::default()
},
presence: crate::providers::UsagePresence {
output: true,
cache_write: true,
..Default::default()
},
}),
done(),
]]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default())
.with_provider_id(crate::providers::ANTHROPIC_PROVIDER);
let output = agent
.run_print_with_tools_streaming(&provider, "count", None, None, Path::new("."), None)
.unwrap();
assert_eq!(
output.usage,
Some(Usage {
input: 10,
output: 7,
cache_read: 20,
cache_write: 3,
total: 40,
reasoning_tokens: None,
})
);
assert_eq!(output.total_tokens, Some(40));
}
#[test]
fn streaming_finalizes_compatible_usage_without_total_tokens() {
let provider = ScriptedProvider::new(vec![vec![
ProviderEvent::UsageObserved(crate::providers::UsageObservation {
usage: Usage {
input: 12,
output: 8,
cache_read: 4,
..Usage::default()
},
presence: crate::providers::UsagePresence {
input: true,
output: true,
cache_read: true,
..Default::default()
},
}),
done(),
]]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let output = agent
.run_print_with_tools_streaming(&provider, "count", None, None, Path::new("."), None)
.unwrap();
assert_eq!(output.usage.as_ref().map(|usage| usage.total), Some(20));
assert_eq!(output.total_tokens, Some(20));
}
#[test]
fn streaming_preserves_authoritative_usage_total() {
let provider = ScriptedProvider::new(vec![vec![
ProviderEvent::UsageObserved(crate::providers::UsageObservation {
usage: Usage {
input: 10,
output: 7,
cache_read: 20,
cache_write: 3,
total: 99,
..Usage::default()
},
presence: crate::providers::UsagePresence {
input: true,
output: true,
cache_read: true,
cache_write: true,
total: true,
..Default::default()
},
}),
done(),
]]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default())
.with_provider_id(crate::providers::ANTHROPIC_PROVIDER);
let output = agent
.run_print_with_tools_streaming(&provider, "count", None, None, Path::new("."), None)
.unwrap();
assert_eq!(output.usage.as_ref().map(|usage| usage.total), Some(99));
assert_eq!(output.total_tokens, Some(99));
}
#[test]
fn streaming_aggregates_exact_usage_total_tokens_only() {
let provider = ScriptedProvider::new(vec![vec![
ProviderEvent::UsagePartial(Usage {
input: 99,
output: 1,
cache_read: 0,
cache_write: 0,
total: 100,
reasoning_tokens: None,
}),
ProviderEvent::Usage(Usage {
input: 4,
output: 6,
cache_read: 0,
cache_write: 0,
total: 0,
reasoning_tokens: None,
}),
ProviderEvent::Usage(Usage {
input: 1,
output: 2,
cache_read: 0,
cache_write: 0,
total: 50,
reasoning_tokens: None,
}),
done(),
]]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let output = agent
.run_print_with_tools_streaming(&provider, "count", None, None, Path::new("."), None)
.unwrap();
assert_eq!(output.total_tokens, Some(50));
}
#[test]
fn agent_preserves_assistant_text_when_batch_contains_visible_content() {
let temp = tempfile::TempDir::new().unwrap();
let provider = ScriptedProvider::new(vec![vec![text("\n\n"), text("visible"), done()]]);
let manager = crate::sessions::SessionManager::new(temp.path().join("sessions"));
let session = manager.create().unwrap();
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let output = agent
.run_print_with_tools_streaming_output(
&provider,
"blank prefix",
None,
Some(&session),
temp.path(),
None,
)
.unwrap();
assert_eq!(output.text, "\n\nvisible");
let events = session.read_events().unwrap();
assert_eq!(assistant_chunk_text(&events), "\n\nvisible");
let assistant_output = events
.iter()
.find(|event| event.event_type == "assistant_output")
.unwrap();
assert_eq!(assistant_output.payload["text"], "\n\nvisible");
}
#[test]
fn provider_error_flushes_pending_assistant_chunk() {
let temp = tempfile::TempDir::new().unwrap();
let provider = FailingProvider {
events: vec![text("partial")],
};
let manager = crate::sessions::SessionManager::new(temp.path().join("sessions"));
let session = manager.create().unwrap();
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let error = agent
.run_print_with_tools(&provider, "say", None, Some(&session), temp.path())
.unwrap_err()
.to_string();
assert_eq!(error, "provider broke");
let events = session.read_events().unwrap();
assert_eq!(assistant_chunk_text(&events), "partial");
assert!(
!events
.iter()
.any(|event| event.event_type == "assistant_output")
);
}
#[test]
fn provider_error_does_not_emit_assistant_complete_for_partial_output() {
let temp = tempfile::TempDir::new().unwrap();
let provider = FailingProvider {
events: vec![text("partial")],
};
let manager = crate::sessions::SessionManager::new(temp.path().join("sessions"));
let session = manager.create().unwrap();
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let mut sink = CapturingOutputSink::default();
let error = agent
.run_print_with_tools_streaming_output(
&provider,
"say",
None,
Some(&session),
temp.path(),
Some(&mut sink),
)
.unwrap_err()
.to_string();
assert_eq!(error, "provider broke");
assert!(
sink.outputs.iter().any(
|event| matches!(event, OutputEvent::AssistantDelta { text } if text == "partial")
)
);
assert!(
!sink
.outputs
.iter()
.any(|event| matches!(event, OutputEvent::AssistantComplete { .. }))
);
let events = session.read_events().unwrap();
assert_eq!(assistant_chunk_text(&events), "partial");
assert!(
!events
.iter()
.any(|event| event.event_type == "assistant_output")
);
}
#[test]
fn sink_error_during_activity_started_stops_before_dispatch() {
let temp = tempfile::TempDir::new().unwrap();
std::fs::write(temp.path().join("file.txt"), "hello").unwrap();
let provider = ScriptedProvider::new(vec![read_done("call_1"), vec![done()]]);
let manager = crate::sessions::SessionManager::new(temp.path().join("sessions"));
let session = manager.create().unwrap();
let tools = ToolRuntime::new(temp.path()).unwrap();
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let mut sink = FailingOutputSink::new(SinkFailurePoint::ActivityStarted);
let error = agent
.run_print_with_tools_streaming_output(
&provider,
"read file",
Some(&tools),
Some(&session),
temp.path(),
Some(&mut sink),
)
.unwrap_err()
.to_string();
assert_eq!(error, "sink activity broke");
assert_eq!(provider.requests().len(), 1);
assert_event_types(&session, &["user_input", "tool_call", "turn_status"]);
}
#[test]
fn required_user_input_append_failure_blocks_provider_run() {
let temp = tempfile::TempDir::new().unwrap();
let sessions_root = temp.path().join("sessions");
let manager = crate::sessions::SessionManager::new(sessions_root.clone());
let session = manager.create().unwrap();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&sessions_root, std::fs::Permissions::from_mode(0o500)).unwrap();
}
#[cfg(not(unix))]
std::fs::create_dir_all(session.path()).unwrap();
let provider = ScriptedProvider::new(vec![vec![done()]]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
let error = agent
.run_print_with_tools(&provider, "persist me", None, Some(&session), temp.path())
.unwrap_err()
.to_string();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&sessions_root, std::fs::Permissions::from_mode(0o700)).unwrap();
}
assert!(error.contains("failed to persist user input before provider run"));
assert_eq!(provider.requests().len(), 0);
}
#[cfg(unix)]
#[test]
fn required_tool_call_append_failure_blocks_dispatch_and_side_effects() {
use std::os::unix::fs::PermissionsExt;
let temp = tempfile::TempDir::new().unwrap();
let sessions_root = temp.path().join("sessions");
let session = crate::sessions::SessionManager::new(sessions_root.clone())
.create()
.unwrap();
std::fs::set_permissions(&sessions_root, std::fs::Permissions::from_mode(0o500)).unwrap();
let tools = ToolRuntime::new(temp.path()).unwrap();
let mut persistence = SessionPersistence::new(Some(&session), temp.path());
let mut turn_state = AgentTurnState::default();
let mut output = AgentRunOutput::default();
let mut capture = CapturingOutputSink::default();
let mut output_sink: Option<&mut dyn AgentOutputSink> = Some(&mut capture);
let cancellation = AgentCancellation::default();
let error = crate::agent::tool_lifecycle::run_tool_lifecycle(ToolLifecycleRun {
tool_calls: vec![ToolCall {
id: "call_durable".to_string(),
name: "write".to_string(),
arguments: json!({"path": "should-not-exist.txt", "content": "blocked"}),
}],
tools: Some(&tools),
hooks: None,
herdr_reporter: None,
output_sink: &mut output_sink,
cancellation: &cancellation,
turn_state: &mut turn_state,
output: &mut output,
session_persistence: &mut persistence,
hook_context: HookContextMetadata::new(InvocationMode::Print),
subdir_instruction_state: None,
})
.unwrap_err()
.to_string();
std::fs::set_permissions(&sessions_root, std::fs::Permissions::from_mode(0o700)).unwrap();
assert!(error.contains("failed to persist tool call before dispatch"));
assert!(!temp.path().join("should-not-exist.txt").exists());
assert!(capture.outputs.is_empty());
assert!(
session
.read_events()
.unwrap()
.iter()
.all(|event| event.event_type != "tool_call")
);
}
#[test]
fn old_plan_before_recent_slice_remains_provider_visible() {
let temp = tempfile::TempDir::new().unwrap();
let manager = crate::sessions::SessionManager::new(temp.path().join("sessions"));
let session = manager.create().unwrap();
let marker = "ANCIENT_PLAN_MARKER";
session
.append(&SessionEvent::new(
"user_input",
session.id().to_string(),
temp.path().to_path_buf(),
json!({"text": marker}),
))
.unwrap();
for index in 0..80 {
session
.append(&SessionEvent::new(
"user_input",
session.id().to_string(),
temp.path().to_path_buf(),
json!({"text": format!("recent filler {index} {}", "x".repeat(1200))}),
))
.unwrap();
}
let provider = ScriptedProvider::new(vec![vec![done()]]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default()).with_context_budget(
crate::context::ContextBudget {
enabled: false,
keep_recent_tokens: 1,
..crate::context::ContextBudget::default()
},
);
agent
.run_print_with_tools(&provider, "current", None, Some(&session), temp.path())
.unwrap();
assert!(
provider.requests()[0]
.conversation_items()
.iter()
.any(|item| matches!(
item,
ProviderConversationItem::Message(message) if message.content == marker
))
);
}
#[test]
fn durable_response_item_order_flushes_assistant_chunk_before_provider_item() {
let temp = tempfile::TempDir::new().unwrap();
std::fs::write(temp.path().join("file.txt"), "file body").unwrap();
let response_item = json!({
"type": "function_call",
"call_id": "call_1",
"name": "read",
"arguments": "{\"path\":\"file.txt\"}",
"status": "completed"
});
let provider = ScriptedProvider::new(vec![
vec![
text("I'll read."),
ProviderEvent::ResponseItem(response_item.clone()),
read_call("call_1"),
done(),
],
vec![done()],
]);
let manager = crate::sessions::SessionManager::new(temp.path().join("sessions"));
let session = manager.create().unwrap();
let tools = ToolRuntime::new(temp.path()).unwrap();
let agent = AgentSession::new("model", &[], &SkillDiscovery::default());
agent
.run_print_with_tools_streaming_output(
&provider,
"read file",
Some(&tools),
Some(&session),
temp.path(),
None,
)
.unwrap();
let events = session.read_events().unwrap();
let event_types = events
.iter()
.map(|event| event.event_type.as_str())
.collect::<Vec<_>>();
assert_eq!(
event_types,
vec![
"user_input",
"assistant_chunk",
"provider_response_item",
"tool_call",
"tool_result",
"assistant_output",
]
);
assert_eq!(events[1].payload["text"], "I'll read.");
assert_eq!(events[2].payload["item"], response_item);
let replay = build_conversation_replay(Some(&session)).unwrap();
let assistant_index = replay
.items
.iter()
.position(|item| {
matches!(
item,
ProviderConversationItem::Message(message)
if message.role == crate::providers::MessageRole::Assistant
&& message.content == "I'll read."
)
})
.unwrap();
let response_index = replay
.items
.iter()
.position(|item| item == &ProviderConversationItem::ResponseItem(response_item.clone()))
.unwrap();
let tool_result_index = replay
.items
.iter()
.position(|item| {
matches!(
item,
ProviderConversationItem::ToolResult(result)
if result.call_id == "call_1" && result.output == "file body"
)
})
.unwrap();
assert!(assistant_index < response_index);
assert!(response_index < tool_result_index);
assert_eq!(
replay
.items
.iter()
.filter(|item| matches!(item, ProviderConversationItem::Message(message) if message.role == crate::providers::MessageRole::Assistant && message.content == "I'll read."))
.count(),
1
);
let requests = provider.requests();
let continuation = requests[1].conversation_items();
let assistant_index = continuation
.iter()
.position(|item| {
matches!(
item,
ProviderConversationItem::Message(message)
if message.role == crate::providers::MessageRole::Assistant
&& message.content == "I'll read."
)
})
.unwrap();
let response_index = continuation
.iter()
.position(|item| item == &ProviderConversationItem::ResponseItem(response_item.clone()))
.unwrap();
let tool_result_index = continuation
.iter()
.position(|item| {
matches!(
item,
ProviderConversationItem::ToolResult(result)
if result.call_id == "call_1" && result.output == "file body"
)
})
.unwrap();
assert!(assistant_index < response_index);
assert!(response_index < tool_result_index);
}
#[test]
fn over_window_full_history_fails_before_provider_dispatch() {
let temp = tempfile::TempDir::new().unwrap();
let manager = crate::sessions::SessionManager::new(temp.path().join("sessions"));
let session = manager.create().unwrap();
session
.append(&SessionEvent::new(
"user_input",
session.id().to_string(),
temp.path().to_path_buf(),
json!({"text":"x".repeat(1200)}),
))
.unwrap();
let provider = ScriptedProvider::new(vec![vec![done()]]);
let agent = AgentSession::new("model", &[], &SkillDiscovery::default()).with_context_budget(
crate::context::ContextBudget {
max_tokens: 220,
reserve_tokens: 20,
keep_recent_tokens: 400,
..crate::context::ContextBudget::default()
},
);
let error = agent
.run_print_with_tools(&provider, "next", None, Some(&session), temp.path())
.unwrap_err()
.to_string();
assert!(error.contains("full session history request is estimated"));
assert!(error.contains("exceeding threshold"));
assert_eq!(provider.requests().len(), 0);
let events = session.read_events().unwrap();
assert!(
!events
.iter()
.any(|event| event.event_type == "context_cache")
);
assert!(
!events
.iter()
.any(|event| event.event_type == "user_input" && event.payload["text"] == "next")
);
}