use super::*;
#[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(
"model",
base_items
.iter()
.cloned()
.chain(std::iter::once(first_turn_item))
.collect(),
);
let full_second = ProviderRequest::from_conversation(
"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_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: 8_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: 8_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");
}
#[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_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(60));
}
#[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")
);
}